From ff782e0f1e9a2a005fcf7b32128d9ca7795dc45d Mon Sep 17 00:00:00 2001 From: prk-Jr Date: Wed, 8 Jul 2026 14:46:47 +0530 Subject: [PATCH 1/6] Quantize auction transport timeouts to stabilize Fastly backend names Fastly dynamic backend names embed the first-byte and between-bytes timeouts so a registration can never be silently reused with a different transport configuration. Deriving those timeouts from the remaining wall-clock auction budget minted a new backend name on nearly every request, defeating cross-request TCP/TLS connection reuse (Fastly pools connections per backend name) and accumulating registrations toward the per-service dynamic backend limit. Compute the effective transport timeout from the configured provider timeout verbatim when the budget allows, floor the budget-bound value to 250ms buckets otherwise, and pass sub-quantum remainders through exactly so publishers with sub-250ms configured budgets keep launching. The quantized value feeds both the backend name and the registered configuration, so they cannot diverge. Rounding down never extends a transport cap past the auction deadline, which the mediator and dispatched-collect paths rely on to bound the hold. Also add a Fastly platform test pinning predict_name == ensure for the same spec, since the orchestrator maps responses back to providers by predicted backend name. Fixes #847 --- .../src/platform.rs | 31 + .../src/auction/orchestrator.rs | 628 ++++++++++++++++-- 2 files changed, 617 insertions(+), 42 deletions(-) diff --git a/crates/trusted-server-adapter-fastly/src/platform.rs b/crates/trusted-server-adapter-fastly/src/platform.rs index 106ced787..c5bb60b9c 100644 --- a/crates/trusted-server-adapter-fastly/src/platform.rs +++ b/crates/trusted-server-adapter-fastly/src/platform.rs @@ -736,6 +736,37 @@ mod tests { ); } + #[test] + fn predict_name_matches_ensured_backend_name() { + // The auction orchestrator maps responses back to providers by the + // predicted backend name, so predict_name and ensure must return the + // identical string for the same spec — a divergence would make + // responses land in the "unknown backend" branch and drop bids + // silently. + let backend = FastlyPlatformBackend; + let spec = PlatformBackendSpec { + scheme: "https".to_string(), + host: "consistency.example.com".to_string(), + port: None, + host_header_override: None, + certificate_check: true, + first_byte_timeout: Duration::from_millis(750), + between_bytes_timeout: Duration::from_millis(750), + }; + + let predicted = backend + .predict_name(&spec) + .expect("should predict backend name"); + let ensured = backend + .ensure(&spec) + .expect("should register backend for valid spec"); + + assert_eq!( + predicted, ensured, + "predicted backend name should match the registered backend name" + ); + } + // --- FastlyPlatformHttpClient ------------------------------------------- #[test] diff --git a/crates/trusted-server-core/src/auction/orchestrator.rs b/crates/trusted-server-core/src/auction/orchestrator.rs index 145059d9e..d884a0220 100644 --- a/crates/trusted-server-core/src/auction/orchestrator.rs +++ b/crates/trusted-server-core/src/auction/orchestrator.rs @@ -157,6 +157,64 @@ fn remaining_budget_ms(start: Instant, timeout_ms: u32) -> u32 { timeout_ms.saturating_sub(elapsed) } +/// Transport-timeout quantum for auction backends. +/// +/// See [`quantize_transport_timeout_ms`] for why provider transport timeouts +/// are rounded to this granularity. +const TRANSPORT_TIMEOUT_QUANTUM_MS: u32 = 250; + +/// Round a transport timeout down to a [`TRANSPORT_TIMEOUT_QUANTUM_MS`] multiple. +/// +/// The Fastly adapter embeds the first-byte and between-bytes timeouts in the +/// dynamic backend name so a registration can never be silently reused with a +/// different transport configuration. Deriving those timeouts from the +/// remaining wall-clock budget minted a new backend name on nearly every +/// request, which defeated cross-request TCP/TLS connection reuse (Fastly +/// pools connections per backend name) and accumulated registrations toward +/// the per-service dynamic backend limit. +/// +/// Quantizing the value — not just the name — keeps the registered backend +/// configuration aligned with its name. Rounding down never extends a +/// transport cap past the auction deadline, which matters on the mediator and +/// dispatched-collect paths where the backend timeouts (not a select-loop +/// deadline check) bound the `` hold. +#[inline] +fn quantize_transport_timeout_ms(timeout_ms: u32) -> u32 { + (timeout_ms / TRANSPORT_TIMEOUT_QUANTUM_MS) * TRANSPORT_TIMEOUT_QUANTUM_MS +} + +/// Compute the transport timeout for a provider launch from the remaining +/// auction budget and the provider's configured timeout. +/// +/// The configured timeout is a per-provider constant, so using it verbatim +/// already yields a stable backend name — including configured values below +/// one quantum, which must not be rounded away or the provider could never +/// launch. Only when the remaining budget is the binding constraint does the +/// wall-clock-derived value enter the name, and that value is quantized via +/// [`quantize_transport_timeout_ms`] so it cannot mint a new backend name on +/// every request. +/// +/// A remaining budget below one quantum is passed through exactly rather +/// than rounded to zero: rounding up would extend the transport cap past the +/// deadline, and rounding down would skip the launch and hard-fail auctions +/// whose configured budget is under one quantum. Name churn in this regime +/// is bounded to sub-quantum values and matches the pre-quantization +/// behavior. The result never exceeds `remaining_ms` and is zero only when +/// `remaining_ms` or `configured_ms` is zero, which callers treat as +/// "budget exhausted — skip the launch". +#[inline] +fn effective_transport_timeout_ms(remaining_ms: u32, configured_ms: u32) -> u32 { + if remaining_ms >= configured_ms { + return configured_ms; + } + let quantized = quantize_transport_timeout_ms(remaining_ms); + if quantized == 0 { + remaining_ms + } else { + quantized + } +} + /// Manages auction execution across multiple providers. pub struct AuctionOrchestrator { config: AuctionConfig, @@ -279,10 +337,14 @@ impl AuctionOrchestrator { // Give the mediator only the remaining time from the auction // deadline, not the full timeout — the bidding phase already - // consumed part of it. + // consumed part of it, and the mediator has no select-loop + // deadline backstop. Quantized for backend-name stability (see + // effective_transport_timeout_ms). let remaining_ms = remaining_budget_ms(mediation_start, context.timeout_ms); + let mediator_timeout = + effective_transport_timeout_ms(remaining_ms, mediator.timeout_ms()); - if remaining_ms == 0 { + if mediator_timeout == 0 { log::warn!("Auction timeout exhausted during bidding phase; skipping mediator"); let winning = self.select_winning_bids(&provider_responses, &floor_prices); return Ok(OrchestrationResult { @@ -297,9 +359,7 @@ impl AuctionOrchestrator { let mediator_context = AuctionContext { settings: context.settings, request: context.request, - // Bound by both the remaining auction budget and the mediator's - // own configured timeout, matching the dispatched collect path. - timeout_ms: remaining_ms.min(mediator.timeout_ms()), + timeout_ms: mediator_timeout, provider_responses: Some(&provider_responses), services: context.services, }; @@ -465,10 +525,11 @@ impl AuctionOrchestrator { // Give each provider only the remaining time from the auction // deadline so that backend transport timeouts do not extend past - // the overall budget. Also respect the provider's own configured - // timeout when it is tighter than the remaining budget. + // the overall budget, quantized for backend-name stability (see + // effective_transport_timeout_ms). let remaining_ms = remaining_budget_ms(auction_start, context.timeout_ms); - let effective_timeout = remaining_ms.min(provider.timeout_ms()); + let effective_timeout = + effective_transport_timeout_ms(remaining_ms, provider.timeout_ms()); if effective_timeout == 0 { log::warn!("Auction timeout exhausted before launching provider request; skipping"); @@ -876,8 +937,11 @@ impl AuctionOrchestrator { continue; } + // Remaining budget quantized for backend-name stability (see + // effective_transport_timeout_ms). let remaining_ms = remaining_budget_ms(auction_start, context.timeout_ms); - let effective_timeout = remaining_ms.min(provider.timeout_ms()); + let effective_timeout = + effective_transport_timeout_ms(remaining_ms, provider.timeout_ms()); if effective_timeout == 0 { log::warn!( @@ -1132,8 +1196,15 @@ impl AuctionOrchestrator { // timeout) at dispatch time, so they cannot run past A_deadline // independently. Giving the mediator an uncapped timeout lets it run // past A_deadline, violating the bounded hold invariant. + // The mediator's only time bound on this path is its + // backend transport timeout, so the effective value must + // never exceed the remaining budget. Quantized for + // backend-name stability (see + // effective_transport_timeout_ms). let remaining = remaining_budget_ms(auction_start, timeout_ms); - if remaining == 0 { + let mediator_timeout = + effective_transport_timeout_ms(remaining, mediator.timeout_ms()); + if mediator_timeout == 0 { log::warn!( "A_deadline exhausted before mediator '{}' — returning {} SSP bids without mediation", mediator.provider_name(), @@ -1148,7 +1219,6 @@ impl AuctionOrchestrator { metadata: HashMap::new(), }; } - let mediator_timeout = remaining.min(mediator.timeout_ms()); let mediator_start = Instant::now(); log::info!( "Running mediator '{}' with {}ms budget (A_deadline remaining: {}ms, configured: {}ms)", @@ -1334,7 +1404,7 @@ mod tests { use crate::test_support::tests::crate_test_settings_str; use error_stack::{Report, ResultExt}; use std::collections::{HashMap, HashSet}; - use std::sync::Arc; + use std::sync::{Arc, Mutex}; use super::AuctionOrchestrator; @@ -1342,9 +1412,49 @@ mod tests { // Minimal test double for AuctionProvider // --------------------------------------------------------------------------- + /// Minimal stub provider. Optionally records every transport timeout it + /// observes — the value passed to `backend_name` and the + /// `context.timeout_ms` handed to `request_bids` — so tests can assert + /// the orchestrator quantizes them. struct StubAuctionProvider { name: &'static str, backend: &'static str, + configured_timeout_ms: u32, + observed_timeouts: Option>>>, + } + + impl StubAuctionProvider { + fn new(name: &'static str, backend: &'static str) -> Self { + Self { + name, + backend, + configured_timeout_ms: 2000, + observed_timeouts: None, + } + } + + fn recording( + name: &'static str, + backend: &'static str, + configured_timeout_ms: u32, + observed_timeouts: Arc>>, + ) -> Self { + Self { + name, + backend, + configured_timeout_ms, + observed_timeouts: Some(observed_timeouts), + } + } + + fn record(&self, timeout_ms: u32) { + if let Some(observed) = &self.observed_timeouts { + observed + .lock() + .expect("should lock observed timeouts") + .push(timeout_ms); + } + } } #[async_trait::async_trait(?Send)] @@ -1358,6 +1468,7 @@ mod tests { _request: &AuctionRequest, context: &AuctionContext<'_>, ) -> Result> { + self.record(context.timeout_ms); let req = PlatformHttpRequest::new( http::Request::builder() .method("POST") @@ -1389,10 +1500,11 @@ mod tests { } fn timeout_ms(&self) -> u32 { - 2000 + self.configured_timeout_ms } - fn backend_name(&self, _services: &RuntimeServices, _timeout_ms: u32) -> Option { + fn backend_name(&self, _services: &RuntimeServices, timeout_ms: u32) -> Option { + self.record(timeout_ms); Some(self.backend.to_string()) } } @@ -1509,10 +1621,10 @@ mod tests { ..Default::default() }; let mut orchestrator = AuctionOrchestrator::new(config); - orchestrator.register_provider(Arc::new(StubAuctionProvider { - name: "bidder", - backend: "bidder-backend", - })); + orchestrator.register_provider(Arc::new(StubAuctionProvider::new( + "bidder", + "bidder-backend", + ))); orchestrator.register_provider(Arc::new(CacheRestoringMediator)); let request = create_test_auction_request(); @@ -1926,6 +2038,438 @@ mod tests { ); } + #[test] + fn quantize_transport_timeout_floors_to_quantum() { + assert_eq!( + super::quantize_transport_timeout_ms(0), + 0, + "should keep zero at zero" + ); + assert_eq!( + super::quantize_transport_timeout_ms(249), + 0, + "should floor a sub-quantum budget to zero" + ); + assert_eq!( + super::quantize_transport_timeout_ms(250), + 250, + "should keep an exact quantum multiple unchanged" + ); + assert_eq!( + super::quantize_transport_timeout_ms(999), + 750, + "should floor to the next-lower quantum multiple" + ); + assert_eq!( + super::quantize_transport_timeout_ms(2000), + 2000, + "should keep a larger exact quantum multiple unchanged" + ); + } + + #[test] + fn effective_transport_timeout_prefers_configured_constant() { + assert_eq!( + super::effective_transport_timeout_ms(2000, 1000), + 1000, + "should use the configured timeout verbatim when the budget allows" + ); + assert_eq!( + super::effective_transport_timeout_ms(2000, 100), + 100, + "should preserve a sub-quantum configured timeout — quantizing it away would permanently disable the provider" + ); + assert_eq!( + super::effective_transport_timeout_ms(999, 2000), + 750, + "should quantize the budget-bound value down to the 750ms bucket" + ); + assert_eq!( + super::effective_transport_timeout_ms(300, 2000), + 250, + "should quantize a tight budget down to one quantum" + ); + assert_eq!( + super::effective_transport_timeout_ms(200, 2000), + 200, + "should pass a sub-quantum budget through exactly instead of rounding to zero" + ); + assert_eq!( + super::effective_transport_timeout_ms(50, 100), + 50, + "should pass through when the budget is below both the quantum and the configured timeout" + ); + assert_eq!( + super::effective_transport_timeout_ms(0, 1000), + 0, + "should return zero for an exhausted budget so the launch is skipped" + ); + assert_eq!( + super::effective_transport_timeout_ms(100, 0), + 0, + "should return zero for a zero configured timeout so the launch is skipped" + ); + } + + #[test] + fn sub_quantum_configured_timeout_still_launches_provider() { + futures::executor::block_on(async { + // A provider whose configured timeout is below one quantum must + // still launch with its exact configured value: the constant is + // name-stable on its own, so only budget-derived values are + // quantized. + let stub = Arc::new(StubHttpClient::new()); + stub.push_response(200, b"{}".to_vec()); + let services = build_services_with_http_client(stub); + // SAFETY: `Box::leak` creates a `'static` reference for test use only. + // The leaked allocation is bounded to the test process lifetime. + let services: &'static RuntimeServices = Box::leak(Box::new(services)); + + let observed = Arc::new(Mutex::new(Vec::new())); + let config = AuctionConfig { + enabled: true, + providers: vec!["bidder".to_string()], + timeout_ms: 2000, + mediator: None, + ..Default::default() + }; + let mut orchestrator = AuctionOrchestrator::new(config); + orchestrator.register_provider(Arc::new(StubAuctionProvider::recording( + "bidder", + "bidder-backend", + 100, + Arc::clone(&observed), + ))); + + let request = create_test_auction_request(); + let settings = create_test_settings(); + let req = http::Request::builder() + .method(http::Method::GET) + .uri("https://example.com/test") + .body(edgezero_core::body::Body::empty()) + .expect("should build request"); + let context = AuctionContext { + settings: &settings, + request: &req, + timeout_ms: 2000, + provider_responses: None, + services, + }; + + orchestrator + .run_auction(&request, &context) + .await + .expect("should complete auction"); + + let observed = observed.lock().expect("should lock observed timeouts"); + assert!( + !observed.is_empty(), + "should launch the sub-quantum-configured provider" + ); + for timeout in observed.iter() { + assert_eq!( + *timeout, 100, + "should pass the configured 100ms timeout through unchanged" + ); + } + }); + } + + #[test] + fn parallel_path_quantizes_provider_transport_timeout() { + futures::executor::block_on(async { + // A 999ms budget must reach the provider as the 750ms quantum + // bucket — both in backend_name (which derives the Fastly backend + // name) and in context.timeout_ms (which configures the backend + // and payload deadlines) — so the backend name stays stable + // across requests with slightly different remaining budgets. + let stub = Arc::new(StubHttpClient::new()); + stub.push_response(200, b"{}".to_vec()); + let services = build_services_with_http_client(stub); + // SAFETY: `Box::leak` creates a `'static` reference for test use only. + // The leaked allocation is bounded to the test process lifetime. + let services: &'static RuntimeServices = Box::leak(Box::new(services)); + + let observed = Arc::new(Mutex::new(Vec::new())); + let config = AuctionConfig { + enabled: true, + providers: vec!["bidder".to_string()], + timeout_ms: 999, + mediator: None, + ..Default::default() + }; + let mut orchestrator = AuctionOrchestrator::new(config); + orchestrator.register_provider(Arc::new(StubAuctionProvider::recording( + "bidder", + "bidder-backend", + 2000, + Arc::clone(&observed), + ))); + + let request = create_test_auction_request(); + let settings = create_test_settings(); + let req = http::Request::builder() + .method(http::Method::GET) + .uri("https://example.com/test") + .body(edgezero_core::body::Body::empty()) + .expect("should build request"); + let context = AuctionContext { + settings: &settings, + request: &req, + timeout_ms: 999, + provider_responses: None, + services, + }; + + orchestrator + .run_auction(&request, &context) + .await + .expect("should complete auction"); + + let observed = observed.lock().expect("should lock observed timeouts"); + assert!( + !observed.is_empty(), + "should record provider transport timeouts" + ); + for timeout in observed.iter() { + assert!( + *timeout % super::TRANSPORT_TIMEOUT_QUANTUM_MS == 0 + && *timeout > 0 + && *timeout <= 750, + "should floor the 999ms budget to a quantum bucket at or below 750ms, got {timeout}ms" + ); + } + }); + } + + #[test] + fn sub_quantum_budget_launches_with_exact_remaining_timeout() { + futures::executor::block_on(async { + // A configured auction budget below one quantum must still launch + // providers with the exact remaining budget — rounding it to zero + // would hard-fail every auction for publishers with sub-250ms + // budgets. + let stub = Arc::new(StubHttpClient::new()); + stub.push_response(200, b"{}".to_vec()); + let services = build_services_with_http_client(stub); + // SAFETY: `Box::leak` creates a `'static` reference for test use only. + // The leaked allocation is bounded to the test process lifetime. + let services: &'static RuntimeServices = Box::leak(Box::new(services)); + + let observed = Arc::new(Mutex::new(Vec::new())); + let config = AuctionConfig { + enabled: true, + providers: vec!["bidder".to_string()], + timeout_ms: 200, + mediator: None, + ..Default::default() + }; + let mut orchestrator = AuctionOrchestrator::new(config); + orchestrator.register_provider(Arc::new(StubAuctionProvider::recording( + "bidder", + "bidder-backend", + 2000, + Arc::clone(&observed), + ))); + + let request = create_test_auction_request(); + let settings = create_test_settings(); + let req = http::Request::builder() + .method(http::Method::GET) + .uri("https://example.com/test") + .body(edgezero_core::body::Body::empty()) + .expect("should build request"); + let context = AuctionContext { + settings: &settings, + request: &req, + timeout_ms: 200, + provider_responses: None, + services, + }; + + let result = orchestrator + .run_auction(&request, &context) + .await + .expect("should complete auction with a sub-quantum budget"); + + assert_eq!( + result.provider_responses.len(), + 1, + "should launch the provider despite the sub-quantum budget" + ); + let observed = observed.lock().expect("should lock observed timeouts"); + assert!( + !observed.is_empty(), + "should record provider transport timeouts" + ); + for timeout in observed.iter() { + assert!( + *timeout > 0 && *timeout <= 200, + "should pass the exact sub-quantum remaining budget through, got {timeout}ms" + ); + } + }); + } + + #[test] + fn synchronous_mediation_quantizes_mediator_timeout() { + futures::executor::block_on(async { + // The mediator has no select-loop deadline backstop, so its + // transport timeout must be quantized by rounding down: a + // quantum-aligned value no larger than the remaining budget. + let stub = Arc::new(StubHttpClient::new()); + stub.push_response(200, b"{}".to_vec()); // bidder send_async + stub.push_response(200, b"{}".to_vec()); // mediator send_async + let services = build_services_with_http_client(stub); + // SAFETY: `Box::leak` creates a `'static` reference for test use only. + // The leaked allocation is bounded to the test process lifetime. + let services: &'static RuntimeServices = Box::leak(Box::new(services)); + + let observed = Arc::new(Mutex::new(Vec::new())); + let config = AuctionConfig { + enabled: true, + providers: vec!["bidder".to_string()], + mediator: Some("mediator".to_string()), + timeout_ms: 999, + ..Default::default() + }; + let mut orchestrator = AuctionOrchestrator::new(config); + orchestrator.register_provider(Arc::new(StubAuctionProvider::new( + "bidder", + "bidder-backend", + ))); + orchestrator.register_provider(Arc::new(StubAuctionProvider::recording( + "mediator", + "mediator-backend", + 2000, + Arc::clone(&observed), + ))); + + let request = create_test_auction_request(); + let settings = create_test_settings(); + let req = http::Request::builder() + .method(http::Method::GET) + .uri("https://example.com/test") + .body(edgezero_core::body::Body::empty()) + .expect("should build request"); + let context = AuctionContext { + settings: &settings, + request: &req, + timeout_ms: 999, + provider_responses: None, + services, + }; + + orchestrator + .run_auction(&request, &context) + .await + .expect("should complete mediated auction"); + + let observed = observed.lock().expect("should lock observed timeouts"); + assert!(!observed.is_empty(), "should run the mediator"); + for timeout in observed.iter() { + assert!( + *timeout % super::TRANSPORT_TIMEOUT_QUANTUM_MS == 0, + "mediator timeout {timeout}ms should be quantum-aligned" + ); + assert!( + *timeout > 0 && *timeout <= 750, + "mediator timeout {timeout}ms should be positive and floored below the 999ms budget" + ); + } + }); + } + + #[test] + fn dispatched_collect_quantizes_mediator_timeout() { + futures::executor::block_on(async { + // Same invariant as the synchronous path, on the split + // dispatch/collect path used by publisher page rendering. + let stub = Arc::new(StubHttpClient::new()); + stub.push_response(200, b"{}".to_vec()); // bidder send_async + stub.push_response(200, b"{}".to_vec()); // mediator send_async + let services = build_services_with_http_client(stub); + // SAFETY: `Box::leak` creates a `'static` reference for test use only. + // The leaked allocation is bounded to the test process lifetime. + let services: &'static RuntimeServices = Box::leak(Box::new(services)); + + let observed_bidder = Arc::new(Mutex::new(Vec::new())); + let observed_mediator = Arc::new(Mutex::new(Vec::new())); + let config = AuctionConfig { + enabled: true, + providers: vec!["bidder".to_string()], + mediator: Some("mediator".to_string()), + timeout_ms: 999, + ..Default::default() + }; + let mut orchestrator = AuctionOrchestrator::new(config); + orchestrator.register_provider(Arc::new(StubAuctionProvider::recording( + "bidder", + "bidder-backend", + 2000, + Arc::clone(&observed_bidder), + ))); + orchestrator.register_provider(Arc::new(StubAuctionProvider::recording( + "mediator", + "mediator-backend", + 2000, + Arc::clone(&observed_mediator), + ))); + + let request = create_test_auction_request(); + let settings = create_test_settings(); + let req = http::Request::builder() + .method(http::Method::GET) + .uri("https://example.com/test") + .body(edgezero_core::body::Body::empty()) + .expect("should build request"); + let context = AuctionContext { + settings: &settings, + request: &req, + timeout_ms: 999, + provider_responses: None, + services, + }; + + let dispatched = match orchestrator.dispatch_auction(&request, &context).await { + DispatchAuctionOutcome::Dispatched(dispatched) => dispatched, + _ => panic!("should dispatch the bidder request"), + }; + orchestrator + .collect_dispatched_auction(dispatched, services, &context) + .await; + + let observed_bidder = observed_bidder.lock().expect("should lock bidder timeouts"); + assert!( + !observed_bidder.is_empty(), + "should record dispatched bidder timeouts" + ); + for timeout in observed_bidder.iter() { + assert!( + *timeout % super::TRANSPORT_TIMEOUT_QUANTUM_MS == 0 + && *timeout > 0 + && *timeout <= 750, + "dispatched bidder timeout should floor 999ms to a quantum bucket at or below 750ms, got {timeout}ms" + ); + } + + let observed_mediator = observed_mediator + .lock() + .expect("should lock mediator timeouts"); + assert!(!observed_mediator.is_empty(), "should run the mediator"); + for timeout in observed_mediator.iter() { + assert!( + *timeout % super::TRANSPORT_TIMEOUT_QUANTUM_MS == 0, + "mediator timeout {timeout}ms should be quantum-aligned" + ); + assert!( + *timeout > 0 && *timeout <= 750, + "mediator timeout {timeout}ms should be positive and floored below the 999ms budget" + ); + } + }); + } + #[test] fn select_error_is_attributed_to_correct_provider() { futures::executor::block_on(async { @@ -1950,14 +2494,14 @@ mod tests { ..Default::default() }; let mut orchestrator = AuctionOrchestrator::new(config); - orchestrator.register_provider(Arc::new(StubAuctionProvider { - name: "provider-a", - backend: "backend-a", - })); - orchestrator.register_provider(Arc::new(StubAuctionProvider { - name: "provider-b", - backend: "backend-b", - })); + orchestrator.register_provider(Arc::new(StubAuctionProvider::new( + "provider-a", + "backend-a", + ))); + orchestrator.register_provider(Arc::new(StubAuctionProvider::new( + "provider-b", + "backend-b", + ))); let request = create_test_auction_request(); let settings = create_test_settings(); @@ -2033,14 +2577,14 @@ mod tests { ..Default::default() }; let mut orchestrator = AuctionOrchestrator::new(config); - orchestrator.register_provider(Arc::new(StubAuctionProvider { - name: "provider-a", - backend: "backend-a", - })); - orchestrator.register_provider(Arc::new(StubAuctionProvider { - name: "provider-b", - backend: "backend-b", - })); + orchestrator.register_provider(Arc::new(StubAuctionProvider::new( + "provider-a", + "backend-a", + ))); + orchestrator.register_provider(Arc::new(StubAuctionProvider::new( + "provider-b", + "backend-b", + ))); let request = create_test_auction_request(); let settings = create_test_settings(); @@ -2098,14 +2642,14 @@ mod tests { ..Default::default() }; let mut orchestrator = AuctionOrchestrator::new(config); - orchestrator.register_provider(Arc::new(StubAuctionProvider { - name: "provider-a", - backend: "backend-a", - })); - orchestrator.register_provider(Arc::new(StubAuctionProvider { - name: "provider-b", - backend: "backend-b", - })); + orchestrator.register_provider(Arc::new(StubAuctionProvider::new( + "provider-a", + "backend-a", + ))); + orchestrator.register_provider(Arc::new(StubAuctionProvider::new( + "provider-b", + "backend-b", + ))); let request = create_test_auction_request(); let settings = create_test_settings(); From 5512d8507cbfeee9777ddbb09be2f44ba734fc86 Mon Sep 17 00:00:00 2001 From: prk-Jr Date: Tue, 14 Jul 2026 18:10:20 +0530 Subject: [PATCH 2/6] Address auction transport-timeout review findings Resolve the PR review by making transport-timeout canonicalization a platform capability and hardening auction backend-name correlation. - Move quantization behind PlatformBackend::canonicalize_transport_timeout_ms. Fastly floors budget-derived timeouts to a 250ms quantum with a bounded sub-quantum ladder [200,150,100,50]; other adapters use the exact remaining budget so bidder deadlines (Prebid tmax, APS timeout) are not shortened where no connection-pooling benefit exists. - Bound sub-quantum backend-name cardinality: exact 1-249ms values no longer pass through, capping the budget-derived names a single origin can mint toward the per-service dynamic backend limit. - Add a provider discriminator to PlatformBackendSpec, folded into every adapter's backend name, so two providers sharing one origin no longer collide on the response-correlation key. Reject a duplicate backend_to_provider insertion with an attributed launch failure instead of silently overwriting and misattributing a response. - Make the orchestrator call-site tests deterministic: record predicted and registered transport timeouts separately and assert exact equality via a controllable platform backend, and enumerate the sub-quantum ladder to assert a bounded name cardinality. - Correct the timeout-semantics comments that overstated absolute-deadline enforcement; the Fastly connect/first-byte/between-bytes timeouts bound connection, first-byte, and inactivity, not total response time. A true absolute deadline carried through the platform HTTP API remains follow-up work (#849). --- .../src/platform.rs | 12 +- .../src/platform.rs | 9 +- .../src/backend.rs | 35 +- .../src/platform.rs | 200 +++++ .../src/tinybird.rs | 1 + .../src/platform.rs | 9 +- .../src/auction/orchestrator.rs | 794 ++++++++++-------- .../trusted-server-core/src/ec/pull_sync.rs | 1 + .../src/integrations/datadome/protection.rs | 1 + .../src/integrations/mod.rs | 4 + .../src/platform/test_support.rs | 28 + .../src/platform/traits.rs | 22 + .../trusted-server-core/src/platform/types.rs | 10 + crates/trusted-server-core/src/proxy.rs | 2 + crates/trusted-server-core/src/publisher.rs | 1 + 15 files changed, 772 insertions(+), 357 deletions(-) diff --git a/crates/trusted-server-adapter-axum/src/platform.rs b/crates/trusted-server-adapter-axum/src/platform.rs index 461b567d1..a511daab2 100644 --- a/crates/trusted-server-adapter-axum/src/platform.rs +++ b/crates/trusted-server-adapter-axum/src/platform.rs @@ -158,11 +158,19 @@ impl PlatformBackend for AxumPlatformBackend { let port = spec .port .unwrap_or(if spec.scheme == "https" { 443 } else { 80 }); + // Keep two providers that share an origin on distinct names so auction + // response correlation cannot cross providers. + let discriminator = spec + .discriminator + .as_deref() + .map(|d| format!("_p_{}", normalize_env_segment(d))) + .unwrap_or_default(); Ok(format!( - "{}_{}_{}", + "{}_{}_{}{}", normalize_env_segment(&spec.scheme), normalize_env_segment(&spec.host), port, + discriminator, )) } @@ -644,6 +652,7 @@ mod tests { first_byte_timeout: Duration::from_secs(15), between_bytes_timeout: Duration::from_secs(15), host_header_override: None, + discriminator: None, }; let name1 = backend.predict_name(&spec).expect("should return a name"); let name2 = backend @@ -664,6 +673,7 @@ mod tests { first_byte_timeout: Duration::from_secs(15), between_bytes_timeout: Duration::from_secs(15), host_header_override: None, + discriminator: None, }; assert_eq!( backend.predict_name(&spec).expect("should return name"), diff --git a/crates/trusted-server-adapter-cloudflare/src/platform.rs b/crates/trusted-server-adapter-cloudflare/src/platform.rs index a01c4d979..9467abb71 100644 --- a/crates/trusted-server-adapter-cloudflare/src/platform.rs +++ b/crates/trusted-server-adapter-cloudflare/src/platform.rs @@ -71,8 +71,15 @@ impl PlatformBackend for NoopBackend { } else { "_nocert" }; + // Keep two providers that share an origin on distinct names so auction + // response correlation cannot cross providers. + let discriminator = spec + .discriminator + .as_deref() + .map(|d| format!("_p_{d}")) + .unwrap_or_default(); Ok(format!( - "{}_{}_{}_{timeout_ms}ms{cert_suffix}", + "{}_{}_{}_{timeout_ms}ms{cert_suffix}{discriminator}", spec.scheme, spec.host, port )) } diff --git a/crates/trusted-server-adapter-fastly/src/backend.rs b/crates/trusted-server-adapter-fastly/src/backend.rs index 4056c81da..7205a8a00 100644 --- a/crates/trusted-server-adapter-fastly/src/backend.rs +++ b/crates/trusted-server-adapter-fastly/src/backend.rs @@ -64,6 +64,7 @@ pub struct BackendConfig<'a> { first_byte_timeout: Duration, between_bytes_timeout: Duration, host_header_override: Option<&'a str>, + discriminator: Option<&'a str>, } impl<'a> BackendConfig<'a> { @@ -81,6 +82,7 @@ impl<'a> BackendConfig<'a> { first_byte_timeout: DEFAULT_FIRST_BYTE_TIMEOUT, between_bytes_timeout: DEFAULT_BETWEEN_BYTES_TIMEOUT, host_header_override: None, + discriminator: None, } } @@ -128,14 +130,30 @@ impl<'a> BackendConfig<'a> { self } + /// Set an optional stable discriminator folded into the backend name. + /// + /// Two callers targeting the same origin with the same transport timeout + /// otherwise share a backend name. Auction response correlation keys on the + /// backend name, so a shared name would let one provider's response be + /// parsed as another's. A per-provider discriminator keeps the names + /// distinct while staying stable across requests. + #[must_use] + pub fn discriminator(mut self, discriminator: Option<&'a str>) -> Self { + self.discriminator = discriminator; + self + } + /// Compute the deterministic backend name and resolved port without /// registering anything. /// - /// The name encodes scheme, host, port, certificate setting, and - /// first-byte timeout so that backends with different configurations - /// never collide. Including the timeout prevents "first-registration-wins" - /// poisoning where a later request for the same origin with a tighter - /// timeout would silently inherit the original registration's value. + /// The name encodes scheme, host, port, certificate setting, optional + /// discriminator, and the first-byte/between-bytes timeouts so that + /// backends with different configurations never collide. Including the + /// timeout prevents "first-registration-wins" poisoning where a later + /// request for the same origin with a tighter timeout would silently + /// inherit the original registration's value. Including the discriminator + /// keeps two callers that target the same origin with the same timeout + /// (e.g. two auction providers behind one gateway) on distinct backends. fn compute_name(&self) -> Result<(String, u16), Report> { if self.host.is_empty() { return Err(Report::new(TrustedServerError::Proxy { @@ -174,13 +192,18 @@ impl<'a> BackendConfig<'a> { } else { "_nocert" }; + let discriminator_suffix = self + .discriminator + .map(|d| format!("_p_{}", sanitize_backend_name_component(d))) + .unwrap_or_default(); let first_byte_timeout_ms = self.first_byte_timeout.as_millis(); let between_bytes_timeout_ms = self.between_bytes_timeout.as_millis(); let backend_name = format!( - "backend_{}{}{}_fb{}_bb{}", + "backend_{}{}{}{}_fb{}_bb{}", sanitize_backend_name_component(&name_base), host_override_suffix, cert_suffix, + discriminator_suffix, first_byte_timeout_ms, between_bytes_timeout_ms ); diff --git a/crates/trusted-server-adapter-fastly/src/platform.rs b/crates/trusted-server-adapter-fastly/src/platform.rs index c5bb60b9c..c89508d36 100644 --- a/crates/trusted-server-adapter-fastly/src/platform.rs +++ b/crates/trusted-server-adapter-fastly/src/platform.rs @@ -156,6 +156,40 @@ fn backend_config_from_spec(spec: &PlatformBackendSpec) -> BackendConfig<'_> { .certificate_check(spec.certificate_check) .first_byte_timeout(spec.first_byte_timeout) .between_bytes_timeout(spec.between_bytes_timeout) + .discriminator(spec.discriminator.as_deref()) +} + +/// Transport-timeout quantum for auction backends (see +/// [`FastlyPlatformBackend::canonicalize_transport_timeout_ms`]). +const TRANSPORT_TIMEOUT_QUANTUM_MS: u32 = 250; + +/// Coarse rungs for budget-bound transport timeouts below one quantum, +/// ordered high to low. +/// +/// A budget-bound value at or above one quantum is floored to a +/// [`TRANSPORT_TIMEOUT_QUANTUM_MS`] multiple. Below one quantum, passing the +/// exact wall-clock remainder through would mint a distinct backend name for +/// every millisecond in `1..250`, so the near-exhausted tail alone could +/// exceed Fastly's per-service dynamic backend limit. Snapping to this finite +/// ladder instead bounds the number of budget-derived names an origin can +/// produce. Budgets below the smallest rung round to zero, which callers treat +/// as "budget exhausted — skip the launch". +const SUB_QUANTUM_LADDER_MS: [u32; 4] = [200, 150, 100, 50]; + +/// Round a budget-bound transport timeout down to a stable bucket. +/// +/// At or above one quantum, floors to a [`TRANSPORT_TIMEOUT_QUANTUM_MS`] +/// multiple. Below one quantum, snaps down to the greatest +/// [`SUB_QUANTUM_LADDER_MS`] rung no larger than `remaining_ms` (or zero). +fn quantize_transport_timeout_ms(remaining_ms: u32) -> u32 { + let floored = (remaining_ms / TRANSPORT_TIMEOUT_QUANTUM_MS) * TRANSPORT_TIMEOUT_QUANTUM_MS; + if floored > 0 { + return floored; + } + SUB_QUANTUM_LADDER_MS + .into_iter() + .find(|&rung| rung <= remaining_ms) + .unwrap_or(0) } impl PlatformBackend for FastlyPlatformBackend { @@ -170,6 +204,28 @@ impl PlatformBackend for FastlyPlatformBackend { .ensure() .change_context(PlatformError::Backend) } + + /// Quantize the transport timeout so budget-derived values do not mint a + /// new dynamic backend name on every request. + /// + /// Fastly embeds the first-byte and between-bytes timeouts in the dynamic + /// backend name (see [`BackendConfig`]) and pools connections per backend + /// name. A per-request wall-clock budget would otherwise defeat that + /// pooling and accumulate registrations toward the per-service dynamic + /// backend limit. + /// + /// A provider's own configured timeout is a constant, so when it is the + /// binding constraint it is returned verbatim — including sub-quantum + /// configured values, which must not be rounded away or the provider could + /// never launch. Only the budget-bound value is snapped to a stable bucket + /// via [`quantize_transport_timeout_ms`]. Rounding down never extends a + /// transport cap past the remaining budget. + fn canonicalize_transport_timeout_ms(&self, remaining_ms: u32, configured_ms: u32) -> u32 { + if remaining_ms >= configured_ms { + return configured_ms; + } + quantize_transport_timeout_ms(remaining_ms) + } } // --------------------------------------------------------------------------- @@ -637,6 +693,7 @@ mod tests { certificate_check: true, first_byte_timeout: Duration::from_secs(15), between_bytes_timeout: Duration::from_secs(15), + discriminator: None, }; let name = backend @@ -660,6 +717,7 @@ mod tests { certificate_check: true, first_byte_timeout: Duration::from_secs(15), between_bytes_timeout: Duration::from_secs(15), + discriminator: None, }; let name = backend @@ -683,6 +741,7 @@ mod tests { certificate_check: false, first_byte_timeout: Duration::from_secs(15), between_bytes_timeout: Duration::from_secs(15), + discriminator: None, }; let name = backend @@ -706,6 +765,7 @@ mod tests { certificate_check: true, first_byte_timeout: Duration::from_secs(15), between_bytes_timeout: Duration::from_secs(15), + discriminator: None, }; let result = backend.predict_name(&spec); @@ -724,6 +784,7 @@ mod tests { certificate_check: true, first_byte_timeout: Duration::from_millis(2000), between_bytes_timeout: Duration::from_millis(2000), + discriminator: None, }; let name = backend @@ -752,6 +813,7 @@ mod tests { certificate_check: true, first_byte_timeout: Duration::from_millis(750), between_bytes_timeout: Duration::from_millis(750), + discriminator: None, }; let predicted = backend @@ -937,4 +999,142 @@ mod tests { "should describe the unsupported streaming body: {err:?}" ); } + + // --- FastlyPlatformBackend::canonicalize_transport_timeout_ms ----------- + + #[test] + fn canonicalize_prefers_configured_timeout_when_budget_allows() { + let backend = FastlyPlatformBackend; + assert_eq!( + backend.canonicalize_transport_timeout_ms(2000, 1000), + 1000, + "should use the configured timeout verbatim when the budget allows" + ); + assert_eq!( + backend.canonicalize_transport_timeout_ms(2000, 100), + 100, + "should preserve a sub-quantum configured constant — it is name-stable on its own" + ); + } + + #[test] + fn canonicalize_floors_budget_bound_value_to_quantum() { + let backend = FastlyPlatformBackend; + assert_eq!( + backend.canonicalize_transport_timeout_ms(999, 2000), + 750, + "should floor a 999ms budget to the 750ms quantum bucket" + ); + assert_eq!( + backend.canonicalize_transport_timeout_ms(300, 2000), + 250, + "should floor a tight budget down to one quantum" + ); + assert_eq!( + backend.canonicalize_transport_timeout_ms(250, 2000), + 250, + "should keep an exact quantum multiple" + ); + } + + #[test] + fn canonicalize_snaps_sub_quantum_budget_to_bounded_ladder() { + let backend = FastlyPlatformBackend; + // Exact wall-clock values in 1..250 must NOT pass through — that is the + // unbounded-cardinality regression this ladder closes. + assert_eq!( + backend.canonicalize_transport_timeout_ms(249, 2000), + 200, + "should snap a sub-quantum budget down to the greatest ladder rung, not pass 249 through" + ); + assert_eq!(backend.canonicalize_transport_timeout_ms(200, 2000), 200); + assert_eq!(backend.canonicalize_transport_timeout_ms(150, 2000), 150); + assert_eq!(backend.canonicalize_transport_timeout_ms(100, 2000), 100); + assert_eq!(backend.canonicalize_transport_timeout_ms(50, 2000), 50); + assert_eq!( + backend.canonicalize_transport_timeout_ms(49, 2000), + 0, + "a budget below the smallest rung rounds to zero (launch skipped)" + ); + assert_eq!( + backend.canonicalize_transport_timeout_ms(0, 1000), + 0, + "an exhausted budget canonicalizes to zero" + ); + assert_eq!( + backend.canonicalize_transport_timeout_ms(100, 0), + 0, + "a zero configured timeout canonicalizes to zero" + ); + } + + #[test] + fn canonicalize_budget_derived_names_stay_within_a_safe_cardinality() { + // Enumerate every reachable remaining budget for a normal 2000ms + // ceiling and confirm the number of distinct backend-name-bearing + // transport values an origin can mint stays far below Fastly's + // per-service dynamic backend limit (documented default 200). + let backend = FastlyPlatformBackend; + let configured = 2000; + let mut distinct = std::collections::BTreeSet::new(); + for remaining in 0..=configured { + let value = backend.canonicalize_transport_timeout_ms(remaining, configured); + if value > 0 { + distinct.insert(value); + } + // No arbitrary clock-derived value may leak: every canonical value + // is either a quantum multiple or one of the bounded ladder rungs. + assert!( + value == 0 + || value % TRANSPORT_TIMEOUT_QUANTUM_MS == 0 + || SUB_QUANTUM_LADDER_MS.contains(&value), + "canonical value {value}ms (from remaining {remaining}ms) is neither a quantum \ + multiple nor a ladder rung" + ); + } + assert!( + distinct.len() <= 16, + "budget-derived transport values should stay well under the dynamic backend limit, \ + got {} distinct values: {distinct:?}", + distinct.len() + ); + } + + // --- FastlyPlatformBackend::predict_name discriminator ------------------ + + #[test] + fn predict_name_includes_provider_discriminator() { + let backend = FastlyPlatformBackend; + let base = PlatformBackendSpec { + scheme: "https".to_string(), + host: "gateway.example.com".to_string(), + port: None, + host_header_override: None, + certificate_check: true, + first_byte_timeout: Duration::from_millis(750), + between_bytes_timeout: Duration::from_millis(750), + discriminator: Some("prebid".to_string()), + }; + let prebid_name = backend + .predict_name(&base) + .expect("should predict name with discriminator"); + assert!( + prebid_name.contains("_p_prebid"), + "should fold the provider discriminator into the name, got {prebid_name}" + ); + + // Same origin + same transport timeout, different provider → distinct + // backend names, so auction response correlation cannot cross them. + let aps = PlatformBackendSpec { + discriminator: Some("aps".to_string()), + ..base.clone() + }; + let aps_name = backend + .predict_name(&aps) + .expect("should predict name for the second provider"); + assert_ne!( + prebid_name, aps_name, + "two providers on one origin must not share a backend name" + ); + } } diff --git a/crates/trusted-server-adapter-fastly/src/tinybird.rs b/crates/trusted-server-adapter-fastly/src/tinybird.rs index b5d332a65..8df6dbe6e 100644 --- a/crates/trusted-server-adapter-fastly/src/tinybird.rs +++ b/crates/trusted-server-adapter-fastly/src/tinybird.rs @@ -212,6 +212,7 @@ fn tinybird_backend_spec(api_host: &str) -> PlatformBackendSpec { certificate_check: true, first_byte_timeout: TINYBIRD_FIRST_BYTE_TIMEOUT, between_bytes_timeout: TINYBIRD_BETWEEN_BYTES_TIMEOUT, + discriminator: None, } } diff --git a/crates/trusted-server-adapter-spin/src/platform.rs b/crates/trusted-server-adapter-spin/src/platform.rs index 1e13ca300..492f1a518 100644 --- a/crates/trusted-server-adapter-spin/src/platform.rs +++ b/crates/trusted-server-adapter-spin/src/platform.rs @@ -92,8 +92,15 @@ impl PlatformBackend for NoopBackend { } else { "_nocert" }; + // Keep two providers that share an origin on distinct names so auction + // response correlation cannot cross providers. + let discriminator = spec + .discriminator + .as_deref() + .map(|d| format!("_p_{d}")) + .unwrap_or_default(); Ok(format!( - "{}_{}_{}_{timeout_ms}ms{cert_suffix}", + "{}_{}_{}_{timeout_ms}ms{cert_suffix}{discriminator}", spec.scheme, spec.host, port )) } diff --git a/crates/trusted-server-core/src/auction/orchestrator.rs b/crates/trusted-server-core/src/auction/orchestrator.rs index d884a0220..02a87cdb5 100644 --- a/crates/trusted-server-core/src/auction/orchestrator.rs +++ b/crates/trusted-server-core/src/auction/orchestrator.rs @@ -157,64 +157,6 @@ fn remaining_budget_ms(start: Instant, timeout_ms: u32) -> u32 { timeout_ms.saturating_sub(elapsed) } -/// Transport-timeout quantum for auction backends. -/// -/// See [`quantize_transport_timeout_ms`] for why provider transport timeouts -/// are rounded to this granularity. -const TRANSPORT_TIMEOUT_QUANTUM_MS: u32 = 250; - -/// Round a transport timeout down to a [`TRANSPORT_TIMEOUT_QUANTUM_MS`] multiple. -/// -/// The Fastly adapter embeds the first-byte and between-bytes timeouts in the -/// dynamic backend name so a registration can never be silently reused with a -/// different transport configuration. Deriving those timeouts from the -/// remaining wall-clock budget minted a new backend name on nearly every -/// request, which defeated cross-request TCP/TLS connection reuse (Fastly -/// pools connections per backend name) and accumulated registrations toward -/// the per-service dynamic backend limit. -/// -/// Quantizing the value — not just the name — keeps the registered backend -/// configuration aligned with its name. Rounding down never extends a -/// transport cap past the auction deadline, which matters on the mediator and -/// dispatched-collect paths where the backend timeouts (not a select-loop -/// deadline check) bound the `` hold. -#[inline] -fn quantize_transport_timeout_ms(timeout_ms: u32) -> u32 { - (timeout_ms / TRANSPORT_TIMEOUT_QUANTUM_MS) * TRANSPORT_TIMEOUT_QUANTUM_MS -} - -/// Compute the transport timeout for a provider launch from the remaining -/// auction budget and the provider's configured timeout. -/// -/// The configured timeout is a per-provider constant, so using it verbatim -/// already yields a stable backend name — including configured values below -/// one quantum, which must not be rounded away or the provider could never -/// launch. Only when the remaining budget is the binding constraint does the -/// wall-clock-derived value enter the name, and that value is quantized via -/// [`quantize_transport_timeout_ms`] so it cannot mint a new backend name on -/// every request. -/// -/// A remaining budget below one quantum is passed through exactly rather -/// than rounded to zero: rounding up would extend the transport cap past the -/// deadline, and rounding down would skip the launch and hard-fail auctions -/// whose configured budget is under one quantum. Name churn in this regime -/// is bounded to sub-quantum values and matches the pre-quantization -/// behavior. The result never exceeds `remaining_ms` and is zero only when -/// `remaining_ms` or `configured_ms` is zero, which callers treat as -/// "budget exhausted — skip the launch". -#[inline] -fn effective_transport_timeout_ms(remaining_ms: u32, configured_ms: u32) -> u32 { - if remaining_ms >= configured_ms { - return configured_ms; - } - let quantized = quantize_transport_timeout_ms(remaining_ms); - if quantized == 0 { - remaining_ms - } else { - quantized - } -} - /// Manages auction execution across multiple providers. pub struct AuctionOrchestrator { config: AuctionConfig, @@ -338,11 +280,16 @@ impl AuctionOrchestrator { // Give the mediator only the remaining time from the auction // deadline, not the full timeout — the bidding phase already // consumed part of it, and the mediator has no select-loop - // deadline backstop. Quantized for backend-name stability (see - // effective_transport_timeout_ms). + // deadline backstop. The platform canonicalizes the value for + // backend-name stability (see + // `PlatformBackend::canonicalize_transport_timeout_ms`); it never + // exceeds the remaining budget. See the transport-deadline note on + // `run_providers_parallel` for the limits of this bound. let remaining_ms = remaining_budget_ms(mediation_start, context.timeout_ms); - let mediator_timeout = - effective_transport_timeout_ms(remaining_ms, mediator.timeout_ms()); + let mediator_timeout = context + .services + .backend() + .canonicalize_transport_timeout_ms(remaining_ms, mediator.timeout_ms()); if mediator_timeout == 0 { log::warn!("Auction timeout exhausted during bidding phase; skipping mediator"); @@ -525,11 +472,14 @@ impl AuctionOrchestrator { // Give each provider only the remaining time from the auction // deadline so that backend transport timeouts do not extend past - // the overall budget, quantized for backend-name stability (see - // effective_transport_timeout_ms). + // the overall budget. The platform canonicalizes the value for + // backend-name stability (see + // `PlatformBackend::canonicalize_transport_timeout_ms`). let remaining_ms = remaining_budget_ms(auction_start, context.timeout_ms); - let effective_timeout = - effective_transport_timeout_ms(remaining_ms, provider.timeout_ms()); + let effective_timeout = context + .services + .backend() + .canonicalize_transport_timeout_ms(remaining_ms, provider.timeout_ms()); if effective_timeout == 0 { log::warn!("Auction timeout exhausted before launching provider request; skipping"); @@ -580,15 +530,35 @@ impl AuctionOrchestrator { ); backend_name.clone() }); - backend_to_provider.insert( - request_backend_name.clone(), - (provider.provider_name(), start_time, provider.as_ref()), - ); - pending_requests.push(pending); - log::debug!( - "Request to '{}' launched successfully", - provider.provider_name() - ); + // Responses are correlated back to providers by backend + // name. If another provider this auction already claimed + // this name (e.g. two providers on one origin whose specs + // canonicalize to the same backend), inserting here would + // silently overwrite the first mapping and misattribute or + // drop a response. Fail this launch attributably instead. + if backend_to_provider.contains_key(&request_backend_name) { + let response_time_ms = start_time.elapsed().as_millis() as u64; + log::warn!( + "Provider '{}' resolved to backend name '{}' already claimed by another \ + provider this auction; skipping launch to avoid response misattribution", + provider.provider_name(), + request_backend_name, + ); + responses.push(provider_launch_failed_response( + provider.provider_name(), + response_time_ms, + )); + } else { + backend_to_provider.insert( + request_backend_name.clone(), + (provider.provider_name(), start_time, provider.as_ref()), + ); + pending_requests.push(pending); + log::debug!( + "Request to '{}' launched successfully", + provider.provider_name() + ); + } } Err(e) => { let response_time_ms = start_time.elapsed().as_millis() as u64; @@ -621,14 +591,25 @@ impl AuctionOrchestrator { ); // Phase 2: Wait for responses using select() to process as they become ready. - // Enforce the auction deadline: after each select() returns, check - // elapsed time and drop remaining requests if the timeout is exceeded. + // After each select() returns, check elapsed time and drop remaining + // requests once the auction deadline passes. // - // NOTE: `select()` blocks until at least one backend responds and, on - // some adapters, buffers the selected response body before returning. - // Hard deadline enforcement therefore depends on every backend's - // first-byte and between-bytes timeouts being set to at most the - // remaining auction budget, which Phase 1 above guarantees. + // TRANSPORT-DEADLINE NOTE: this select loop is the only *absolute* + // wall-clock bound on the parallel path — it drops still-pending + // requests once `auction_start.elapsed()` exceeds the deadline. The + // per-backend transport timeouts set in Phase 1 are a complementary, + // not equivalent, bound: Fastly's connect timeout is a fixed ~1s, the + // first-byte timeout only starts after the connection is established, + // and the between-bytes timeout is an inactivity timer that resets on + // every byte received. A backend that connects slowly or trickles one + // byte just inside the between-bytes window can therefore outlive the + // configured budget. Bounding them to the remaining budget (Phase 1) + // guarantees they never *extend past* the deadline by their own + // configuration, but does not by itself enforce a hard total-response + // deadline. Paths without this select loop (the mediator and the + // dispatched-collect body read) inherit that weaker bound; a true + // absolute deadline carried through the platform HTTP API is tracked + // as follow-up work (see the streaming/deadline effort, #849). let mut remaining = pending_requests; while !remaining.is_empty() { @@ -937,11 +918,13 @@ impl AuctionOrchestrator { continue; } - // Remaining budget quantized for backend-name stability (see - // effective_transport_timeout_ms). + // Remaining budget canonicalized by the platform for backend-name + // stability (see `PlatformBackend::canonicalize_transport_timeout_ms`). let remaining_ms = remaining_budget_ms(auction_start, context.timeout_ms); - let effective_timeout = - effective_transport_timeout_ms(remaining_ms, provider.timeout_ms()); + let effective_timeout = context + .services + .backend() + .canonicalize_transport_timeout_ms(remaining_ms, provider.timeout_ms()); if effective_timeout == 0 { log::warn!( @@ -974,21 +957,39 @@ impl AuctionOrchestrator { let start_time = Instant::now(); match provider.request_bids(request, &provider_context).await { Ok(pending) => { - log::info!( - "Dispatching bid request to '{}' (backend: {}, budget: {}ms)", - provider.provider_name(), - backend_name, - effective_timeout - ); - backend_to_provider.insert( - backend_name.clone(), - ( - provider.provider_name().to_string(), - start_time, - Arc::clone(provider), - ), - ); - pending_requests.push(pending.with_backend_name(backend_name)); + // See the parallel path: a backend name already claimed by + // another provider this auction would misattribute the + // collected response, so fail this launch attributably + // rather than overwrite the mapping. + if backend_to_provider.contains_key(&backend_name) { + let response_time_ms = start_time.elapsed().as_millis() as u64; + log::warn!( + "Provider '{}' resolved to backend name '{}' already claimed by another \ + provider this auction; skipping dispatch to avoid response misattribution", + provider.provider_name(), + backend_name, + ); + launch_responses.push(provider_launch_failed_response( + provider.provider_name(), + response_time_ms, + )); + } else { + log::info!( + "Dispatching bid request to '{}' (backend: {}, budget: {}ms)", + provider.provider_name(), + backend_name, + effective_timeout + ); + backend_to_provider.insert( + backend_name.clone(), + ( + provider.provider_name().to_string(), + start_time, + Arc::clone(provider), + ), + ); + pending_requests.push(pending.with_backend_name(backend_name)); + } } Err(e) => { let response_time_ms = start_time.elapsed().as_millis() as u64; @@ -1189,21 +1190,27 @@ impl AuctionOrchestrator { match self.providers.get(mediator_name.as_str()) { Some(mediator) => { // Cap the mediator at whichever is tighter: its own configured - // timeout or the remaining auction budget (A_deadline). The old - // comment here claimed origin drain could exhaust the budget before - // collection, but SSP backends are given first-byte and between-bytes - // timeouts equal to effective_timeout (capped at their provider - // timeout) at dispatch time, so they cannot run past A_deadline - // independently. Giving the mediator an uncapped timeout lets it run - // past A_deadline, violating the bounded hold invariant. - // The mediator's only time bound on this path is its - // backend transport timeout, so the effective value must - // never exceed the remaining budget. Quantized for - // backend-name stability (see - // effective_transport_timeout_ms). + // timeout or the remaining auction budget (A_deadline). Giving + // the mediator an uncapped timeout would let it hold `` + // well past A_deadline, so the effective value must never + // exceed the remaining budget. + // + // Caveat: unlike the parallel select loop, this path has no + // absolute wall-clock backstop around the mediator call, and a + // backend transport timeout bounds first-byte/inactivity rather + // than total response time (see the transport-deadline note on + // `run_providers_parallel`). Capping the value to the remaining + // budget therefore prevents the mediator from *extending* the + // hold by its own configuration, but a slow-connecting or + // byte-trickling mediator can still overrun; a true absolute + // deadline is tracked as follow-up (#849). + // + // The platform canonicalizes the value for backend-name + // stability (see `PlatformBackend::canonicalize_transport_timeout_ms`). let remaining = remaining_budget_ms(auction_start, timeout_ms); - let mediator_timeout = - effective_transport_timeout_ms(remaining, mediator.timeout_ms()); + let mediator_timeout = services + .backend() + .canonicalize_transport_timeout_ms(remaining, mediator.timeout_ms()); if mediator_timeout == 0 { log::warn!( "A_deadline exhausted before mediator '{}' — returning {} SSP bids without mediation", @@ -1397,9 +1404,13 @@ mod tests { MediaType, PublisherInfo, UserInfo, }; use crate::error::TrustedServerError; - use crate::platform::test_support::{build_services_with_http_client, StubHttpClient}; + use crate::platform::test_support::{ + build_services_with_backend_and_http_client, build_services_with_http_client, + StubHttpClient, + }; use crate::platform::{ - PlatformHttpRequest, PlatformPendingRequest, PlatformResponse, RuntimeServices, + PlatformBackend, PlatformBackendSpec, PlatformError, PlatformHttpRequest, + PlatformPendingRequest, PlatformResponse, RuntimeServices, }; use crate::test_support::tests::crate_test_settings_str; use error_stack::{Report, ResultExt}; @@ -1412,15 +1423,19 @@ mod tests { // Minimal test double for AuctionProvider // --------------------------------------------------------------------------- - /// Minimal stub provider. Optionally records every transport timeout it - /// observes — the value passed to `backend_name` and the - /// `context.timeout_ms` handed to `request_bids` — so tests can assert - /// the orchestrator quantizes them. + /// Minimal stub provider. Optionally records the transport timeouts it + /// observes, keeping the value passed to `backend_name` (which derives the + /// predicted backend name) separate from the `context.timeout_ms` handed to + /// `request_bids` (which configures the registered request). Recording them + /// separately lets tests assert the orchestrator hands the *same* + /// canonicalized value to both — a divergence would land responses in the + /// "unknown backend" branch and drop bids. struct StubAuctionProvider { name: &'static str, backend: &'static str, configured_timeout_ms: u32, - observed_timeouts: Option>>>, + predicted_timeouts: Option>>>, + request_timeouts: Option>>>, } impl StubAuctionProvider { @@ -1429,7 +1444,8 @@ mod tests { name, backend, configured_timeout_ms: 2000, - observed_timeouts: None, + predicted_timeouts: None, + request_timeouts: None, } } @@ -1437,18 +1453,20 @@ mod tests { name: &'static str, backend: &'static str, configured_timeout_ms: u32, - observed_timeouts: Arc>>, + predicted_timeouts: Arc>>, + request_timeouts: Arc>>, ) -> Self { Self { name, backend, configured_timeout_ms, - observed_timeouts: Some(observed_timeouts), + predicted_timeouts: Some(predicted_timeouts), + request_timeouts: Some(request_timeouts), } } - fn record(&self, timeout_ms: u32) { - if let Some(observed) = &self.observed_timeouts { + fn record(slot: &Option>>>, timeout_ms: u32) { + if let Some(observed) = slot { observed .lock() .expect("should lock observed timeouts") @@ -1468,7 +1486,7 @@ mod tests { _request: &AuctionRequest, context: &AuctionContext<'_>, ) -> Result> { - self.record(context.timeout_ms); + Self::record(&self.request_timeouts, context.timeout_ms); let req = PlatformHttpRequest::new( http::Request::builder() .method("POST") @@ -1504,7 +1522,7 @@ mod tests { } fn backend_name(&self, _services: &RuntimeServices, timeout_ms: u32) -> Option { - self.record(timeout_ms); + Self::record(&self.predicted_timeouts, timeout_ms); Some(self.backend.to_string()) } } @@ -2038,94 +2056,71 @@ mod tests { ); } - #[test] - fn quantize_transport_timeout_floors_to_quantum() { - assert_eq!( - super::quantize_transport_timeout_ms(0), - 0, - "should keep zero at zero" - ); - assert_eq!( - super::quantize_transport_timeout_ms(249), - 0, - "should floor a sub-quantum budget to zero" - ); - assert_eq!( - super::quantize_transport_timeout_ms(250), - 250, - "should keep an exact quantum multiple unchanged" - ); - assert_eq!( - super::quantize_transport_timeout_ms(999), - 750, - "should floor to the next-lower quantum multiple" - ); - assert_eq!( - super::quantize_transport_timeout_ms(2000), - 2000, - "should keep a larger exact quantum multiple unchanged" - ); + /// Test backend whose [`PlatformBackend::canonicalize_transport_timeout_ms`] + /// returns a fixed value regardless of the wall-clock budget, so the + /// orchestrator's transport-timeout wiring can be asserted without timing + /// flakiness. Records every `(remaining_ms, configured_ms)` pair it sees. + /// + /// The exact quantization arithmetic lives in the Fastly adapter (the only + /// platform that overrides `canonicalize_transport_timeout_ms`); these core + /// tests only prove the orchestrator applies whatever the platform returns + /// and applies it identically to the predicted name and the launched + /// request. + struct CanonicalTimeoutBackend { + canonical_ms: u32, + calls: Arc>>, } - #[test] - fn effective_transport_timeout_prefers_configured_constant() { - assert_eq!( - super::effective_transport_timeout_ms(2000, 1000), - 1000, - "should use the configured timeout verbatim when the budget allows" - ); - assert_eq!( - super::effective_transport_timeout_ms(2000, 100), - 100, - "should preserve a sub-quantum configured timeout — quantizing it away would permanently disable the provider" - ); - assert_eq!( - super::effective_transport_timeout_ms(999, 2000), - 750, - "should quantize the budget-bound value down to the 750ms bucket" - ); - assert_eq!( - super::effective_transport_timeout_ms(300, 2000), - 250, - "should quantize a tight budget down to one quantum" - ); - assert_eq!( - super::effective_transport_timeout_ms(200, 2000), - 200, - "should pass a sub-quantum budget through exactly instead of rounding to zero" - ); - assert_eq!( - super::effective_transport_timeout_ms(50, 100), - 50, - "should pass through when the budget is below both the quantum and the configured timeout" - ); - assert_eq!( - super::effective_transport_timeout_ms(0, 1000), - 0, - "should return zero for an exhausted budget so the launch is skipped" - ); - assert_eq!( - super::effective_transport_timeout_ms(100, 0), - 0, - "should return zero for a zero configured timeout so the launch is skipped" - ); + impl CanonicalTimeoutBackend { + fn new(canonical_ms: u32, calls: Arc>>) -> Self { + Self { + canonical_ms, + calls, + } + } + } + + impl PlatformBackend for CanonicalTimeoutBackend { + fn predict_name( + &self, + _spec: &PlatformBackendSpec, + ) -> Result> { + Ok("stub-backend".to_owned()) + } + + fn ensure(&self, _spec: &PlatformBackendSpec) -> Result> { + Ok("stub-backend".to_owned()) + } + + fn canonicalize_transport_timeout_ms(&self, remaining_ms: u32, configured_ms: u32) -> u32 { + self.calls + .lock() + .expect("should lock canonicalize calls") + .push((remaining_ms, configured_ms)); + self.canonical_ms + } } #[test] - fn sub_quantum_configured_timeout_still_launches_provider() { + fn parallel_launch_applies_canonical_timeout_to_name_and_request() { futures::executor::block_on(async { - // A provider whose configured timeout is below one quantum must - // still launch with its exact configured value: the constant is - // name-stable on its own, so only budget-derived values are - // quantized. + // The orchestrator must hand the platform-canonicalized value to + // BOTH `backend_name` (which derives the correlation key) and + // `request_bids` (via `context.timeout_ms`). Recording them + // separately and asserting exact equality catches a regression that + // predicts one bucket but registers another — which would drop the + // response into the "unknown backend" branch. let stub = Arc::new(StubHttpClient::new()); stub.push_response(200, b"{}".to_vec()); - let services = build_services_with_http_client(stub); + let calls = Arc::new(Mutex::new(Vec::new())); + let backend = Arc::new(CanonicalTimeoutBackend::new(750, Arc::clone(&calls))); + let services = build_services_with_backend_and_http_client(backend, stub); // SAFETY: `Box::leak` creates a `'static` reference for test use only. // The leaked allocation is bounded to the test process lifetime. let services: &'static RuntimeServices = Box::leak(Box::new(services)); - let observed = Arc::new(Mutex::new(Vec::new())); + let predicted = Arc::new(Mutex::new(Vec::new())); + let requested = Arc::new(Mutex::new(Vec::new())); let config = AuctionConfig { enabled: true, providers: vec!["bidder".to_string()], @@ -2137,8 +2132,9 @@ mod tests { orchestrator.register_provider(Arc::new(StubAuctionProvider::recording( "bidder", "bidder-backend", - 100, - Arc::clone(&observed), + 1000, + Arc::clone(&predicted), + Arc::clone(&requested), ))); let request = create_test_auction_request(); @@ -2161,49 +2157,62 @@ mod tests { .await .expect("should complete auction"); - let observed = observed.lock().expect("should lock observed timeouts"); + let predicted = predicted.lock().expect("should lock predicted"); + let requested = requested.lock().expect("should lock requested"); + assert_eq!( + *predicted, + vec![750], + "backend_name should receive the canonicalized value" + ); + assert_eq!( + *requested, + vec![750], + "request_bids should receive the same canonicalized value" + ); + assert_eq!( + *predicted, *requested, + "predicted and registered transport timeouts must be identical" + ); + + let calls = calls.lock().expect("should lock calls"); + assert_eq!(calls.len(), 1, "should canonicalize once for the launch"); + let (remaining_ms, configured_ms) = calls[0]; + assert_eq!( + configured_ms, 1000, + "should pass the provider's configured timeout as the configured bound" + ); assert!( - !observed.is_empty(), - "should launch the sub-quantum-configured provider" + remaining_ms > 0 && remaining_ms <= 2000, + "should pass the live remaining budget, got {remaining_ms}ms" ); - for timeout in observed.iter() { - assert_eq!( - *timeout, 100, - "should pass the configured 100ms timeout through unchanged" - ); - } }); } #[test] - fn parallel_path_quantizes_provider_transport_timeout() { + fn zero_canonical_timeout_skips_parallel_launch() { futures::executor::block_on(async { - // A 999ms budget must reach the provider as the 750ms quantum - // bucket — both in backend_name (which derives the Fastly backend - // name) and in context.timeout_ms (which configures the backend - // and payload deadlines) — so the backend name stays stable - // across requests with slightly different remaining budgets. + // A platform that canonicalizes to zero signals "budget exhausted"; + // the orchestrator must skip the launch. With the only provider + // skipped, no requests launch and the auction errors. let stub = Arc::new(StubHttpClient::new()); - stub.push_response(200, b"{}".to_vec()); - let services = build_services_with_http_client(stub); + let calls = Arc::new(Mutex::new(Vec::new())); + let backend = Arc::new(CanonicalTimeoutBackend::new(0, Arc::clone(&calls))); + let services = build_services_with_backend_and_http_client(backend, stub); // SAFETY: `Box::leak` creates a `'static` reference for test use only. // The leaked allocation is bounded to the test process lifetime. let services: &'static RuntimeServices = Box::leak(Box::new(services)); - let observed = Arc::new(Mutex::new(Vec::new())); let config = AuctionConfig { enabled: true, providers: vec!["bidder".to_string()], - timeout_ms: 999, + timeout_ms: 2000, mediator: None, ..Default::default() }; let mut orchestrator = AuctionOrchestrator::new(config); - orchestrator.register_provider(Arc::new(StubAuctionProvider::recording( + orchestrator.register_provider(Arc::new(StubAuctionProvider::new( "bidder", "bidder-backend", - 2000, - Arc::clone(&observed), ))); let request = create_test_auction_request(); @@ -2216,60 +2225,55 @@ mod tests { let context = AuctionContext { settings: &settings, request: &req, - timeout_ms: 999, + timeout_ms: 2000, provider_responses: None, services, }; - orchestrator - .run_auction(&request, &context) - .await - .expect("should complete auction"); - - let observed = observed.lock().expect("should lock observed timeouts"); + let result = orchestrator.run_auction(&request, &context).await; assert!( - !observed.is_empty(), - "should record provider transport timeouts" + result.is_err(), + "should error when the only provider is skipped for an exhausted budget" ); - for timeout in observed.iter() { - assert!( - *timeout % super::TRANSPORT_TIMEOUT_QUANTUM_MS == 0 - && *timeout > 0 - && *timeout <= 750, - "should floor the 999ms budget to a quantum bucket at or below 750ms, got {timeout}ms" - ); - } }); } #[test] - fn sub_quantum_budget_launches_with_exact_remaining_timeout() { + fn synchronous_mediation_applies_canonical_timeout_to_mediator() { futures::executor::block_on(async { - // A configured auction budget below one quantum must still launch - // providers with the exact remaining budget — rounding it to zero - // would hard-fail every auction for publishers with sub-250ms - // budgets. + // The mediator runs after the bidding phase and has no select-loop + // backstop; it must still receive the platform-canonicalized value + // for both prediction and request. let stub = Arc::new(StubHttpClient::new()); - stub.push_response(200, b"{}".to_vec()); - let services = build_services_with_http_client(stub); + stub.push_response(200, b"{}".to_vec()); // bidder send_async + stub.push_response(200, b"{}".to_vec()); // mediator send_async + let calls = Arc::new(Mutex::new(Vec::new())); + let backend = Arc::new(CanonicalTimeoutBackend::new(500, Arc::clone(&calls))); + let services = build_services_with_backend_and_http_client(backend, stub); // SAFETY: `Box::leak` creates a `'static` reference for test use only. // The leaked allocation is bounded to the test process lifetime. let services: &'static RuntimeServices = Box::leak(Box::new(services)); - let observed = Arc::new(Mutex::new(Vec::new())); + let predicted = Arc::new(Mutex::new(Vec::new())); + let requested = Arc::new(Mutex::new(Vec::new())); let config = AuctionConfig { enabled: true, providers: vec!["bidder".to_string()], - timeout_ms: 200, - mediator: None, + mediator: Some("mediator".to_string()), + timeout_ms: 2000, ..Default::default() }; let mut orchestrator = AuctionOrchestrator::new(config); - orchestrator.register_provider(Arc::new(StubAuctionProvider::recording( + orchestrator.register_provider(Arc::new(StubAuctionProvider::new( "bidder", "bidder-backend", + ))); + orchestrator.register_provider(Arc::new(StubAuctionProvider::recording( + "mediator", + "mediator-backend", 2000, - Arc::clone(&observed), + Arc::clone(&predicted), + Arc::clone(&requested), ))); let request = create_test_auction_request(); @@ -2282,67 +2286,76 @@ mod tests { let context = AuctionContext { settings: &settings, request: &req, - timeout_ms: 200, + timeout_ms: 2000, provider_responses: None, services, }; - let result = orchestrator + orchestrator .run_auction(&request, &context) .await - .expect("should complete auction with a sub-quantum budget"); + .expect("should complete mediated auction"); - assert_eq!( - result.provider_responses.len(), - 1, - "should launch the provider despite the sub-quantum budget" - ); - let observed = observed.lock().expect("should lock observed timeouts"); + let predicted = predicted.lock().expect("should lock predicted"); + let requested = requested.lock().expect("should lock requested"); + // The orchestrator hands the mediator its budget through + // `context.timeout_ms` and calls `request_bids` directly; it does not + // call the mediator's `backend_name` (the mediator self-registers its + // backend), so only the request side is observed here. assert!( - !observed.is_empty(), - "should record provider transport timeouts" + predicted.is_empty(), + "orchestrator should not separately predict a backend name for the mediator" + ); + assert_eq!( + *requested, + vec![500], + "mediator request should use the canonical value" ); - for timeout in observed.iter() { - assert!( - *timeout > 0 && *timeout <= 200, - "should pass the exact sub-quantum remaining budget through, got {timeout}ms" - ); - } }); } #[test] - fn synchronous_mediation_quantizes_mediator_timeout() { + fn dispatched_collect_applies_canonical_timeout_to_both_paths() { futures::executor::block_on(async { - // The mediator has no select-loop deadline backstop, so its - // transport timeout must be quantized by rounding down: a - // quantum-aligned value no larger than the remaining budget. + // Same wiring invariant on the split dispatch/collect path used by + // publisher page rendering: the dispatched bidder and the collected + // mediator both receive the canonicalized value for prediction and + // request. let stub = Arc::new(StubHttpClient::new()); stub.push_response(200, b"{}".to_vec()); // bidder send_async stub.push_response(200, b"{}".to_vec()); // mediator send_async - let services = build_services_with_http_client(stub); + let calls = Arc::new(Mutex::new(Vec::new())); + let backend = Arc::new(CanonicalTimeoutBackend::new(500, Arc::clone(&calls))); + let services = build_services_with_backend_and_http_client(backend, stub); // SAFETY: `Box::leak` creates a `'static` reference for test use only. // The leaked allocation is bounded to the test process lifetime. let services: &'static RuntimeServices = Box::leak(Box::new(services)); - let observed = Arc::new(Mutex::new(Vec::new())); + let bidder_predicted = Arc::new(Mutex::new(Vec::new())); + let bidder_requested = Arc::new(Mutex::new(Vec::new())); + let mediator_predicted = Arc::new(Mutex::new(Vec::new())); + let mediator_requested = Arc::new(Mutex::new(Vec::new())); let config = AuctionConfig { enabled: true, providers: vec!["bidder".to_string()], mediator: Some("mediator".to_string()), - timeout_ms: 999, + timeout_ms: 2000, ..Default::default() }; let mut orchestrator = AuctionOrchestrator::new(config); - orchestrator.register_provider(Arc::new(StubAuctionProvider::new( + orchestrator.register_provider(Arc::new(StubAuctionProvider::recording( "bidder", "bidder-backend", + 2000, + Arc::clone(&bidder_predicted), + Arc::clone(&bidder_requested), ))); orchestrator.register_provider(Arc::new(StubAuctionProvider::recording( "mediator", "mediator-backend", 2000, - Arc::clone(&observed), + Arc::clone(&mediator_predicted), + Arc::clone(&mediator_requested), ))); let request = create_test_auction_request(); @@ -2355,65 +2368,168 @@ mod tests { let context = AuctionContext { settings: &settings, request: &req, - timeout_ms: 999, + timeout_ms: 2000, provider_responses: None, services, }; + let dispatched = match orchestrator.dispatch_auction(&request, &context).await { + DispatchAuctionOutcome::Dispatched(dispatched) => dispatched, + _ => panic!("should dispatch the bidder request"), + }; orchestrator + .collect_dispatched_auction(dispatched, services, &context) + .await; + + let bidder_predicted = bidder_predicted + .lock() + .expect("should lock bidder predicted"); + let bidder_requested = bidder_requested + .lock() + .expect("should lock bidder requested"); + assert_eq!( + *bidder_predicted, + vec![500], + "dispatched bidder name should use canonical value" + ); + assert_eq!( + *bidder_requested, + vec![500], + "dispatched bidder request should use canonical value" + ); + assert_eq!( + *bidder_predicted, *bidder_requested, + "dispatched bidder predicted and registered timeouts must be identical" + ); + + let mediator_predicted = mediator_predicted + .lock() + .expect("should lock mediator predicted"); + let mediator_requested = mediator_requested + .lock() + .expect("should lock mediator requested"); + // As on the synchronous path, the orchestrator calls the mediator's + // `request_bids` directly without predicting a backend name for it. + assert!( + mediator_predicted.is_empty(), + "orchestrator should not separately predict a backend name for the mediator" + ); + assert_eq!( + *mediator_requested, + vec![500], + "mediator request should use the canonical value" + ); + }); + } + + #[test] + fn parallel_duplicate_backend_name_fails_second_provider_attributably() { + futures::executor::block_on(async { + // Two providers that canonicalize to the SAME backend name (e.g. two + // auction providers behind one gateway origin). The correlation map + // keys on backend name, so the second must not silently overwrite + // the first — it must fail attributably so no bid is misparsed or + // lost. + let stub = Arc::new(StubHttpClient::new()); + stub.push_response(200, b"{}".to_vec()); // provider-a send_async + stub.push_response(200, b"{}".to_vec()); // provider-b send_async (dropped after guard) + let services = build_services_with_http_client(stub); + // SAFETY: `Box::leak` creates a `'static` reference for test use only. + // The leaked allocation is bounded to the test process lifetime. + let services: &'static RuntimeServices = Box::leak(Box::new(services)); + + let config = AuctionConfig { + enabled: true, + providers: vec!["provider-a".to_string(), "provider-b".to_string()], + timeout_ms: 2000, + mediator: None, + ..Default::default() + }; + let mut orchestrator = AuctionOrchestrator::new(config); + orchestrator.register_provider(Arc::new(StubAuctionProvider::new( + "provider-a", + "shared-backend", + ))); + orchestrator.register_provider(Arc::new(StubAuctionProvider::new( + "provider-b", + "shared-backend", + ))); + + let request = create_test_auction_request(); + let settings = create_test_settings(); + let req = http::Request::builder() + .method(http::Method::GET) + .uri("https://example.com/test") + .body(edgezero_core::body::Body::empty()) + .expect("should build request"); + let context = AuctionContext { + settings: &settings, + request: &req, + timeout_ms: 2000, + provider_responses: None, + services, + }; + + let result = orchestrator .run_auction(&request, &context) .await - .expect("should complete mediated auction"); + .expect("should complete auction despite the name collision"); - let observed = observed.lock().expect("should lock observed timeouts"); - assert!(!observed.is_empty(), "should run the mediator"); - for timeout in observed.iter() { - assert!( - *timeout % super::TRANSPORT_TIMEOUT_QUANTUM_MS == 0, - "mediator timeout {timeout}ms should be quantum-aligned" - ); - assert!( - *timeout > 0 && *timeout <= 750, - "mediator timeout {timeout}ms should be positive and floored below the 999ms budget" - ); - } + assert_eq!( + result.provider_responses.len(), + 2, + "should account for both providers" + ); + let provider_a = result + .provider_responses + .iter() + .find(|r| r.provider == "provider-a") + .expect("should have provider-a response"); + let provider_b = result + .provider_responses + .iter() + .find(|r| r.provider == "provider-b") + .expect("should have provider-b response"); + assert_eq!( + provider_a.status, + BidStatus::Success, + "the first provider on the shared name should launch and succeed" + ); + assert_eq!( + provider_b.status, + BidStatus::Error, + "the second provider on the shared name should fail attributably, not be dropped" + ); }); } #[test] - fn dispatched_collect_quantizes_mediator_timeout() { + fn dispatched_duplicate_backend_name_fails_second_provider_attributably() { futures::executor::block_on(async { - // Same invariant as the synchronous path, on the split - // dispatch/collect path used by publisher page rendering. + // Same collision defense on the dispatch/collect path. let stub = Arc::new(StubHttpClient::new()); - stub.push_response(200, b"{}".to_vec()); // bidder send_async - stub.push_response(200, b"{}".to_vec()); // mediator send_async + stub.push_response(200, b"{}".to_vec()); // provider-a send_async + stub.push_response(200, b"{}".to_vec()); // provider-b send_async (dropped after guard) let services = build_services_with_http_client(stub); // SAFETY: `Box::leak` creates a `'static` reference for test use only. // The leaked allocation is bounded to the test process lifetime. let services: &'static RuntimeServices = Box::leak(Box::new(services)); - let observed_bidder = Arc::new(Mutex::new(Vec::new())); - let observed_mediator = Arc::new(Mutex::new(Vec::new())); let config = AuctionConfig { enabled: true, - providers: vec!["bidder".to_string()], - mediator: Some("mediator".to_string()), - timeout_ms: 999, + providers: vec!["provider-a".to_string(), "provider-b".to_string()], + timeout_ms: 2000, + mediator: None, ..Default::default() }; let mut orchestrator = AuctionOrchestrator::new(config); - orchestrator.register_provider(Arc::new(StubAuctionProvider::recording( - "bidder", - "bidder-backend", - 2000, - Arc::clone(&observed_bidder), + orchestrator.register_provider(Arc::new(StubAuctionProvider::new( + "provider-a", + "shared-backend", ))); - orchestrator.register_provider(Arc::new(StubAuctionProvider::recording( - "mediator", - "mediator-backend", - 2000, - Arc::clone(&observed_mediator), + orchestrator.register_provider(Arc::new(StubAuctionProvider::new( + "provider-b", + "shared-backend", ))); let request = create_test_auction_request(); @@ -2426,47 +2542,29 @@ mod tests { let context = AuctionContext { settings: &settings, request: &req, - timeout_ms: 999, + timeout_ms: 2000, provider_responses: None, services, }; let dispatched = match orchestrator.dispatch_auction(&request, &context).await { DispatchAuctionOutcome::Dispatched(dispatched) => dispatched, - _ => panic!("should dispatch the bidder request"), + _ => panic!("should dispatch the first provider despite the name collision"), }; - orchestrator + let result = orchestrator .collect_dispatched_auction(dispatched, services, &context) .await; - let observed_bidder = observed_bidder.lock().expect("should lock bidder timeouts"); - assert!( - !observed_bidder.is_empty(), - "should record dispatched bidder timeouts" + let provider_b = result + .provider_responses + .iter() + .find(|r| r.provider == "provider-b") + .expect("should have provider-b response"); + assert_eq!( + provider_b.status, + BidStatus::Error, + "the second provider on the shared name should fail attributably, not be dropped" ); - for timeout in observed_bidder.iter() { - assert!( - *timeout % super::TRANSPORT_TIMEOUT_QUANTUM_MS == 0 - && *timeout > 0 - && *timeout <= 750, - "dispatched bidder timeout should floor 999ms to a quantum bucket at or below 750ms, got {timeout}ms" - ); - } - - let observed_mediator = observed_mediator - .lock() - .expect("should lock mediator timeouts"); - assert!(!observed_mediator.is_empty(), "should run the mediator"); - for timeout in observed_mediator.iter() { - assert!( - *timeout % super::TRANSPORT_TIMEOUT_QUANTUM_MS == 0, - "mediator timeout {timeout}ms should be quantum-aligned" - ); - assert!( - *timeout > 0 && *timeout <= 750, - "mediator timeout {timeout}ms should be positive and floored below the 999ms budget" - ); - } }); } diff --git a/crates/trusted-server-core/src/ec/pull_sync.rs b/crates/trusted-server-core/src/ec/pull_sync.rs index fa096d59d..833898b50 100644 --- a/crates/trusted-server-core/src/ec/pull_sync.rs +++ b/crates/trusted-server-core/src/ec/pull_sync.rs @@ -174,6 +174,7 @@ pub fn dispatch_pull_sync( certificate_check: settings.proxy.certificate_check, first_byte_timeout: DEFAULT_FIRST_BYTE_TIMEOUT, between_bytes_timeout: DEFAULT_FIRST_BYTE_TIMEOUT, + discriminator: None, }) { Ok(name) => name, Err(err) => { diff --git a/crates/trusted-server-core/src/integrations/datadome/protection.rs b/crates/trusted-server-core/src/integrations/datadome/protection.rs index 717ad46e8..eedcf7cd2 100644 --- a/crates/trusted-server-core/src/integrations/datadome/protection.rs +++ b/crates/trusted-server-core/src/integrations/datadome/protection.rs @@ -160,6 +160,7 @@ impl DataDomeIntegration { certificate_check: true, first_byte_timeout: Duration::from_millis(u64::from(self.config.timeout_ms)), between_bytes_timeout: Duration::from_millis(u64::from(self.config.timeout_ms)), + discriminator: None, }; services.backend().ensure(&spec).change_context(Self::error( diff --git a/crates/trusted-server-core/src/integrations/mod.rs b/crates/trusted-server-core/src/integrations/mod.rs index 2052215f2..75b4693ff 100644 --- a/crates/trusted-server-core/src/integrations/mod.rs +++ b/crates/trusted-server-core/src/integrations/mod.rs @@ -153,6 +153,10 @@ fn integration_backend_spec( certificate_check, first_byte_timeout, between_bytes_timeout: first_byte_timeout, + // Distinguish this integration's backend from any other provider that + // targets the same origin, so auction response correlation by backend + // name cannot cross providers. + discriminator: Some(integration.to_string()), }) } diff --git a/crates/trusted-server-core/src/platform/test_support.rs b/crates/trusted-server-core/src/platform/test_support.rs index ee7201fb8..d6ffee23b 100644 --- a/crates/trusted-server-core/src/platform/test_support.rs +++ b/crates/trusted-server-core/src/platform/test_support.rs @@ -649,6 +649,33 @@ pub(crate) fn noop_services_with_client_ip(ip: IpAddr) -> RuntimeServices { .build() } +/// Build a [`RuntimeServices`] with a caller-supplied [`PlatformBackend`] and +/// HTTP client. +/// +/// Lets auction tests inject a backend whose +/// [`PlatformBackend::canonicalize_transport_timeout_ms`] returns a controlled +/// value, so the orchestrator's transport-timeout wiring can be asserted +/// deterministically without depending on wall-clock timing. +pub(crate) fn build_services_with_backend_and_http_client( + backend: Arc, + http_client: Arc, +) -> RuntimeServices { + RuntimeServices::builder() + .config_store(Arc::new(NoopConfigStore)) + .secret_store(Arc::new(NoopSecretStore)) + .kv_store(Arc::new(edgezero_core::key_value_store::NoopKvStore)) + .backend(backend) + .http_client(http_client) + .geo(Arc::new(NoopGeo)) + .client_info(ClientInfo { + client_ip: None, + tls_protocol: None, + tls_cipher: None, + ..ClientInfo::default() + }) + .build() +} + /// Build a [`RuntimeServices`] with a custom secret store, [`StubBackend`], and HTTP client. pub(crate) fn build_services_with_secret_and_http_client( secret_store: impl PlatformSecretStore + 'static, @@ -856,6 +883,7 @@ mod tests { certificate_check: true, first_byte_timeout: DEFAULT_FIRST_BYTE_TIMEOUT, between_bytes_timeout: DEFAULT_FIRST_BYTE_TIMEOUT, + discriminator: None, }; let name = stub.ensure(&spec).expect("should return a backend name"); assert_eq!(name, "stub-backend", "should return fixed name"); diff --git a/crates/trusted-server-core/src/platform/traits.rs b/crates/trusted-server-core/src/platform/traits.rs index ecc886c9d..c6af0a307 100644 --- a/crates/trusted-server-core/src/platform/traits.rs +++ b/crates/trusted-server-core/src/platform/traits.rs @@ -110,6 +110,28 @@ pub trait PlatformBackend: Send + Sync { /// Returns [`PlatformError::Backend`] when the backend cannot be /// registered on the platform. fn ensure(&self, spec: &PlatformBackendSpec) -> Result>; + + /// Canonicalize a per-provider transport timeout for backend-name stability. + /// + /// `remaining_ms` is the wall-clock budget left in the auction and + /// `configured_ms` is the provider's own configured timeout. The returned + /// value is used both to derive the dynamic backend name and as the + /// provider's request deadline, so it must be identical for prediction and + /// registration of the same launch. + /// + /// Adapters that embed the transport timeout in the dynamic backend name + /// (Fastly) override this to round budget-derived values to a coarse + /// ladder, so per-request wall-clock jitter neither defeats cross-request + /// connection pooling nor accumulates registrations toward the per-service + /// dynamic backend limit. + /// + /// The default returns the exact budget-bound value + /// (`remaining_ms.min(configured_ms)`): adapters that neither register nor + /// enforce a backend-name transport timeout gain nothing from rounding and + /// must not shorten bidder deadlines for no benefit. + fn canonicalize_transport_timeout_ms(&self, remaining_ms: u32, configured_ms: u32) -> u32 { + remaining_ms.min(configured_ms) + } } /// Synchronous, object-safe geo lookup. diff --git a/crates/trusted-server-core/src/platform/types.rs b/crates/trusted-server-core/src/platform/types.rs index a81c24105..a39a26430 100644 --- a/crates/trusted-server-core/src/platform/types.rs +++ b/crates/trusted-server-core/src/platform/types.rs @@ -143,6 +143,16 @@ pub struct PlatformBackendSpec { pub first_byte_timeout: Duration, /// Maximum time to wait between response body bytes. pub between_bytes_timeout: Duration, + /// Optional stable discriminator folded into the backend name. + /// + /// Two callers can target the same origin (scheme, host, port, TLS) with + /// the same transport timeout yet need distinct dynamic backends — for + /// example two auction providers behind one gateway host. Because the + /// auction orchestrator correlates responses back to providers by backend + /// name, a shared name would let one provider's response be parsed as + /// another's. Setting this to a per-provider/integration identifier keeps + /// their names distinct while remaining stable across requests. + pub discriminator: Option, } /// Cloneable container of platform services for a single request. diff --git a/crates/trusted-server-core/src/proxy.rs b/crates/trusted-server-core/src/proxy.rs index 19c10a80d..d1a5cdc57 100644 --- a/crates/trusted-server-core/src/proxy.rs +++ b/crates/trusted-server-core/src/proxy.rs @@ -1099,6 +1099,7 @@ pub async fn handle_asset_proxy_request( certificate_check: settings.proxy.certificate_check, first_byte_timeout: DEFAULT_FIRST_BYTE_TIMEOUT, between_bytes_timeout: DEFAULT_FIRST_BYTE_TIMEOUT, + discriminator: None, }) .change_context(TrustedServerError::Proxy { message: "asset backend registration failed".to_string(), @@ -1289,6 +1290,7 @@ async fn proxy_with_redirects( certificate_check: settings.proxy.certificate_check, first_byte_timeout: DEFAULT_FIRST_BYTE_TIMEOUT, between_bytes_timeout: DEFAULT_FIRST_BYTE_TIMEOUT, + discriminator: None, }) .change_context(TrustedServerError::Proxy { message: "backend registration failed".to_string(), diff --git a/crates/trusted-server-core/src/publisher.rs b/crates/trusted-server-core/src/publisher.rs index 6b0ea3a5d..7a2b9b437 100644 --- a/crates/trusted-server-core/src/publisher.rs +++ b/crates/trusted-server-core/src/publisher.rs @@ -1368,6 +1368,7 @@ pub async fn handle_publisher_request( certificate_check: settings.proxy.certificate_check, first_byte_timeout: DEFAULT_PUBLISHER_FIRST_BYTE_TIMEOUT, between_bytes_timeout: DEFAULT_PUBLISHER_FIRST_BYTE_TIMEOUT, + discriminator: None, }) .change_context(TrustedServerError::Proxy { message: "backend registration failed".to_string(), From 4828512f0e66febff899ccfae9d5de4082db2668 Mon Sep 17 00:00:00 2001 From: prk-Jr Date: Thu, 16 Jul 2026 17:55:42 +0530 Subject: [PATCH 3/6] Harden Fastly backend naming and transport-timeout quantization Address the auction transport-timeout review findings: - Derive dynamic backend names from a bounded readable prefix plus a SHA-256 digest of the complete backend spec, so distinct specs never alias to one name. Name equality now implies spec equality, making NameInUse reuse provably safe, and the name is bounded to Fastly's 255-char limit regardless of host or discriminator length. - Bound budget-derived transport-timeout cardinality with a globally finite ladder: keep the 250ms quantum through 2000ms, then snap larger budgets to a small fixed set of coarse rungs so a large configured ceiling can no longer mint hundreds of dynamic backends. - Reject duplicate configured providers at startup, and check the predicted backend name before request_bids fires the outbound send, retaining the post-launch check as defense against a provider that resolves to an unexpected name. --- Cargo.lock | 1 + .../trusted-server-adapter-fastly/Cargo.toml | 1 + .../src/backend.rs | 261 +++++++++++++++--- .../src/platform.rs | 95 ++++++- .../src/auction/orchestrator.rs | 84 +++++- 5 files changed, 390 insertions(+), 52 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index a19be7abb..82e644964 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5209,6 +5209,7 @@ dependencies = [ "log-fastly", "serde", "serde_json", + "sha2 0.10.9", "trusted-server-core", "url", "urlencoding", diff --git a/crates/trusted-server-adapter-fastly/Cargo.toml b/crates/trusted-server-adapter-fastly/Cargo.toml index 8547fd519..6f3d092d2 100644 --- a/crates/trusted-server-adapter-fastly/Cargo.toml +++ b/crates/trusted-server-adapter-fastly/Cargo.toml @@ -21,6 +21,7 @@ log = { workspace = true } log-fastly = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } +sha2 = { workspace = true } trusted-server-core = { workspace = true } url = { workspace = true } urlencoding = { workspace = true } diff --git a/crates/trusted-server-adapter-fastly/src/backend.rs b/crates/trusted-server-adapter-fastly/src/backend.rs index 7205a8a00..09b5d54dd 100644 --- a/crates/trusted-server-adapter-fastly/src/backend.rs +++ b/crates/trusted-server-adapter-fastly/src/backend.rs @@ -2,6 +2,7 @@ use std::time::Duration; use error_stack::{Report, ResultExt as _}; use fastly::backend::Backend; +use sha2::{Digest as _, Sha256}; use url::Url; use trusted_server_core::error::TrustedServerError; @@ -47,6 +48,34 @@ fn sanitize_backend_name_component(value: &str) -> String { .collect() } +/// Fastly's documented maximum length for a dynamic backend name. +const MAX_BACKEND_NAME_LEN: usize = 255; +/// Maximum length of the human-readable prefix folded into a backend name. +/// +/// Bounds the name so that `backend__` can never exceed +/// [`MAX_BACKEND_NAME_LEN`]: 8 (`backend_`) + 200 + 1 (`_`) + +/// [`SPEC_DIGEST_HEX_LEN`] = 241 ≤ 255. +const MAX_READABLE_PREFIX_LEN: usize = 200; +/// Width of the hex digest suffix — the first 128 bits of a SHA-256 over the +/// full backend spec, which is collision-resistant at the handful-of-hundreds +/// scale of a service's dynamic backends. +const SPEC_DIGEST_HEX_LEN: usize = 32; + +/// Hex-encode the first 128 bits of a SHA-256 digest of `canonical`. +/// +/// Used to make a backend name a collision-resistant function of the complete +/// backend spec (see [`BackendConfig::canonical_spec_string`]). +fn spec_digest_hex(canonical: &str) -> String { + let mut hasher = Sha256::new(); + hasher.update(canonical.as_bytes()); + let digest = hasher.finalize(); + digest + .iter() + .take(SPEC_DIGEST_HEX_LEN / 2) + .map(|byte| format!("{byte:02x}")) + .collect() +} + /// Default first-byte timeout for backends (15 seconds). pub(crate) const DEFAULT_FIRST_BYTE_TIMEOUT: Duration = Duration::from_secs(15); /// Default timeout between response body bytes for backends (10 seconds). @@ -143,17 +172,61 @@ impl<'a> BackendConfig<'a> { self } + /// Build an unambiguous, length-prefixed encoding of the complete backend + /// spec for digesting. + /// + /// Every field is prefixed with its byte length so that no two distinct + /// specs can encode to the same string (a lossy substitution like + /// `sanitize_backend_name_component` cannot guarantee this). `Option` fields + /// are presence-tagged so a `None` never aliases a `Some("")`. The result is + /// fed to [`spec_digest_hex`]; it is never parsed, only hashed. + fn canonical_spec_string(&self, target_port: u16) -> String { + fn push_field(buf: &mut String, field: &str) { + buf.push_str(&field.len().to_string()); + buf.push(':'); + buf.push_str(field); + } + + let mut buf = String::new(); + push_field(&mut buf, self.scheme); + push_field(&mut buf, self.host); + push_field(&mut buf, &target_port.to_string()); + push_field(&mut buf, if self.certificate_check { "1" } else { "0" }); + match self.host_header_override { + Some(value) => { + buf.push('s'); + push_field(&mut buf, value); + } + None => buf.push('n'), + } + match self.discriminator { + Some(value) => { + buf.push('s'); + push_field(&mut buf, value); + } + None => buf.push('n'), + } + push_field(&mut buf, &self.first_byte_timeout.as_millis().to_string()); + push_field(&mut buf, &self.between_bytes_timeout.as_millis().to_string()); + buf + } + /// Compute the deterministic backend name and resolved port without /// registering anything. /// - /// The name encodes scheme, host, port, certificate setting, optional - /// discriminator, and the first-byte/between-bytes timeouts so that - /// backends with different configurations never collide. Including the - /// timeout prevents "first-registration-wins" poisoning where a later - /// request for the same origin with a tighter timeout would silently - /// inherit the original registration's value. Including the discriminator - /// keeps two callers that target the same origin with the same timeout - /// (e.g. two auction providers behind one gateway) on distinct backends. + /// The name is `backend__`, where `` is a + /// collision-resistant SHA-256 over an unambiguous encoding of the + /// *complete* backend spec — scheme, host, port, certificate setting, Host + /// override, provider discriminator, and the first-byte/between-bytes + /// timeouts (see [`canonical_spec_string`](Self::canonical_spec_string)). + /// Because distinct specs yield distinct digests, name equality implies spec + /// equality: that is what makes reusing a `NameInUse` backend provably safe, + /// and it prevents "first-registration-wins" poisoning where a later request + /// with a tighter timeout would inherit an earlier registration's value. The + /// `` half is a lossy, bounded slug carried only for logs — any + /// collision there is harmless because uniqueness comes from the digest. The + /// whole name is bounded to [`MAX_BACKEND_NAME_LEN`] so a long host or + /// discriminator can never produce a name Fastly rejects at registration. fn compute_name(&self) -> Result<(String, u16), Report> { if self.host.is_empty() { return Err(Report::new(TrustedServerError::Proxy { @@ -198,8 +271,13 @@ impl<'a> BackendConfig<'a> { .unwrap_or_default(); let first_byte_timeout_ms = self.first_byte_timeout.as_millis(); let between_bytes_timeout_ms = self.between_bytes_timeout.as_millis(); - let backend_name = format!( - "backend_{}{}{}{}_fb{}_bb{}", + + // Lossy, human-readable slug for logs. Correctness does not depend on + // it — uniqueness comes from the digest below — so it is bounded to a + // fixed length. Sanitization only emits ASCII, so a char-boundary take + // is byte-exact. + let readable_full = format!( + "{}{}{}{}_fb{}_bb{}", sanitize_backend_name_component(&name_base), host_override_suffix, cert_suffix, @@ -207,6 +285,23 @@ impl<'a> BackendConfig<'a> { first_byte_timeout_ms, between_bytes_timeout_ms ); + let readable: String = readable_full.chars().take(MAX_READABLE_PREFIX_LEN).collect(); + + // Collision-resistant over the *complete* spec, so name equality implies + // spec equality and `NameInUse` reuse is safe. + let digest = spec_digest_hex(&self.canonical_spec_string(target_port)); + let backend_name = format!("backend_{readable}_{digest}"); + + // Bounded by construction; assert it so any future format change fails + // attributably during prediction rather than at Fastly registration. + if backend_name.len() > MAX_BACKEND_NAME_LEN { + return Err(Report::new(TrustedServerError::Proxy { + message: format!( + "backend name exceeds {MAX_BACKEND_NAME_LEN}-char limit ({} chars)", + backend_name.len() + ), + })); + } Ok((backend_name, target_port)) } @@ -373,7 +468,35 @@ impl<'a> BackendConfig<'a> { #[cfg(test)] mod tests { - use super::{compute_host_header, BackendConfig}; + use super::{ + compute_host_header, BackendConfig, MAX_BACKEND_NAME_LEN, SPEC_DIGEST_HEX_LEN, + }; + + /// Assert a computed name is `backend__` and stays within + /// Fastly's length limit. The digest is what makes the name injective, so + /// checking its presence and width guards the collision-safety property. + fn assert_backend_name_shape(name: &str, expected_body: &str) { + let prefix = format!("backend_{expected_body}_"); + assert!( + name.starts_with(&prefix), + "name should start with the readable body `{prefix}`, got {name}" + ); + let digest = &name[prefix.len()..]; + assert_eq!( + digest.len(), + SPEC_DIGEST_HEX_LEN, + "digest suffix should be {SPEC_DIGEST_HEX_LEN} hex chars, got {digest}" + ); + assert!( + digest.bytes().all(|byte| byte.is_ascii_hexdigit()), + "digest suffix should be hex, got {digest}" + ); + assert!( + name.len() <= MAX_BACKEND_NAME_LEN, + "name should stay within the {MAX_BACKEND_NAME_LEN}-char limit, got {}", + name.len() + ); + } // Tests for compute_host_header - the fix for port preservation in Host header #[test] @@ -422,7 +545,7 @@ mod tests { let name = BackendConfig::new("https", "origin.example.com") .ensure() .expect("should create backend for valid HTTPS origin"); - assert_eq!(name, "backend_https_origin_example_com_443_fb15000_bb10000"); + assert_backend_name_shape(&name, "https_origin_example_com_443_fb15000_bb10000"); } #[test] @@ -431,10 +554,7 @@ mod tests { .certificate_check(false) .ensure() .expect("should create backend with cert check disabled"); - assert_eq!( - name, - "backend_https_origin_example_com_443_nocert_fb15000_bb10000" - ); + assert_backend_name_shape(&name, "https_origin_example_com_443_nocert_fb15000_bb10000"); } #[test] @@ -443,7 +563,7 @@ mod tests { .port(Some(8080)) .ensure() .expect("should create backend for HTTP origin with explicit port"); - assert_eq!(name, "backend_http_api_test-site_org_8080_fb15000_bb10000"); + assert_backend_name_shape(&name, "http_api_test-site_org_8080_fb15000_bb10000"); } #[test] @@ -451,7 +571,7 @@ mod tests { let name = BackendConfig::new("http", "example.org") .ensure() .expect("should create backend defaulting to port 80 for HTTP"); - assert_eq!(name, "backend_http_example_org_80_fb15000_bb10000"); + assert_backend_name_shape(&name, "http_example_org_80_fb15000_bb10000"); } #[test] @@ -506,13 +626,13 @@ mod tests { name_a, name_b, "backends with different host header overrides should have different names" ); - assert_eq!( - name_a, - "backend_https_origin_example_com_443_oh_www_example_com_fb15000_bb10000" + assert_backend_name_shape( + &name_a, + "https_origin_example_com_443_oh_www_example_com_fb15000_bb10000", ); - assert_eq!( - name_b, - "backend_https_origin_example_com_443_oh_m_example_com_fb15000_bb10000" + assert_backend_name_shape( + &name_b, + "https_origin_example_com_443_oh_m_example_com_fb15000_bb10000", ); } @@ -567,12 +687,12 @@ mod tests { "backends with different timeouts should have different names" ); assert!( - name_a.ends_with("_fb2000_bb10000"), - "name should include first-byte and between-bytes timeout suffix" + name_a.contains("_fb2000_bb10000_"), + "name should include first-byte and between-bytes timeout in the readable body" ); assert!( - name_b.ends_with("_fb500_bb10000"), - "name should include first-byte and between-bytes timeout suffix" + name_b.contains("_fb500_bb10000_"), + "name should include first-byte and between-bytes timeout in the readable body" ); } @@ -594,12 +714,89 @@ mod tests { "backends with different between-bytes timeouts should have different names" ); assert!( - name_a.ends_with("_fb15000_bb2000"), - "name should include first-byte and between-bytes timeout suffix" + name_a.contains("_fb15000_bb2000_"), + "name should include first-byte and between-bytes timeout in the readable body" ); assert!( - name_b.ends_with("_fb15000_bb500"), - "name should include first-byte and between-bytes timeout suffix" + name_b.contains("_fb15000_bb500_"), + "name should include first-byte and between-bytes timeout in the readable body" + ); + } + + #[test] + fn discriminators_that_sanitize_alike_produce_distinct_names() { + // `provider.a` and `provider_a` both sanitize to the same readable slug + // (`.` maps to `_`). Before the spec digest they collided to one backend + // name, so the second registration silently reused the first — routing + // one provider's auction traffic through another's backend. The digest + // over the raw spec must keep them distinct. + let dotted = BackendConfig::new("https", "gateway.example.com") + .discriminator(Some("provider.a")) + .predict_name() + .expect("should predict name for dotted discriminator"); + let underscored = BackendConfig::new("https", "gateway.example.com") + .discriminator(Some("provider_a")) + .predict_name() + .expect("should predict name for underscored discriminator"); + assert_ne!( + dotted, underscored, + "discriminators differing only by a sanitized character must not collide" + ); + } + + #[test] + fn host_overrides_that_sanitize_alike_produce_distinct_names() { + // `host.example.com:8443` (host+port) and `host.example.com.8443` (DNS + // label) are both valid overrides that sanitize to the same readable + // slug (`:` and `.` both map to `_`). The digest over the raw value must + // keep the two backends — with different Host routing — distinct. + let with_port = BackendConfig::new("https", "origin.example.com") + .host_header_override(Some("host.example.com:8443")) + .predict_name() + .expect("should predict name for host:port override"); + let with_label = BackendConfig::new("https", "origin.example.com") + .host_header_override(Some("host.example.com.8443")) + .predict_name() + .expect("should predict name for dotted-label override"); + assert_ne!( + with_port, with_label, + "host overrides differing only by a sanitized character must not collide" + ); + } + + #[test] + fn long_host_and_discriminator_stay_within_the_length_limit() { + // A syntactically valid maximum-length DNS host plus a discriminator + // previously pushed the name past Fastly's 255-char limit, so + // `predict_name` succeeded while `ensure` failed at registration. The + // bounded prefix + fixed-width digest must keep prediction, and the name + // it predicts, within the limit. + let label = "a".repeat(63); + let long_host = format!("{label}.{label}.{label}.{label}.example.com"); + assert!( + long_host.len() > 200, + "should exercise a host longer than the readable-prefix bound" + ); + let name = BackendConfig::new("https", &long_host) + .discriminator(Some("prebid")) + .predict_name() + .expect("should predict a bounded name for a long host and discriminator"); + assert!( + name.len() <= MAX_BACKEND_NAME_LEN, + "name should stay within the {MAX_BACKEND_NAME_LEN}-char limit, got {}", + name.len() + ); + + // Two long hosts sharing the truncated prefix must still resolve to + // different backends via the digest. + let other_host = format!("{label}.{label}.{label}.{label}.example.net"); + let other = BackendConfig::new("https", &other_host) + .discriminator(Some("prebid")) + .predict_name() + .expect("should predict a bounded name for the sibling host"); + assert_ne!( + name, other, + "hosts sharing a truncated prefix must stay distinct via the digest" ); } } diff --git a/crates/trusted-server-adapter-fastly/src/platform.rs b/crates/trusted-server-adapter-fastly/src/platform.rs index c89508d36..e38e40f58 100644 --- a/crates/trusted-server-adapter-fastly/src/platform.rs +++ b/crates/trusted-server-adapter-fastly/src/platform.rs @@ -163,25 +163,60 @@ fn backend_config_from_spec(spec: &PlatformBackendSpec) -> BackendConfig<'_> { /// [`FastlyPlatformBackend::canonicalize_transport_timeout_ms`]). const TRANSPORT_TIMEOUT_QUANTUM_MS: u32 = 250; +/// Upper bound of the fine-grained quantum range. +/// +/// Budget-bound values below this ceiling are floored to a +/// [`TRANSPORT_TIMEOUT_QUANTUM_MS`] multiple (the issue #847 behavior for the +/// default 2000 ms auction). At or above it, values snap to the coarse +/// [`TRANSPORT_TIMEOUT_COARSE_LADDER_MS`] instead so the total number of +/// distinct budget-derived buckets stays globally bounded regardless of how +/// large the configured ceiling is. +const TRANSPORT_TIMEOUT_QUANTUM_CEILING_MS: u32 = 2000; + /// Coarse rungs for budget-bound transport timeouts below one quantum, /// ordered high to low. /// -/// A budget-bound value at or above one quantum is floored to a -/// [`TRANSPORT_TIMEOUT_QUANTUM_MS`] multiple. Below one quantum, passing the -/// exact wall-clock remainder through would mint a distinct backend name for -/// every millisecond in `1..250`, so the near-exhausted tail alone could -/// exceed Fastly's per-service dynamic backend limit. Snapping to this finite -/// ladder instead bounds the number of budget-derived names an origin can -/// produce. Budgets below the smallest rung round to zero, which callers treat -/// as "budget exhausted — skip the launch". +/// Below one quantum, passing the exact wall-clock remainder through would mint +/// a distinct backend name for every millisecond in `1..250`, so the +/// near-exhausted tail alone could exceed Fastly's per-service dynamic backend +/// limit. Snapping to this finite ladder instead bounds the number of +/// budget-derived names an origin can produce. Budgets below the smallest rung +/// round to zero, which callers treat as "budget exhausted — skip the launch". const SUB_QUANTUM_LADDER_MS: [u32; 4] = [200, 150, 100, 50]; -/// Round a budget-bound transport timeout down to a stable bucket. +/// Coarse rungs for budget-bound transport timeouts at or above the quantum +/// ceiling, ascending. Every rung is a [`TRANSPORT_TIMEOUT_QUANTUM_MS`] +/// multiple. /// -/// At or above one quantum, floors to a [`TRANSPORT_TIMEOUT_QUANTUM_MS`] -/// multiple. Below one quantum, snaps down to the greatest -/// [`SUB_QUANTUM_LADDER_MS`] rung no larger than `remaining_ms` (or zero). +/// Above [`TRANSPORT_TIMEOUT_QUANTUM_CEILING_MS`], flooring to a 250 ms multiple +/// would let a large configured ceiling (e.g. 60,000 ms) mint hundreds of +/// distinct backend names — recreating the per-service dynamic backend +/// exhaustion this quantization exists to prevent. This fixed, globally finite +/// ladder caps the number of high-budget buckets instead: values are floored to +/// the greatest rung no larger than the remaining budget, and anything above +/// the top rung clamps to it. Rounding down never extends a transport cap past +/// the remaining budget. +const TRANSPORT_TIMEOUT_COARSE_LADDER_MS: [u32; 8] = + [2000, 3000, 5000, 10000, 20000, 30000, 45000, 60000]; + +/// Round a budget-bound transport timeout down to a stable, globally bounded +/// bucket. +/// +/// - At or above [`TRANSPORT_TIMEOUT_QUANTUM_CEILING_MS`], floors to the +/// greatest [`TRANSPORT_TIMEOUT_COARSE_LADDER_MS`] rung no larger than +/// `remaining_ms` (clamping to the top rung above it). +/// - Within the quantum range, floors to a [`TRANSPORT_TIMEOUT_QUANTUM_MS`] +/// multiple. +/// - Below one quantum, snaps down to the greatest [`SUB_QUANTUM_LADDER_MS`] +/// rung no larger than `remaining_ms` (or zero). fn quantize_transport_timeout_ms(remaining_ms: u32) -> u32 { + if remaining_ms >= TRANSPORT_TIMEOUT_QUANTUM_CEILING_MS { + return TRANSPORT_TIMEOUT_COARSE_LADDER_MS + .into_iter() + .rev() + .find(|&rung| rung <= remaining_ms) + .unwrap_or(TRANSPORT_TIMEOUT_QUANTUM_CEILING_MS); + } let floored = (remaining_ms / TRANSPORT_TIMEOUT_QUANTUM_MS) * TRANSPORT_TIMEOUT_QUANTUM_MS; if floored > 0 { return floored; @@ -1100,6 +1135,42 @@ mod tests { ); } + #[test] + fn canonicalize_budget_derived_names_stay_bounded_for_large_ceiling() { + // A large configured ceiling (e.g. a 60s mediator budget) must not let + // the budget-derived buckets grow with the ceiling. Without the coarse + // ladder a 60,000ms ceiling would mint ~240 distinct 250ms buckets and + // blow past Fastly's documented per-service dynamic backend limit (200). + let backend = FastlyPlatformBackend; + let configured = 60_000; + let mut distinct = std::collections::BTreeSet::new(); + for remaining in 0..=configured { + let value = backend.canonicalize_transport_timeout_ms(remaining, configured); + if value > 0 { + distinct.insert(value); + } + assert!( + value == 0 + || value % TRANSPORT_TIMEOUT_QUANTUM_MS == 0 + || SUB_QUANTUM_LADDER_MS.contains(&value), + "canonical value {value}ms (from remaining {remaining}ms) is neither a quantum \ + multiple nor a ladder rung" + ); + } + // A budget above the top coarse rung must clamp to it, not open a new + // bucket per 250ms step. + assert_eq!( + backend.canonicalize_transport_timeout_ms(120_000, 240_000), + 60_000, + "a budget above the top coarse rung clamps to it" + ); + assert!( + distinct.len() <= 20, + "large-ceiling budget-derived values must stay bounded, got {} distinct values: {distinct:?}", + distinct.len() + ); + } + // --- FastlyPlatformBackend::predict_name discriminator ------------------ #[test] diff --git a/crates/trusted-server-core/src/auction/orchestrator.rs b/crates/trusted-server-core/src/auction/orchestrator.rs index 02a87cdb5..16fc6ee57 100644 --- a/crates/trusted-server-core/src/auction/orchestrator.rs +++ b/crates/trusted-server-core/src/auction/orchestrator.rs @@ -1,7 +1,7 @@ //! Auction orchestrator for managing multi-provider auctions. use error_stack::{Report, ResultExt}; -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::sync::Arc; use std::time::Duration; use web_time::Instant; @@ -194,6 +194,21 @@ impl AuctionOrchestrator { return Ok(()); } + // A provider listed twice would launch the same auction request twice + // (its backend name canonicalizes identically), so the duplicate is + // detected only after the second outbound send has already fired. Reject + // it at startup instead. + let mut seen = HashSet::new(); + for provider_name in &self.config.providers { + if !seen.insert(provider_name.as_str()) { + return Err(Report::new(TrustedServerError::Configuration { + message: format!( + "Auction provider `{provider_name}` is listed more than once in [auction].providers; each provider may appear at most once" + ), + })); + } + } + for provider_name in self .config .providers @@ -500,6 +515,23 @@ impl AuctionOrchestrator { } }; + // Pre-launch guard: `request_bids` fires the outbound send, and + // discarding the returned pending handle afterwards does not retract + // it. If another provider this auction already claimed the predicted + // backend name, skip *before* dispatching so a duplicate never hits + // the wire. The post-launch check below stays as a defense for a + // provider that resolves to an unexpected name. + if backend_to_provider.contains_key(&backend_name) { + log::warn!( + "Provider '{}' predicted backend name '{}' already claimed by another provider \ + this auction; skipping launch before dispatch to avoid a duplicate request", + provider.provider_name(), + backend_name, + ); + responses.push(provider_launch_failed_response(provider.provider_name(), 0)); + continue; + } + let provider_context = AuctionContext { settings: context.settings, request: context.request, @@ -530,12 +562,11 @@ impl AuctionOrchestrator { ); backend_name.clone() }); - // Responses are correlated back to providers by backend - // name. If another provider this auction already claimed - // this name (e.g. two providers on one origin whose specs - // canonicalize to the same backend), inserting here would - // silently overwrite the first mapping and misattribute or - // drop a response. Fail this launch attributably instead. + // Post-launch defense: the resolved name differs from the + // prediction and collides with another provider's. Responses + // are correlated by backend name, so inserting here would + // overwrite the first mapping and misattribute or drop a + // response. Fail this launch attributably instead. if backend_to_provider.contains_key(&request_backend_name) { let response_time_ms = start_time.elapsed().as_millis() as u64; log::warn!( @@ -946,6 +977,21 @@ impl AuctionOrchestrator { } }; + // Pre-launch guard: skip before `request_bids` fires the outbound + // send when another provider this auction already claimed the + // predicted backend name (see the parallel path). Dropping the + // pending handle afterwards would not retract the request. + if backend_to_provider.contains_key(&backend_name) { + log::warn!( + "Provider '{}' predicted backend name '{}' already claimed by another provider \ + this auction; skipping dispatch before send to avoid a duplicate request", + provider.provider_name(), + backend_name, + ); + launch_responses.push(provider_launch_failed_response(provider.provider_name(), 0)); + continue; + } + let provider_context = AuctionContext { settings: context.settings, request: context.request, @@ -957,7 +1003,7 @@ impl AuctionOrchestrator { let start_time = Instant::now(); match provider.request_bids(request, &provider_context).await { Ok(pending) => { - // See the parallel path: a backend name already claimed by + // Post-launch defense: a backend name already claimed by // another provider this auction would misattribute the // collected response, so fail this launch attributably // rather than overwrite the mapping. @@ -2004,6 +2050,28 @@ mod tests { }); } + #[test] + fn rejects_duplicate_configured_providers() { + // A provider listed twice canonicalizes to one backend name, so the + // duplicate would only be caught after its second outbound request had + // already fired. Startup validation must reject it up front. + let config = AuctionConfig { + enabled: true, + providers: vec!["prebid".to_string(), "prebid".to_string()], + timeout_ms: 2000, + ..Default::default() + }; + let orchestrator = AuctionOrchestrator::new(config); + + let err = orchestrator + .validate_configured_provider_names() + .expect_err("should reject a provider listed more than once"); + assert!( + err.to_string().contains("listed more than once"), + "should explain the duplicate provider, got: {err}" + ); + } + #[test] fn test_orchestrator_is_enabled() { let config = AuctionConfig { From 77f35211626eae864061eded782792adab1a8368 Mon Sep 17 00:00:00 2001 From: prk-Jr Date: Thu, 16 Jul 2026 20:14:47 +0530 Subject: [PATCH 4/6] Update predict_name tests for digest-based backend names The backend name now carries a SHA-256 digest suffix, so the three platform predict_name tests that pinned the exact/ending name string no longer hold. Assert the readable body via starts_with/contains instead. These were missed earlier because the local run filtered the platform suite too narrowly; CI aborted on the first panic. --- .../src/platform.rs | 18 ++++++++++-------- 1 file changed, 10 insertions(+), 8 deletions(-) diff --git a/crates/trusted-server-adapter-fastly/src/platform.rs b/crates/trusted-server-adapter-fastly/src/platform.rs index a2c09a138..bcc18ae5d 100644 --- a/crates/trusted-server-adapter-fastly/src/platform.rs +++ b/crates/trusted-server-adapter-fastly/src/platform.rs @@ -733,9 +733,9 @@ mod tests { .predict_name(&spec) .expect("should compute backend name for valid spec"); - assert_eq!( - name, "backend_https_origin_example_com_443_fb15000_bb15000", - "should match BackendConfig naming convention" + assert!( + name.starts_with("backend_https_origin_example_com_443_fb15000_bb15000_"), + "should match BackendConfig naming convention, got {name}" ); } @@ -757,9 +757,11 @@ mod tests { .predict_name(&spec) .expect("should compute backend name for host header override"); - assert_eq!( - name, "backend_https_origin_example_com_443_oh_www_example_com_fb15000_bb15000", - "should match BackendConfig naming convention with host header override" + assert!( + name.starts_with( + "backend_https_origin_example_com_443_oh_www_example_com_fb15000_bb15000_" + ), + "should match BackendConfig naming convention with host header override, got {name}" ); } @@ -825,8 +827,8 @@ mod tests { .expect("should compute name with custom timeout"); assert!( - name.ends_with("_fb2000_bb2000"), - "should encode 2000ms first-byte and between-bytes timeouts in name" + name.contains("_fb2000_bb2000_"), + "should encode 2000ms first-byte and between-bytes timeouts in name, got {name}" ); } From bfde031c036d87713a5cd5958e95d7458efa41e6 Mon Sep 17 00:00:00 2001 From: prk-Jr Date: Fri, 17 Jul 2026 11:56:55 +0530 Subject: [PATCH 5/6] Address fourth-round review feedback on backend naming and quantization - Reject a mediator that is also listed in [auction].providers at startup; the overlap would fire the same provider twice per auction and its demand already flows through the mediation response - Assert in both cardinality enumeration tests that canonical values never exceed the remaining budget (the mediator hold bound invariant) - Document the accepted worst-case rounding haircut on the coarse ladder - Write the spec digest hex into one pre-sized buffer instead of allocating per byte, and defer the ensure() doc to compute_name --- .../src/backend.rs | 20 +++++----- .../src/platform.rs | 19 ++++++++++ .../src/auction/orchestrator.rs | 37 +++++++++++++++++++ 3 files changed, 66 insertions(+), 10 deletions(-) diff --git a/crates/trusted-server-adapter-fastly/src/backend.rs b/crates/trusted-server-adapter-fastly/src/backend.rs index e812b95bd..f2ff5d9e5 100644 --- a/crates/trusted-server-adapter-fastly/src/backend.rs +++ b/crates/trusted-server-adapter-fastly/src/backend.rs @@ -1,3 +1,4 @@ +use core::fmt::Write as _; use std::time::Duration; use error_stack::{Report, ResultExt as _}; @@ -69,11 +70,11 @@ fn spec_digest_hex(canonical: &str) -> String { let mut hasher = Sha256::new(); hasher.update(canonical.as_bytes()); let digest = hasher.finalize(); - digest - .iter() - .take(SPEC_DIGEST_HEX_LEN / 2) - .map(|byte| format!("{byte:02x}")) - .collect() + let mut hex = String::with_capacity(SPEC_DIGEST_HEX_LEN); + for byte in digest.iter().take(SPEC_DIGEST_HEX_LEN / 2) { + write!(hex, "{byte:02x}").expect("should write hex digit to string"); + } + hex } /// Default first-byte timeout for backends (15 seconds). @@ -327,11 +328,10 @@ impl<'a> BackendConfig<'a> { /// Ensure a dynamic backend exists for this configuration and return its name. /// - /// The backend name is derived from the scheme, host, port, certificate - /// setting, `first_byte_timeout`, and `between_bytes_timeout` to avoid - /// collisions. Different timeout values produce different backend - /// registrations so that a tight deadline cannot be silently widened by an - /// earlier registration. + /// The name is a collision-resistant function of the complete backend spec + /// (see `Self::compute_name`), so different specs — for example, different + /// timeout values — always produce different backend registrations and a + /// tight deadline cannot be silently widened by an earlier registration. /// /// # Errors /// diff --git a/crates/trusted-server-adapter-fastly/src/platform.rs b/crates/trusted-server-adapter-fastly/src/platform.rs index bcc18ae5d..ffa58122f 100644 --- a/crates/trusted-server-adapter-fastly/src/platform.rs +++ b/crates/trusted-server-adapter-fastly/src/platform.rs @@ -196,6 +196,13 @@ const SUB_QUANTUM_LADDER_MS: [u32; 4] = [200, 150, 100, 50]; /// the greatest rung no larger than the remaining budget, and anything above /// the top rung clamps to it. Rounding down never extends a transport cap past /// the remaining budget. +/// +/// The rung spacing trades transport window for cardinality: just below a rung +/// the haircut approaches the gap to the rung beneath (worst case ~50%, e.g. a +/// remaining budget of 9,999 ms snaps to 5,000 ms). This is accepted — on the +/// mediator path this value is the effective bound, but a denser ladder would +/// buy back at most half a bucket of transport time at the cost of +/// proportionally more backend names. const TRANSPORT_TIMEOUT_COARSE_LADDER_MS: [u32; 8] = [2000, 3000, 5000, 10000, 20000, 30000, 45000, 60000]; @@ -1126,6 +1133,12 @@ mod tests { "canonical value {value}ms (from remaining {remaining}ms) is neither a quantum \ multiple nor a ladder rung" ); + // The mediator hold bound relies on canonicalization never + // extending a transport cap past the wall-clock budget. + assert!( + value <= remaining, + "canonical value {value}ms must not extend past the remaining {remaining}ms budget" + ); } assert!( distinct.len() <= 16, @@ -1156,6 +1169,12 @@ mod tests { "canonical value {value}ms (from remaining {remaining}ms) is neither a quantum \ multiple nor a ladder rung" ); + // The mediator hold bound relies on canonicalization never + // extending a transport cap past the wall-clock budget. + assert!( + value <= remaining, + "canonical value {value}ms must not extend past the remaining {remaining}ms budget" + ); } // A budget above the top coarse rung must clamp to it, not open a new // bucket per 250ms step. diff --git a/crates/trusted-server-core/src/auction/orchestrator.rs b/crates/trusted-server-core/src/auction/orchestrator.rs index 6c137af11..9331ceef7 100644 --- a/crates/trusted-server-core/src/auction/orchestrator.rs +++ b/crates/trusted-server-core/src/auction/orchestrator.rs @@ -209,6 +209,20 @@ impl AuctionOrchestrator { } } + // A provider that is also the mediator would be called twice per + // auction — once in the bidding phase and again as the mediator. The + // mediator's own demand already flows through its mediation response, + // so the overlap is never a legitimate configuration. + if let Some(mediator_name) = &self.config.mediator + && seen.contains(mediator_name.as_str()) + { + return Err(Report::new(TrustedServerError::Configuration { + message: format!( + "Auction mediator `{mediator_name}` is also listed in [auction].providers; a provider may not mediate its own auction" + ), + })); + } + for provider_name in self .config .providers @@ -2074,6 +2088,29 @@ mod tests { ); } + #[test] + fn rejects_mediator_also_listed_as_provider() { + // A provider acting as its own mediator would be called twice per + // auction (bidding phase, then mediation); its demand already flows + // through the mediation response, so reject the overlap at startup. + let config = AuctionConfig { + enabled: true, + providers: vec!["prebid".to_string()], + mediator: Some("prebid".to_string()), + timeout_ms: 2000, + ..Default::default() + }; + let orchestrator = AuctionOrchestrator::new(config); + + let err = orchestrator + .validate_configured_provider_names() + .expect_err("should reject a mediator that is also a provider"); + assert!( + err.to_string().contains("may not mediate its own auction"), + "should explain the mediator/provider overlap, got: {err}" + ); + } + #[test] fn test_orchestrator_is_enabled() { let config = AuctionConfig { From 7af1b5f262870c166fc3e156ec69e306206a733c Mon Sep 17 00:00:00 2001 From: prk-Jr Date: Wed, 22 Jul 2026 13:35:31 +0530 Subject: [PATCH 6/6] Correlate dispatched auction bids by resolved backend name The dispatched-auction post-launch guard re-checked the predicted backend name the pre-launch guard had already proved absent, never reading pending.backend_name(). Because collection correlates by the name the response reports (Fastly's get_backend_name()), keying the map on the prediction could drop a response whose resolved name diverged, and the guard could never catch a resolved-name collision. Derive request_backend_name from pending.backend_name() (falling back to the prediction only when unset), then check, insert, and preserve the pending under that resolved name, mirroring run_providers_parallel. Add regression tests via a DivergentBackendProvider whose prediction differs from its resolved name: one asserts a diverging response still correlates, one asserts a post-launch resolved-name collision fails the second provider attributably. --- .../src/auction/orchestrator.rs | 253 +++++++++++++++++- 1 file changed, 244 insertions(+), 9 deletions(-) diff --git a/crates/trusted-server-core/src/auction/orchestrator.rs b/crates/trusted-server-core/src/auction/orchestrator.rs index e9b5e2669..bf9ecad7b 100644 --- a/crates/trusted-server-core/src/auction/orchestrator.rs +++ b/crates/trusted-server-core/src/auction/orchestrator.rs @@ -1022,17 +1022,35 @@ impl AuctionOrchestrator { let start_time = Instant::now(); match provider.request_bids(request, &provider_context).await { Ok(pending) => { - // Post-launch defense: a backend name already claimed by - // another provider this auction would misattribute the - // collected response, so fail this launch attributably - // rather than overwrite the mapping. - if backend_to_provider.contains_key(&backend_name) { + // Correlate on the *resolved* backend name the pending + // request carries, falling back to the prediction only when + // the adapter left it unset. Collection keys on the name the + // response reports (Fastly's `get_backend_name()`), so keying + // the map on the prediction here would drop a response whose + // resolved name diverged. Mirrors run_providers_parallel. + let request_backend_name = pending + .backend_name() + .map(str::to_string) + .unwrap_or_else(|| { + log::warn!( + "Provider '{}' pending request returned no backend name; \ + using predicted name '{}'", + provider.provider_name(), + backend_name, + ); + backend_name.clone() + }); + // Post-launch defense: a resolved backend name already claimed + // by another provider this auction would misattribute the + // collected response, so fail this launch attributably rather + // than overwrite the mapping. + if backend_to_provider.contains_key(&request_backend_name) { let response_time_ms = start_time.elapsed().as_millis() as u64; log::warn!( "Provider '{}' resolved to backend name '{}' already claimed by another \ provider this auction; skipping dispatch to avoid response misattribution", provider.provider_name(), - backend_name, + request_backend_name, ); launch_responses.push(provider_launch_failed_response( provider.provider_name(), @@ -1042,18 +1060,18 @@ impl AuctionOrchestrator { log::info!( "Dispatching bid request to '{}' (backend: {}, budget: {}ms)", provider.provider_name(), - backend_name, + request_backend_name, effective_timeout ); backend_to_provider.insert( - backend_name.clone(), + request_backend_name.clone(), ( provider.provider_name().to_string(), start_time, Arc::clone(provider), ), ); - pending_requests.push(pending.with_backend_name(backend_name)); + pending_requests.push(pending.with_backend_name(request_backend_name)); } } Err(e) => { @@ -1594,6 +1612,77 @@ mod tests { } } + /// Provider whose `backend_name` prediction deliberately differs from the + /// backend name its `request_bids` puts on the wire (the resolved name the + /// pending request carries and collection correlates by). Models an adapter + /// that resolves a name unequal to the orchestrator's prediction, exercising + /// the dispatched path's resolved-name correlation and post-launch guard. + struct DivergentBackendProvider { + name: &'static str, + predicted: &'static str, + resolved: &'static str, + } + + impl DivergentBackendProvider { + fn new(name: &'static str, predicted: &'static str, resolved: &'static str) -> Self { + Self { + name, + predicted, + resolved, + } + } + } + + #[async_trait::async_trait(?Send)] + impl AuctionProvider for DivergentBackendProvider { + fn provider_name(&self) -> &'static str { + self.name + } + + async fn request_bids( + &self, + _request: &AuctionRequest, + context: &AuctionContext<'_>, + ) -> Result> { + let req = PlatformHttpRequest::new( + http::Request::builder() + .method("POST") + .uri("https://example.com/bid") + .body(edgezero_core::body::Body::empty()) + .expect("should build stub bid request"), + self.resolved, + ); + context + .services + .http_client() + .send_async(req) + .await + .change_context(TrustedServerError::Auction { + message: "stub launch failed".to_string(), + }) + } + + async fn parse_response( + &self, + _response: PlatformResponse, + response_time_ms: u64, + ) -> Result> { + Ok(AuctionResponse::success( + self.name, + vec![], + response_time_ms, + )) + } + + fn timeout_ms(&self) -> u32 { + 2000 + } + + fn backend_name(&self, _services: &RuntimeServices, _timeout_ms: u32) -> Option { + Some(self.predicted.to_string()) + } + } + /// Mediator whose context-aware parse restores `nurl`/`ad_id` (mirroring /// `adserver_mock`), while its context-free parse does not. Lets a test prove /// the synchronous mediation path calls `parse_response_with_context`. @@ -2680,6 +2769,152 @@ mod tests { }); } + #[test] + fn dispatched_resolved_backend_name_diverging_from_prediction_still_correlates() { + futures::executor::block_on(async { + // The dispatched path must correlate on the backend name the pending + // request resolves to, not the orchestrator's prediction. A provider + // whose resolved name differs from its prediction would otherwise land + // its response in the "unknown backend" branch and drop the bid. + let stub = Arc::new(StubHttpClient::new()); + stub.push_response(200, b"{}".to_vec()); + let services = build_services_with_http_client(stub); + // SAFETY: `Box::leak` creates a `'static` reference for test use only. + // The leaked allocation is bounded to the test process lifetime. + let services: &'static RuntimeServices = Box::leak(Box::new(services)); + + let config = AuctionConfig { + enabled: true, + providers: vec!["provider-a".to_string()], + timeout_ms: 2000, + mediator: None, + ..Default::default() + }; + let mut orchestrator = AuctionOrchestrator::new(config); + orchestrator.register_provider(Arc::new(DivergentBackendProvider::new( + "provider-a", + "predicted-backend", + "resolved-backend", + ))); + + let request = create_test_auction_request(); + let settings = create_test_settings(); + let req = http::Request::builder() + .method(http::Method::GET) + .uri("https://example.com/test") + .body(edgezero_core::body::Body::empty()) + .expect("should build request"); + let context = AuctionContext { + settings: &settings, + request: &req, + timeout_ms: 2000, + provider_responses: None, + services, + }; + + let dispatched = match orchestrator.dispatch_auction(&request, &context).await { + DispatchAuctionOutcome::Dispatched(dispatched) => dispatched, + _ => panic!("should dispatch the diverging provider"), + }; + let result = orchestrator + .collect_dispatched_auction(dispatched, services, &context) + .await; + + let provider_a = result + .provider_responses + .iter() + .find(|r| r.provider == "provider-a") + .expect("should have provider-a response"); + assert_eq!( + provider_a.status, + BidStatus::Success, + "response should correlate by the resolved backend name, not the prediction" + ); + }); + } + + #[test] + fn dispatched_post_launch_resolved_name_collision_fails_second_provider_attributably() { + futures::executor::block_on(async { + // Two providers with *distinct* predictions that resolve to the SAME + // backend name. The pre-launch guard keys on the predictions and lets + // both through; only the post-launch guard, reading the resolved name, + // can catch the collision. The second must fail attributably. + let stub = Arc::new(StubHttpClient::new()); + stub.push_response(200, b"{}".to_vec()); // provider-a send_async + stub.push_response(200, b"{}".to_vec()); // provider-b send_async (dropped after guard) + let services = build_services_with_http_client(stub); + // SAFETY: `Box::leak` creates a `'static` reference for test use only. + // The leaked allocation is bounded to the test process lifetime. + let services: &'static RuntimeServices = Box::leak(Box::new(services)); + + let config = AuctionConfig { + enabled: true, + providers: vec!["provider-a".to_string(), "provider-b".to_string()], + timeout_ms: 2000, + mediator: None, + ..Default::default() + }; + let mut orchestrator = AuctionOrchestrator::new(config); + orchestrator.register_provider(Arc::new(DivergentBackendProvider::new( + "provider-a", + "predicted-a", + "shared-resolved", + ))); + orchestrator.register_provider(Arc::new(DivergentBackendProvider::new( + "provider-b", + "predicted-b", + "shared-resolved", + ))); + + let request = create_test_auction_request(); + let settings = create_test_settings(); + let req = http::Request::builder() + .method(http::Method::GET) + .uri("https://example.com/test") + .body(edgezero_core::body::Body::empty()) + .expect("should build request"); + let context = AuctionContext { + settings: &settings, + request: &req, + timeout_ms: 2000, + provider_responses: None, + services, + }; + + let dispatched = match orchestrator.dispatch_auction(&request, &context).await { + DispatchAuctionOutcome::Dispatched(dispatched) => dispatched, + _ => { + panic!("should dispatch the first provider despite the resolved-name collision") + } + }; + let result = orchestrator + .collect_dispatched_auction(dispatched, services, &context) + .await; + + let provider_a = result + .provider_responses + .iter() + .find(|r| r.provider == "provider-a") + .expect("should have provider-a response"); + let provider_b = result + .provider_responses + .iter() + .find(|r| r.provider == "provider-b") + .expect("should have provider-b response"); + assert_eq!( + provider_a.status, + BidStatus::Success, + "the first provider on the shared resolved name should launch and succeed" + ); + assert_eq!( + provider_b.status, + BidStatus::Error, + "the second provider colliding on the resolved name should fail attributably" + ); + }); + } + #[test] fn select_error_is_attributed_to_correct_provider() { futures::executor::block_on(async {