Thread EC KV read through the request and recover orphaned cookies

crates/trusted-server-adapter-fastly/src/app.rs

@@ -409,6 +409,13 @@ fn build_ec_request_state(
         match EcContext::read_from_request_with_geo(settings, req, services, geo_info.as_ref()) {
             Ok(mut context) => {
                 context.set_device_signals(device_signals);
+                // Orphan-recovery eligibility is intentionally left false here.
+                // Authorizing it during generic pre-routing would let named
+                // routes and request-filter short circuits (e.g. a DataDome
+                // challenge) reach EC finalization and rotate an identity off a
+                // non-publisher response. It is granted only inside the
+                // publisher fallback, after filters pass and the origin start
+                // succeeds — see `dispatch_fallback`.
                 (context, None)
             }
             Err(report) => (EcContext::default(), Some(report)),
@@ -773,8 +780,8 @@ async fn dispatch_fallback(
         // Generate an EC ID if needed — mirrors the legacy catch-all arm.
         // Only for document navigations by recognised browsers; subresource
         // requests may lack consent signals such as Sec-GPC.
-        if ec.is_real_browser
-            && is_navigation_request(&req)
+        let is_publisher_navigation = ec.is_real_browser && is_navigation_request(&req);
+        if is_publisher_navigation
             && let Err(err) = ec
                 .ec_context
                 .generate_if_needed(&state.settings, ec.kv_graph.as_ref())
@@ -812,6 +819,14 @@ async fn dispatch_fallback(
                         .await
                         {
                             Ok(pub_response) => {
+                                // Origin start succeeded on the sole publisher-
+                                // page path: authorize orphan recovery now, and
+                                // only for real-browser document navigations.
+                                // Restricting it here keeps identity rotation
+                                // within the publisher-navigation boundary —
+                                // named routes, integration proxies, and filter
+                                // short circuits never reach this point.
+                                ec.ec_context.set_recovery_eligible(is_publisher_navigation);
                                 publisher_response_into_streaming_response(
                                     pub_response,
                                     &method,
@@ -2429,4 +2444,80 @@ mod tests {
             "the filter's response-header effect must be threaded out"
         );
     }
+
+    fn recovery_eligible_of(response: &Response) -> bool {
+        response
+            .extensions()
+            .get::<super::EcFinalizeState>()
+            .expect("response should carry EcFinalizeState")
+            .ec_context
+            .recovery_eligible()
+    }
+
+    fn browser_navigation_request(path: &str) -> edgezero_core::http::Request {
+        let uri = format!("https://test-publisher.com{path}");
+        let mut req = request_builder()
+            .method(Method::GET)
+            .uri(uri)
+            .header("sec-fetch-dest", "document")
+            .body(Body::empty())
+            .expect("should build request");
+        req.extensions_mut().insert(DeviceSignals::derive(
+            "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 \
+             (KHTML, like Gecko) Chrome/146.0.0.0 Safari/537.36",
+            Some("t13d1516h2_8daaf6152771_b186095e22b6"),
+            Some("1:65536;2:0;4:6291456;6:262144"),
+        ));
+        req
+    }
+
+    #[test]
+    fn named_route_response_is_not_recovery_eligible() {
+        // Orphan recovery must never be authorized on a named route: it is not a
+        // publisher-page navigation, so a missing KV row must not rotate the
+        // identity there.
+        let router = test_router();
+        let response = route(
+            &router,
+            empty_request(Method::GET, "/.well-known/trusted-server.json"),
+        );
+
+        assert!(
+            !recovery_eligible_of(&response),
+            "named-route responses must not authorize orphan recovery"
+        );
+    }
+
+    #[test]
+    fn filter_short_circuit_response_is_not_recovery_eligible() {
+        // A request-filter short circuit (e.g. a DataDome challenge/block) must
+        // not authorize orphan recovery even for a would-be publisher
+        // navigation: no publisher page was served.
+        let router = router_with_request_filters(vec![Arc::new(ChallengeRequestFilter)]);
+        let response = route(&router, browser_navigation_request("/some-page"));
+
+        assert_eq!(
+            response.status(),
+            StatusCode::FORBIDDEN,
+            "the challenge filter should short-circuit routing"
+        );
+        assert!(
+            !recovery_eligible_of(&response),
+            "a short-circuit filter response must not authorize orphan recovery"
+        );
+    }
+
+    #[test]
+    fn publisher_navigation_origin_start_failure_is_not_recovery_eligible() {
+        // Recovery is authorized only after a successful origin start. With no
+        // live backend the publisher origin fails, so even a real-browser
+        // document navigation must leave recovery unauthorized.
+        let router = test_router();
+        let response = route(&router, browser_navigation_request("/some-page"));
+
+        assert!(
+            !recovery_eligible_of(&response),
+            "an origin-start failure must not authorize orphan recovery"
+        );
+    }
 }

crates/trusted-server-adapter-fastly/src/main.rs

@@ -205,9 +205,9 @@ fn edgezero_main(mut req: FastlyRequest) {
         policy.apply_after_route_finalization(&mut response);
     }
 
-    if let Some(ec_state) = ec_state {
+    if let Some(mut ec_state) = ec_state {
         if let Some(settings) = settings_snapshot.as_deref() {
-            match apply_edgezero_ec_finalize(settings, &ec_state, &mut response) {
+            match apply_edgezero_ec_finalize(settings, &mut ec_state, &mut response) {
                 Ok(partner_registry) => {
                     send_edgezero_response(response, request_filter_effects.as_ref());
                     run_edgezero_pull_sync_after_send(settings, &partner_registry, &ec_state);
@@ -222,7 +222,7 @@ fn edgezero_main(mut req: FastlyRequest) {
         } else {
             match load_settings_from_config_store() {
                 Ok(settings) => {
-                    match apply_edgezero_ec_finalize(&settings, &ec_state, &mut response) {
+                    match apply_edgezero_ec_finalize(&settings, &mut ec_state, &mut response) {
                         Ok(partner_registry) => {
                             send_edgezero_response(response, request_filter_effects.as_ref());
                             run_edgezero_pull_sync_after_send(
@@ -286,7 +286,7 @@ fn apply_entry_point_finalize_headers(
 
 fn apply_edgezero_ec_finalize(
     settings: &Settings,
-    ec_state: &EcFinalizeState,
+    ec_state: &mut EcFinalizeState,
     response: &mut HttpResponse,
 ) -> Result<PartnerRegistry, Report<TrustedServerError>> {
     let partner_registry = PartnerRegistry::from_config(&settings.ec.partners)?;
@@ -297,7 +297,7 @@ fn apply_edgezero_ec_finalize(
     };
     ec_finalize_response(
         settings,
-        &ec_state.ec_context,
+        &mut ec_state.ec_context,
         finalize_kv_graph.as_ref(),
         &partner_registry,
         ec_state.eids_cookie.as_deref(),

crates/trusted-server-adapter-fastly/src/platform.rs

@@ -551,6 +551,10 @@ impl PlatformHttpClient for FastlyPlatformHttpClient {
         true
     }
 
+    fn supports_pending_streaming_responses(&self) -> bool {
+        true
+    }
+
     async fn send(
         &self,
         request: PlatformHttpRequest,
@@ -580,17 +584,17 @@ impl PlatformHttpClient for FastlyPlatformHttpClient {
             return Err(Report::new(PlatformError::HttpClient)
                 .attach("Image Optimizer is not supported with Fastly send_async"));
         }
-        if request.stream_response {
-            return Err(Report::new(PlatformError::HttpClient)
-                .attach("streaming responses are not supported with Fastly send_async"));
-        }
+        let stream_response = request.stream_response;
+        let request_method = request.request.method().clone();
         let bypass_cache = request.bypass_cache;
         let mut fastly_req = edge_request_to_fastly(request.request)?;
         apply_fastly_cache_bypass(&mut fastly_req, bypass_cache);
         let pending = fastly_req
             .send_async(&backend_name)
             .change_context(PlatformError::HttpClient)?;
-        Ok(PlatformPendingRequest::new(pending).with_backend_name(backend_name))
+        Ok(PlatformPendingRequest::new(pending)
+            .with_backend_name(backend_name)
+            .with_response_handling(stream_response, request_method))
     }
 
     async fn select(
@@ -604,6 +608,14 @@ impl PlatformHttpClient for FastlyPlatformHttpClient {
                 .attach("select called with an empty pending_requests list"));
         }
 
+        if pending_requests
+            .iter()
+            .any(PlatformPendingRequest::stream_response)
+        {
+            return Err(Report::new(PlatformError::HttpClient)
+                .attach("stream-marked pending request requires direct wait"));
+        }
+
         let mut fastly_pending: Vec<PendingRequest> = Vec::with_capacity(pending_requests.len());
 
         for platform_req in pending_requests {
@@ -660,6 +672,33 @@ impl PlatformHttpClient for FastlyPlatformHttpClient {
             failed_backend_name,
         })
     }
+
+    async fn wait(
+        &self,
+        pending: PlatformPendingRequest,
+    ) -> Result<PlatformResponse, Report<PlatformError>> {
+        use fastly::http::request::PendingRequest;
+
+        let backend_hint = pending.backend_name().map(str::to_owned);
+        let stream_response = pending.stream_response();
+        let request_is_head = pending.request_method() == Some(&edgezero_core::http::Method::HEAD);
+        let pending = pending.downcast::<PendingRequest>().map_err(|pending| {
+            let backend_name = pending.backend_name().unwrap_or("<unknown>");
+            Report::new(PlatformError::HttpClient).attach(format!(
+                "PlatformPendingRequest inner type is not fastly::PendingRequest for backend '{backend_name}'"
+            ))
+        })?;
+        let response = pending.wait().change_context(PlatformError::HttpClient)?;
+        let backend_name = response
+            .get_backend_name()
+            .map(str::to_owned)
+            .or(backend_hint)
+            .ok_or_else(|| {
+                Report::new(PlatformError::HttpClient)
+                    .attach("wait: response has no backend name; correlation impossible")
+            })?;
+        fastly_response_to_platform(response, backend_name, stream_response, request_is_head)
+    }
 }
 
 // ---------------------------------------------------------------------------
@@ -1081,6 +1120,21 @@ mod tests {
         );
     }
 
+    #[test]
+    fn fastly_platform_http_client_rejects_stream_marked_pending_from_select() {
+        let client = FastlyPlatformHttpClient;
+        let pending = PlatformPendingRequest::new(42_u32)
+            .with_backend_name("origin-a")
+            .with_response_handling(true, edgezero_core::http::Method::GET);
+        let err = futures::executor::block_on(client.select(vec![pending]))
+            .expect_err("should reject stream-marked pending handles from select");
+
+        assert!(
+            format!("{err:?}").contains("stream-marked pending request"),
+            "should explain that streaming pendings require direct wait: {err:?}"
+        );
+    }
+
     #[test]
     fn fastly_platform_http_client_send_returns_error_for_streaming_body() {
         let client = FastlyPlatformHttpClient;
@@ -1132,8 +1186,13 @@ mod tests {
     }
 
     #[test]
-    fn fastly_platform_http_client_send_async_rejects_stream_response() {
+    fn fastly_platform_http_client_supports_pending_streaming_responses() {
         let client = FastlyPlatformHttpClient;
+        assert!(
+            client.supports_pending_streaming_responses(),
+            "should advertise direct pending-response streaming"
+        );
+
         let request = request_builder()
             .method("GET")
             .uri("https://example.com/image.jpg")
@@ -1143,11 +1202,11 @@ mod tests {
             PlatformHttpRequest::new(request, "nonexistent-backend").with_stream_response();
 
         let err = futures::executor::block_on(client.send_async(platform_request))
-            .expect_err("should reject async streaming-response requests");
+            .expect_err("should fail only because the backend is unregistered");
 
         assert!(
-            format!("{err:?}").contains("streaming responses"),
-            "should explain unsupported async streaming-response path: {err:?}"
+            !format!("{err:?}").contains("streaming responses are not supported"),
+            "should accept streaming on the async path before backend dispatch: {err:?}"
         );
     }