diff --git a/crates/trusted-server-adapter-fastly/src/app.rs b/crates/trusted-server-adapter-fastly/src/app.rs index e56498b1..12b2fd16 100644 --- a/crates/trusted-server-adapter-fastly/src/app.rs +++ b/crates/trusted-server-adapter-fastly/src/app.rs @@ -407,6 +407,7 @@ fn build_ec_request_state( match EcContext::read_from_request_with_geo(settings, req, services, geo_info.as_ref()) { Ok(mut context) => { context.set_device_signals(device_signals); + context.set_recovery_eligible(is_real_browser && is_navigation_request(req)); (context, None) } Err(report) => (EcContext::default(), Some(report)), diff --git a/crates/trusted-server-adapter-fastly/src/main.rs b/crates/trusted-server-adapter-fastly/src/main.rs index d20de533..fd8c60ae 100644 --- a/crates/trusted-server-adapter-fastly/src/main.rs +++ b/crates/trusted-server-adapter-fastly/src/main.rs @@ -205,9 +205,9 @@ fn edgezero_main(mut req: FastlyRequest) { policy.apply_after_route_finalization(&mut response); } - if let Some(ec_state) = ec_state { + if let Some(mut ec_state) = ec_state { if let Some(settings) = settings_snapshot.as_deref() { - match apply_edgezero_ec_finalize(settings, &ec_state, &mut response) { + match apply_edgezero_ec_finalize(settings, &mut ec_state, &mut response) { Ok(partner_registry) => { send_edgezero_response(response, request_filter_effects.as_ref()); run_edgezero_pull_sync_after_send(settings, &partner_registry, &ec_state); @@ -222,7 +222,7 @@ fn edgezero_main(mut req: FastlyRequest) { } else { match load_settings_from_config_store() { Ok(settings) => { - match apply_edgezero_ec_finalize(&settings, &ec_state, &mut response) { + match apply_edgezero_ec_finalize(&settings, &mut ec_state, &mut response) { Ok(partner_registry) => { send_edgezero_response(response, request_filter_effects.as_ref()); run_edgezero_pull_sync_after_send( @@ -286,7 +286,7 @@ fn apply_entry_point_finalize_headers( fn apply_edgezero_ec_finalize( settings: &Settings, - ec_state: &EcFinalizeState, + ec_state: &mut EcFinalizeState, response: &mut HttpResponse, ) -> Result> { let partner_registry = PartnerRegistry::from_config(&settings.ec.partners)?; @@ -297,7 +297,7 @@ fn apply_edgezero_ec_finalize( }; ec_finalize_response( settings, - &ec_state.ec_context, + &mut ec_state.ec_context, finalize_kv_graph.as_ref(), &partner_registry, ec_state.eids_cookie.as_deref(), diff --git a/crates/trusted-server-core/src/auction/endpoints.rs b/crates/trusted-server-core/src/auction/endpoints.rs index 83c7df34..d665afdc 100644 --- a/crates/trusted-server-core/src/auction/endpoints.rs +++ b/crates/trusted-server-core/src/auction/endpoints.rs @@ -14,10 +14,9 @@ use crate::constants::COOKIE_TS_EIDS; use crate::ec::eids::{resolve_partner_ids, to_eids}; use crate::ec::kv::KvIdentityGraph; use crate::ec::kv_types::MAX_UID_LENGTH; -use crate::ec::log_id; use crate::ec::prebid_eids::parse_prebid_eids_cookie; use crate::ec::registry::PartnerRegistry; -use crate::ec::EcContext; +use crate::ec::{EcContext, EcKvSnapshot}; use crate::error::TrustedServerError; use crate::openrtb::{Eid, Uid}; use crate::platform::RuntimeServices; @@ -247,7 +246,11 @@ pub async fn handle_auction( // Resolve partner EIDs from the KV identity graph when the user has // a valid EC and both KV and partner stores are available. - let eids = resolve_auction_eids(kv, registry, ec_context); + let auction_kv_snapshot = match (kv, ec_id) { + (Some(graph), Some(ec_id)) => graph.load_snapshot(ec_id), + _ => EcKvSnapshot::NotRead, + }; + let eids = resolve_auction_eids(&auction_kv_snapshot, registry, ec_context); // Look up geo for device info. let geo = services @@ -345,11 +348,10 @@ pub async fn handle_auction( /// store, no EC, consent denied). On KV or partner-resolution errors, logs a /// warning and returns empty EIDs so the auction can proceed in degraded mode. pub(crate) fn resolve_auction_eids( - kv: Option<&KvIdentityGraph>, + snapshot: &EcKvSnapshot, registry: Option<&PartnerRegistry>, ec_context: &EcContext, ) -> Option> { - let kv = kv?; let registry = registry?; if !ec_context.ec_allowed() { @@ -358,19 +360,15 @@ pub(crate) fn resolve_auction_eids( let ec_id = ec_context.ec_value()?; - let entry = match kv.get(ec_id) { - Ok(Some((entry, _generation))) => entry, - Ok(None) => return Some(Vec::new()), - Err(err) => { - log::warn!( - "Auction KV read failed for EC ID '{}': {err:?}", - log_id(ec_id) - ); - return Some(Vec::new()); - } + let Some(entry) = snapshot.entry_for(ec_id) else { + return Some(Vec::new()); }; - let resolved = resolve_partner_ids(registry, &entry); + if !entry.consent.ok { + return Some(Vec::new()); + } + + let resolved = resolve_partner_ids(registry, entry); Some(to_eids(&resolved)) } @@ -858,22 +856,24 @@ mod tests { } #[test] - fn resolve_auction_eids_returns_none_without_kv() { + fn resolve_auction_eids_returns_empty_without_snapshot() { let registry = PartnerRegistry::empty(); let ec_id = format!("{}.ABC123", "a".repeat(64)); let ec_context = make_ec_context(Jurisdiction::NonRegulated, Some(&ec_id)); - let result = resolve_auction_eids(None, Some(®istry), &ec_context); - assert!(result.is_none(), "should return None when KV is missing"); + let result = resolve_auction_eids(&EcKvSnapshot::NotRead, Some(®istry), &ec_context); + assert!( + result.is_some_and(|eids| eids.is_empty()), + "should degrade to empty EIDs without a snapshot" + ); } #[test] fn resolve_auction_eids_returns_none_without_registry() { - let kv = KvIdentityGraph::failing("test_store"); let ec_id = format!("{}.ABC123", "a".repeat(64)); let ec_context = make_ec_context(Jurisdiction::NonRegulated, Some(&ec_id)); - let result = resolve_auction_eids(Some(&kv), None, &ec_context); + let result = resolve_auction_eids(&EcKvSnapshot::NotRead, None, &ec_context); assert!( result.is_none(), "should return None when registry is missing" @@ -882,12 +882,11 @@ mod tests { #[test] fn resolve_auction_eids_returns_none_when_consent_denied() { - let kv = KvIdentityGraph::failing("test_store"); let registry = PartnerRegistry::empty(); let ec_id = format!("{}.ABC123", "a".repeat(64)); let ec_context = make_ec_context(Jurisdiction::Unknown, Some(&ec_id)); - let result = resolve_auction_eids(Some(&kv), Some(®istry), &ec_context); + let result = resolve_auction_eids(&EcKvSnapshot::NotRead, Some(®istry), &ec_context); assert!( result.is_none(), "should return None when consent is denied" @@ -896,11 +895,10 @@ mod tests { #[test] fn resolve_auction_eids_returns_none_when_no_ec() { - let kv = KvIdentityGraph::failing("test_store"); let registry = PartnerRegistry::empty(); let ec_context = make_ec_context(Jurisdiction::NonRegulated, None); - let result = resolve_auction_eids(Some(&kv), Some(®istry), &ec_context); + let result = resolve_auction_eids(&EcKvSnapshot::NotRead, Some(®istry), &ec_context); assert!( result.is_none(), "should return None when no EC value is present" @@ -909,14 +907,14 @@ mod tests { #[test] fn resolve_auction_eids_returns_empty_on_kv_miss() { - let kv = KvIdentityGraph::failing("nonexistent_store"); let registry = PartnerRegistry::empty(); let ec_id = format!("{}.ABC123", "a".repeat(64)); let ec_context = make_ec_context(Jurisdiction::NonRegulated, Some(&ec_id)); - // KV store doesn't exist, so the get() call will error — should return - // empty Vec (degraded mode), not None. - let result = resolve_auction_eids(Some(&kv), Some(®istry), &ec_context); + let snapshot = EcKvSnapshot::Failed { + ec_id: ec_id.clone(), + }; + let result = resolve_auction_eids(&snapshot, Some(®istry), &ec_context); let eids = result.expect("should return Some on KV error (degraded mode)"); assert!( eids.is_empty(), diff --git a/crates/trusted-server-core/src/ec/finalize.rs b/crates/trusted-server-core/src/ec/finalize.rs index 2c10d2df..8c72f978 100644 --- a/crates/trusted-server-core/src/ec/finalize.rs +++ b/crates/trusted-server-core/src/ec/finalize.rs @@ -12,12 +12,12 @@ use super::consent::{ec_consent_granted, ec_consent_withdrawn}; use crate::settings::Settings; use super::cookies::{expire_ec_cookie, set_ec_cookie}; -use super::generation::is_valid_ec_id; -use super::kv::KvIdentityGraph; -use super::log_id; -use super::prebid_eids::ingest_eid_cookies; +use super::generation::{generate_ec_id, is_valid_ec_id}; +use super::kv::{apply_partner_id_updates, CreateIfAbsentOutcome, KvIdentityGraph}; +use super::kv_types::KvEntry; +use super::prebid_eids::collect_eid_cookie_updates; use super::registry::PartnerRegistry; -use super::EcContext; +use super::{current_timestamp, EcContext, EcKvSnapshot}; /// TS-managed response headers tied to EC identity output. const EC_RESPONSE_HEADERS: &[&str] = &[ @@ -40,7 +40,7 @@ const EC_RESPONSE_HEADERS: &[&str] = &[ /// from the request *before* routing consumes it. pub fn ec_finalize_response( settings: &Settings, - ec_context: &EcContext, + ec_context: &mut EcContext, kv: Option<&KvIdentityGraph>, registry: &PartnerRegistry, eids_cookie: Option<&str>, @@ -69,11 +69,14 @@ pub fn ec_finalize_response( // for subsequent EC behavior. if let Some(graph) = kv { apply_withdrawal_tombstones(&ids_to_withdraw, |ec_id| { - if let Err(err) = graph.write_withdrawal_tombstone(ec_id) { - log::error!( - "Failed to write withdrawal tombstone for EC ID '{}': {err:?}", - log_id(ec_id), - ); + let initial = if ec_context.kv_snapshot().belongs_to(ec_id) { + ec_context.kv_snapshot().clone() + } else { + EcKvSnapshot::NotRead + }; + let outcome = graph.tombstone_existing_from_snapshot(ec_id, initial); + if ec_context.ec_value() == Some(ec_id) { + ec_context.set_kv_snapshot(outcome); } }); } @@ -84,8 +87,19 @@ pub fn ec_finalize_response( // Returning user: consent is granted and EC came from request. if ec_context.ec_was_present() && !ec_context.ec_generated() && consent_allows_ec { - if let (Some(graph), Some(ec_id)) = (kv, ec_context.ec_value()) { - ingest_eid_cookies(eids_cookie, sharedid_cookie, ec_id, graph, registry); + if let (Some(graph), Some(ec_id)) = (kv, ec_context.ec_value().map(str::to_owned)) { + let updates = collect_eid_cookie_updates(eids_cookie, sharedid_cookie, registry); + let snapshot = graph.upsert_partner_ids_from_snapshot( + &ec_id, + &updates, + ec_context.kv_snapshot().clone(), + ); + ec_context.set_kv_snapshot(snapshot); + if matches!(ec_context.kv_snapshot(), EcKvSnapshot::Missing { .. }) + && ec_context.recovery_eligible() + { + recover_orphaned_ec(settings, ec_context, graph, &updates, response); + } } // Ordinary returning-user page views no longer refresh the browser @@ -97,16 +111,95 @@ pub fn ec_finalize_response( // there is no KV graph: that would mint a browser cookie with no backing // identity-graph row, producing a phantom ID on later requests. if ec_context.ec_generated() { - let (Some(graph), Some(ec_id)) = (kv, ec_context.ec_value()) else { + let (Some(graph), Some(ec_id)) = (kv, ec_context.ec_value().map(str::to_owned)) else { log::info!("Skipping generated EC response write because KV graph is unavailable"); return; }; - ingest_eid_cookies(eids_cookie, sharedid_cookie, ec_id, graph, registry); - set_ec_cookie_on_response(settings, ec_context, response); + let updates = collect_eid_cookie_updates(eids_cookie, sharedid_cookie, registry); + let snapshot = graph.upsert_partner_ids_from_snapshot( + &ec_id, + &updates, + ec_context.kv_snapshot().clone(), + ); + ec_context.set_kv_snapshot(snapshot); + if ec_context.kv_snapshot().entry_for(&ec_id).is_some() { + set_ec_cookie_on_response(settings, ec_context, response); + } else { + log::warn!("Skipping generated EC cookie because backing row is not authoritative"); + } } } +fn recover_orphaned_ec( + settings: &Settings, + ec_context: &mut EcContext, + graph: &KvIdentityGraph, + updates: &[super::kv::PartnerIdUpdate], + response: &mut Response, +) { + // Snapshot the orphaned ID once so every fail-closed exit binds the failed + // snapshot to the same key. + let orphan_id = ec_context.ec_value().unwrap_or_default().to_owned(); + let Some(client_ip) = ec_context.client_ip().map(str::to_owned) else { + log::warn!("Orphan EC recovery skipped because client IP is unavailable"); + ec_context.set_kv_snapshot(EcKvSnapshot::Failed { + ec_id: orphan_id.clone(), + }); + return; + }; + + const MAX_RECOVERY_ATTEMPTS: usize = 5; + for _attempt in 0..MAX_RECOVERY_ATTEMPTS { + let ec_id = match generate_ec_id(settings, &client_ip) { + Ok(ec_id) => ec_id, + Err(err) => { + log::warn!("Orphan EC recovery ID generation failed: {err:?}"); + ec_context.set_kv_snapshot(EcKvSnapshot::Failed { + ec_id: orphan_id.clone(), + }); + return; + } + }; + let mut entry = KvEntry::new( + ec_context.consent(), + ec_context.geo_info(), + current_timestamp(), + &settings.publisher.domain, + ); + entry.device = ec_context + .device_signals() + .map(super::device::DeviceSignals::to_kv_device); + apply_partner_id_updates(&mut entry, updates); + + match graph.create_if_absent(&ec_id, &entry) { + Ok(CreateIfAbsentOutcome::Written) => { + let snapshot = EcKvSnapshot::Present { + ec_id: ec_id.clone(), + entry: Box::new(entry), + generation: None, + }; + ec_context.replace_with_generated(ec_id, snapshot); + set_ec_cookie_on_response(settings, ec_context, response); + return; + } + Ok(CreateIfAbsentOutcome::AlreadyExists) => continue, + Err(err) => { + log::warn!("Orphan EC recovery failed: {err:?}"); + ec_context.set_kv_snapshot(EcKvSnapshot::Failed { + ec_id: orphan_id.clone(), + }); + return; + } + } + } + + log::warn!("Orphan EC recovery exhausted collision retries"); + ec_context.set_kv_snapshot(EcKvSnapshot::Failed { + ec_id: orphan_id.clone(), + }); +} + /// Sets the EC cookie on response when an EC ID is available. pub fn set_ec_cookie_on_response( settings: &Settings, @@ -401,7 +494,7 @@ mod tests { source: ConsentSource::Cookie, ..Default::default() }; - let ec_context = + let mut ec_context = make_context_with_consent(Some(&ec_id), Some(&ec_id), true, false, consent); let mut response = empty_response(); set_header(&mut response, "x-ts-ec", "stale"); @@ -413,7 +506,7 @@ mod tests { let test_registry = PartnerRegistry::from_config(&partners).expect("should build registry"); ec_finalize_response( &settings, - &ec_context, + &mut ec_context, None, &test_registry, None, @@ -453,7 +546,7 @@ mod tests { let settings = create_test_settings(); let active_ec = sample_ec_id("activ1"); let cookie_ec = sample_ec_id("cook1e"); - let ec_context = make_context( + let mut ec_context = make_context( Some(&active_ec), Some(&cookie_ec), true, @@ -465,7 +558,7 @@ mod tests { let test_registry = PartnerRegistry::empty(); ec_finalize_response( &settings, - &ec_context, + &mut ec_context, None, &test_registry, None, @@ -487,7 +580,7 @@ mod tests { fn finalize_returning_user_sets_no_header_or_cookie() { let settings = create_test_settings(); let ec_id = sample_ec_id("mtch01"); - let ec_context = make_context( + let mut ec_context = make_context( Some(&ec_id), Some(&ec_id), true, @@ -499,7 +592,7 @@ mod tests { let test_registry = PartnerRegistry::empty(); ec_finalize_response( &settings, - &ec_context, + &mut ec_context, None, &test_registry, None, @@ -521,7 +614,7 @@ mod tests { fn finalize_generated_ec_without_kv_skips_cookie_and_header() { let settings = create_test_settings(); let generated_ec = sample_ec_id("gen123"); - let ec_context = make_context( + let mut ec_context = make_context( Some(&generated_ec), None, false, @@ -533,7 +626,7 @@ mod tests { let test_registry = PartnerRegistry::empty(); ec_finalize_response( &settings, - &ec_context, + &mut ec_context, None, &test_registry, None, @@ -551,16 +644,95 @@ mod tests { ); } + #[test] + fn finalize_rotates_orphaned_cookie_to_new_backed_ec() { + let settings = create_test_settings(); + let orphaned_ec = sample_ec_id("orphn1"); + let consent = ConsentContext { + jurisdiction: Jurisdiction::NonRegulated, + source: ConsentSource::Cookie, + ..Default::default() + }; + let mut ec_context = EcContext::new_for_test_with_ip( + Some(orphaned_ec.clone()), + consent, + Some("192.0.2.10".to_owned()), + ); + ec_context.set_recovery_eligible(true); + ec_context.set_kv_snapshot(EcKvSnapshot::Missing { + ec_id: orphaned_ec.clone(), + }); + let graph = KvIdentityGraph::in_memory("test_store"); + let mut response = empty_response(); + + ec_finalize_response( + &settings, + &mut ec_context, + Some(&graph), + &PartnerRegistry::empty(), + None, + None, + &mut response, + ); + + let replacement = ec_context.ec_value().expect("should rotate orphan"); + assert_ne!(replacement, orphaned_ec); + assert!( + graph + .get(replacement) + .expect("should read replacement") + .is_some(), + "replacement cookie should have a backing row" + ); + assert!( + get_header(&response, "set-cookie").is_some(), + "should emit replacement cookie after persistence" + ); + } + + #[test] + fn finalize_generated_ec_does_not_emit_cookie_for_authoritative_missing_row() { + let settings = create_test_settings(); + let generated_ec = sample_ec_id("genmis"); + let mut ec_context = make_context( + Some(&generated_ec), + None, + false, + true, + Jurisdiction::NonRegulated, + ); + ec_context.set_kv_snapshot(EcKvSnapshot::Missing { + ec_id: generated_ec, + }); + let graph = KvIdentityGraph::in_memory("test_store"); + let mut response = empty_response(); + + ec_finalize_response( + &settings, + &mut ec_context, + Some(&graph), + &PartnerRegistry::empty(), + None, + None, + &mut response, + ); + + assert!( + get_header(&response, "set-cookie").is_none(), + "must not emit a cookie without an authoritative backing row" + ); + } + #[test] fn finalize_denied_without_cookie_is_noop() { let settings = create_test_settings(); - let ec_context = make_context(None, None, false, false, Jurisdiction::Unknown); + let mut ec_context = make_context(None, None, false, false, Jurisdiction::Unknown); let mut response = empty_response(); let test_registry = PartnerRegistry::empty(); ec_finalize_response( &settings, - &ec_context, + &mut ec_context, None, &test_registry, None, @@ -582,7 +754,7 @@ mod tests { fn finalize_unknown_jurisdiction_strips_headers_without_expiring_cookie() { let settings = create_test_settings(); let ec_id = sample_ec_id("unk001"); - let ec_context = make_context( + let mut ec_context = make_context( Some(&ec_id), Some(&ec_id), true, @@ -596,7 +768,7 @@ mod tests { let test_registry = PartnerRegistry::empty(); ec_finalize_response( &settings, - &ec_context, + &mut ec_context, None, &test_registry, None, @@ -617,4 +789,202 @@ mod tests { "should not expire the cookie without an explicit withdrawal signal" ); } + + // ----------------------------------------------------------------------- + // Orphan-recovery gating and two-ID withdrawal + // ----------------------------------------------------------------------- + + fn granting_consent() -> ConsentContext { + ConsentContext { + jurisdiction: Jurisdiction::NonRegulated, + source: ConsentSource::Cookie, + ..Default::default() + } + } + + fn returning_user_context( + orphan: &str, + snapshot: EcKvSnapshot, + recovery_eligible: bool, + ) -> EcContext { + let mut ec = EcContext::new_for_test_with_ip( + Some(orphan.to_owned()), + granting_consent(), + Some("192.0.2.10".to_owned()), + ); + ec.set_recovery_eligible(recovery_eligible); + ec.set_kv_snapshot(snapshot); + ec + } + + fn assert_did_not_rotate(ec_context: &EcContext, orphan: &str, response: &Response) { + assert_eq!( + ec_context.ec_value(), + Some(orphan), + "must not rotate the active EC ID" + ); + assert!(!ec_context.ec_generated(), "must not mark a rotated EC"); + assert!( + get_header(response, "set-cookie").is_none(), + "must not emit a replacement cookie" + ); + } + + #[test] + fn finalize_not_read_snapshot_does_not_rotate() { + let settings = create_test_settings(); + let orphan = sample_ec_id("notrd1"); + let mut ec_context = returning_user_context(&orphan, EcKvSnapshot::NotRead, true); + let graph = KvIdentityGraph::in_memory("test_store"); + let mut response = empty_response(); + + ec_finalize_response( + &settings, + &mut ec_context, + Some(&graph), + &PartnerRegistry::empty(), + None, + None, + &mut response, + ); + + assert_did_not_rotate(&ec_context, &orphan, &response); + } + + #[test] + fn finalize_failed_snapshot_does_not_rotate() { + let settings = create_test_settings(); + let orphan = sample_ec_id("faild1"); + let mut ec_context = returning_user_context( + &orphan, + EcKvSnapshot::Failed { + ec_id: orphan.clone(), + }, + true, + ); + let graph = KvIdentityGraph::in_memory("test_store"); + let mut response = empty_response(); + + ec_finalize_response( + &settings, + &mut ec_context, + Some(&graph), + &PartnerRegistry::empty(), + None, + None, + &mut response, + ); + + assert_did_not_rotate(&ec_context, &orphan, &response); + } + + #[test] + fn finalize_tombstone_snapshot_does_not_rotate() { + let settings = create_test_settings(); + let orphan = sample_ec_id("tomb01"); + let tombstone = EcKvSnapshot::Present { + ec_id: orphan.clone(), + entry: Box::new(KvEntry::tombstone(current_timestamp())), + generation: Some(1), + }; + let mut ec_context = returning_user_context(&orphan, tombstone, true); + let graph = KvIdentityGraph::in_memory("test_store"); + let mut response = empty_response(); + + ec_finalize_response( + &settings, + &mut ec_context, + Some(&graph), + &PartnerRegistry::empty(), + None, + None, + &mut response, + ); + + assert_did_not_rotate(&ec_context, &orphan, &response); + } + + #[test] + fn finalize_subresource_missing_row_does_not_rotate() { + let settings = create_test_settings(); + let orphan = sample_ec_id("subrs1"); + // Missing row, but the request is not a recovery-eligible browser navigation. + let mut ec_context = returning_user_context( + &orphan, + EcKvSnapshot::Missing { + ec_id: orphan.clone(), + }, + false, + ); + let graph = KvIdentityGraph::in_memory("test_store"); + let mut response = empty_response(); + + ec_finalize_response( + &settings, + &mut ec_context, + Some(&graph), + &PartnerRegistry::empty(), + None, + None, + &mut response, + ); + + assert_did_not_rotate(&ec_context, &orphan, &response); + assert!( + graph.get(&orphan).expect("should read store").is_none(), + "a non-eligible request must not create the missing root" + ); + } + + #[test] + fn finalize_withdrawal_tombstones_present_id_and_skips_missing_other() { + let settings = create_test_settings(); + let active_ec = sample_ec_id("activ2"); + let cookie_ec = sample_ec_id("cook2e"); + let consent = ConsentContext { + jurisdiction: Jurisdiction::UsState("CA".to_owned()), + gpc: true, + source: ConsentSource::Cookie, + ..Default::default() + }; + let mut ec_context = + make_context_with_consent(Some(&active_ec), Some(&cookie_ec), true, false, consent); + // Carry a snapshot only for the active ID; the other ID must be looked up + // independently and never created if absent. + let graph = KvIdentityGraph::in_memory("test_store"); + graph + .create(&active_ec, &live_entry()) + .expect("should seed active row"); + ec_context.set_kv_snapshot(graph.load_snapshot(&active_ec)); + let mut response = empty_response(); + + ec_finalize_response( + &settings, + &mut ec_context, + Some(&graph), + &PartnerRegistry::empty(), + None, + None, + &mut response, + ); + + let (active_stored, _) = graph + .get(&active_ec) + .expect("should read active row") + .expect("active row should remain as a tombstone"); + assert!( + !active_stored.consent.ok, + "the present active ID should be tombstoned via its carried snapshot" + ); + assert!( + graph.get(&cookie_ec).expect("should read store").is_none(), + "a missing second ID must never be created by withdrawal" + ); + } + + fn live_entry() -> KvEntry { + let mut entry = KvEntry::tombstone(1000); + entry.consent.ok = true; + entry + } } diff --git a/crates/trusted-server-core/src/ec/kv.rs b/crates/trusted-server-core/src/ec/kv.rs index d6e3b6f4..1c14ce9a 100644 --- a/crates/trusted-server-core/src/ec/kv.rs +++ b/crates/trusted-server-core/src/ec/kv.rs @@ -23,7 +23,7 @@ use super::current_timestamp; use super::generation::ec_hash; use super::kv_backend::{EcKvStore, EcKvWrite, EcKvWriteMode, EcKvWriteOutcome}; use super::kv_types::{KvEntry, KvMetadata, KvNetwork}; -use super::log_id; +use super::{log_id, EcKvSnapshot}; /// Maximum number of CAS retry attempts before giving up. const MAX_CAS_RETRIES: u32 = 5; @@ -56,6 +56,15 @@ pub enum UpsertResult { Unchanged, } +/// Outcome of atomically creating an identity-graph root when absent. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum CreateIfAbsentOutcome { + /// The candidate entry was persisted. + Written, + /// A row already exists for the candidate key. + AlreadyExists, +} + /// Partner UID update to apply to a KV identity graph entry. #[derive(Debug, Clone, PartialEq, Eq)] pub(crate) struct PartnerIdUpdate { @@ -75,7 +84,7 @@ impl PartnerIdUpdate { } } -fn apply_partner_id_updates(entry: &mut KvEntry, updates: &[PartnerIdUpdate]) -> bool { +pub(crate) fn apply_partner_id_updates(entry: &mut KvEntry, updates: &[PartnerIdUpdate]) -> bool { let mut latest_updates = BTreeMap::new(); for update in updates { latest_updates.insert(update.partner_id.as_str(), update.uid.as_str()); @@ -186,6 +195,30 @@ impl KvIdentityGraph { Ok(Some((entry, lookup.generation))) } + /// Loads one request-scoped snapshot, preserving miss versus failure at the caller boundary. + #[must_use] + pub fn load_snapshot(&self, ec_id: &str) -> EcKvSnapshot { + match self.get(ec_id) { + Ok(Some((entry, generation))) => EcKvSnapshot::Present { + ec_id: ec_id.to_owned(), + entry: Box::new(entry), + generation: Some(generation), + }, + Ok(None) => EcKvSnapshot::Missing { + ec_id: ec_id.to_owned(), + }, + Err(err) => { + log::warn!( + "EC KV snapshot read failed for '{}': {err:?}", + log_id(ec_id) + ); + EcKvSnapshot::Failed { + ec_id: ec_id.to_owned(), + } + } + } + } + fn deserialize_entry( store_name: &str, ec_id: &str, @@ -254,6 +287,23 @@ impl KvIdentityGraph { } } + /// Atomically creates an entry while preserving collision as normal control flow. + /// + /// # Errors + /// + /// Returns [`TrustedServerError::KvStore`] when serialization or store I/O fails. + pub fn create_if_absent( + &self, + ec_id: &str, + entry: &KvEntry, + ) -> Result> { + let (body, meta_str) = Self::serialize_entry(entry, self.store_name())?; + match self.write_entry(ec_id, &body, &meta_str, ENTRY_TTL, EcKvWriteMode::Add)? { + EcKvWriteOutcome::Written => Ok(CreateIfAbsentOutcome::Written), + EcKvWriteOutcome::PreconditionFailed => Ok(CreateIfAbsentOutcome::AlreadyExists), + } + } + /// Low-level write with shared error context. fn write_entry( &self, @@ -445,6 +495,86 @@ impl KvIdentityGraph { ))) } + /// Merges partner IDs using request-scoped persisted state as the first CAS input. + pub(crate) fn upsert_partner_ids_from_snapshot( + &self, + ec_id: &str, + updates: &[PartnerIdUpdate], + snapshot: EcKvSnapshot, + ) -> EcKvSnapshot { + if updates.is_empty() { + return snapshot; + } + + let mut current = snapshot; + for _attempt in 0..MAX_CAS_RETRIES { + let (mut entry, generation) = match current { + EcKvSnapshot::Present { + ec_id: ref snapshot_id, + ref entry, + generation: Some(generation), + } if snapshot_id == ec_id => (entry.as_ref().clone(), generation), + EcKvSnapshot::Present { .. } | EcKvSnapshot::NotRead => { + current = self.load_snapshot(ec_id); + continue; + } + EcKvSnapshot::Missing { + ec_id: ref snapshot_id, + } + | EcKvSnapshot::Failed { + ec_id: ref snapshot_id, + } if snapshot_id == ec_id => return current, + EcKvSnapshot::Missing { .. } | EcKvSnapshot::Failed { .. } => { + current = self.load_snapshot(ec_id); + continue; + } + }; + + if !entry.consent.ok { + return current; + } + if !apply_partner_id_updates(&mut entry, updates) { + return current; + } + let Ok((body, meta_str)) = Self::serialize_entry(&entry, self.store_name()) else { + return EcKvSnapshot::Failed { + ec_id: ec_id.to_owned(), + }; + }; + match self.write_entry( + ec_id, + &body, + &meta_str, + ENTRY_TTL, + EcKvWriteMode::IfGenerationMatch(generation), + ) { + Ok(EcKvWriteOutcome::Written) => { + return EcKvSnapshot::Present { + ec_id: ec_id.to_owned(), + entry: Box::new(entry), + generation: None, + }; + } + Ok(EcKvWriteOutcome::PreconditionFailed) => { + current = self.load_snapshot(ec_id); + } + Err(err) => { + log::warn!( + "snapshot partner upsert failed for '{}': {err:?}", + log_id(ec_id) + ); + return EcKvSnapshot::Failed { + ec_id: ec_id.to_owned(), + }; + } + } + } + + EcKvSnapshot::Failed { + ec_id: ec_id.to_owned(), + } + } + /// Atomically merges a partner ID into the existing entry. /// /// Uses CAS (generation markers) to avoid clobbering concurrent writes @@ -638,6 +768,70 @@ impl KvIdentityGraph { } } + /// Writes a tombstone only when an authoritative row already exists. + pub(crate) fn tombstone_existing_from_snapshot( + &self, + ec_id: &str, + snapshot: EcKvSnapshot, + ) -> EcKvSnapshot { + let mut current = snapshot; + for _attempt in 0..MAX_CAS_RETRIES { + let generation = match current { + EcKvSnapshot::Present { + ec_id: ref snapshot_id, + generation: Some(generation), + .. + } if snapshot_id == ec_id => generation, + EcKvSnapshot::Missing { + ec_id: ref snapshot_id, + } if snapshot_id == ec_id => return current, + EcKvSnapshot::Failed { + ec_id: ref snapshot_id, + } if snapshot_id == ec_id => return current, + _ => { + current = self.load_snapshot(ec_id); + continue; + } + }; + let tombstone = KvEntry::tombstone(current_timestamp()); + let Ok((body, meta_str)) = Self::serialize_entry(&tombstone, self.store_name()) else { + return EcKvSnapshot::Failed { + ec_id: ec_id.to_owned(), + }; + }; + match self.write_entry( + ec_id, + &body, + &meta_str, + TOMBSTONE_TTL, + EcKvWriteMode::IfGenerationMatch(generation), + ) { + Ok(EcKvWriteOutcome::Written) => { + return EcKvSnapshot::Present { + ec_id: ec_id.to_owned(), + entry: Box::new(tombstone), + generation: None, + }; + } + Ok(EcKvWriteOutcome::PreconditionFailed) => { + current = self.load_snapshot(ec_id); + } + Err(err) => { + log::warn!( + "conditional withdrawal tombstone failed for '{}': {err:?}", + log_id(ec_id) + ); + return EcKvSnapshot::Failed { + ec_id: ec_id.to_owned(), + }; + } + } + } + EcKvSnapshot::Failed { + ec_id: ec_id.to_owned(), + } + } + /// Counts the number of keys sharing the same EC hash prefix. /// /// Uses the platform KV list API with a prefix filter, limited to @@ -1181,6 +1375,34 @@ mod tests { ); } + #[test] + fn create_if_absent_reports_written_and_collision() { + let kv = KvIdentityGraph::in_memory("test_store"); + let ec_id = format!("{}.ABC123", "a".repeat(64)); + + assert_eq!( + kv.create_if_absent(&ec_id, &live_entry()) + .expect("should create absent entry"), + CreateIfAbsentOutcome::Written + ); + assert_eq!( + kv.create_if_absent(&ec_id, &live_entry()) + .expect("should report collision"), + CreateIfAbsentOutcome::AlreadyExists + ); + } + + #[test] + fn create_if_absent_propagates_store_error() { + let kv = KvIdentityGraph::failing("test_store"); + let ec_id = format!("{}.ABC123", "a".repeat(64)); + + assert!( + kv.create_if_absent(&ec_id, &live_entry()).is_err(), + "should preserve store failures instead of reporting a collision" + ); + } + #[test] fn create_or_revive_revives_tombstone() { let kv = KvIdentityGraph::in_memory("test_store"); @@ -1239,6 +1461,48 @@ mod tests { assert_eq!(result, UpsertResult::ConsentWithdrawn); } + #[test] + fn snapshot_bulk_upsert_returns_persisted_entry_without_stale_generation() { + let kv = KvIdentityGraph::in_memory("test_store"); + let ec_id = format!("{}.ABC123", "a".repeat(64)); + kv.create(&ec_id, &live_entry()).expect("should create"); + let snapshot = kv.load_snapshot(&ec_id); + let updates = [PartnerIdUpdate::new("ssp_x", "uid-1")]; + + let outcome = kv.upsert_partner_ids_from_snapshot(&ec_id, &updates, snapshot); + + let entry = outcome + .entry_for(&ec_id) + .expect("should retain persisted entry"); + assert_eq!( + entry.ids.get("ssp_x").map(|id| id.uid.as_str()), + Some("uid-1") + ); + assert_eq!( + outcome.generation_for(&ec_id), + None, + "backend does not return the post-write generation" + ); + } + + #[test] + fn snapshot_bulk_upsert_does_not_create_missing_root() { + let kv = KvIdentityGraph::in_memory("test_store"); + let ec_id = format!("{}.ABC123", "a".repeat(64)); + let updates = [PartnerIdUpdate::new("ssp_x", "uid-1")]; + + let outcome = kv.upsert_partner_ids_from_snapshot( + &ec_id, + &updates, + EcKvSnapshot::Missing { + ec_id: ec_id.clone(), + }, + ); + + assert!(matches!(outcome, EcKvSnapshot::Missing { .. })); + assert!(kv.get(&ec_id).expect("should read store").is_none()); + } + #[test] fn write_withdrawal_tombstone_overwrites_live_entry() { let kv = KvIdentityGraph::in_memory("test_store"); @@ -1254,4 +1518,403 @@ mod tests { .expect("should find tombstone entry"); assert!(!loaded.consent.ok, "should be withdrawn after tombstone"); } + + #[test] + fn tombstone_existing_from_snapshot_never_creates_missing_key() { + let kv = KvIdentityGraph::in_memory("test_store"); + let ec_id = format!("{}.ABC123", "a".repeat(64)); + let snapshot = EcKvSnapshot::Missing { + ec_id: ec_id.clone(), + }; + + let outcome = kv.tombstone_existing_from_snapshot(&ec_id, snapshot); + + assert!(matches!(outcome, EcKvSnapshot::Missing { .. })); + assert!( + kv.get(&ec_id).expect("should read store").is_none(), + "withdrawal must not create a tombstone for an absent key" + ); + } + + #[test] + fn tombstone_existing_from_snapshot_uses_existing_generation() { + let kv = KvIdentityGraph::in_memory("test_store"); + let ec_id = format!("{}.ABC123", "a".repeat(64)); + kv.create(&ec_id, &live_entry()).expect("should create"); + let snapshot = kv.load_snapshot(&ec_id); + + let outcome = kv.tombstone_existing_from_snapshot(&ec_id, snapshot); + + assert!( + outcome + .entry_for(&ec_id) + .is_some_and(|entry| !entry.consent.ok), + "should return the persisted tombstone" + ); + let (stored, _) = kv + .get(&ec_id) + .expect("should read store") + .expect("should preserve existing key"); + assert!(!stored.consent.ok, "should persist withdrawal state"); + } + + // ----------------------------------------------------------------------- + // Snapshot-aware mutation stores and tests + // ----------------------------------------------------------------------- + + /// [`EcKvStore`] wrapper that counts `lookup` calls through a shared counter + /// so tests can prove exactly how many reads a mutation performs. + struct CountingEcKv { + inner: InMemoryEcKv, + lookups: std::sync::Arc, + } + + impl CountingEcKv { + fn new(lookups: std::sync::Arc) -> Self { + Self { + inner: InMemoryEcKv::new("counting-store"), + lookups, + } + } + } + + impl EcKvStore for CountingEcKv { + fn store_name(&self) -> &str { + self.inner.store_name() + } + fn lookup(&self, key: &str) -> Result, Report> { + self.lookups + .fetch_add(1, std::sync::atomic::Ordering::Relaxed); + self.inner.lookup(key) + } + fn insert( + &self, + key: &str, + write: EcKvWrite<'_>, + ) -> Result> { + self.inner.insert(key, write) + } + fn count_keys_with_prefix( + &self, + prefix: &str, + limit: u32, + ) -> Result> { + self.inner.count_keys_with_prefix(prefix, limit) + } + fn delete(&self, key: &str) -> Result<(), Report> { + self.inner.delete(key) + } + } + + /// [`EcKvStore`] whose reads succeed but every write fails, simulating a + /// store that becomes unwritable mid-request. + struct WriteFailingEcKv { + inner: InMemoryEcKv, + } + + impl WriteFailingEcKv { + fn new() -> Self { + Self { + inner: InMemoryEcKv::new("write-failing-store"), + } + } + } + + impl EcKvStore for WriteFailingEcKv { + fn store_name(&self) -> &str { + self.inner.store_name() + } + fn lookup(&self, key: &str) -> Result, Report> { + self.inner.lookup(key) + } + fn insert( + &self, + _key: &str, + _write: EcKvWrite<'_>, + ) -> Result> { + Err(Report::new(TrustedServerError::KvStore { + store_name: self.inner.store_name().to_owned(), + message: "write failing test store".to_owned(), + })) + } + fn count_keys_with_prefix( + &self, + prefix: &str, + limit: u32, + ) -> Result> { + self.inner.count_keys_with_prefix(prefix, limit) + } + fn delete(&self, key: &str) -> Result<(), Report> { + self.inner.delete(key) + } + } + + /// [`EcKvStore`] wrapper whose first CAS write both fails the precondition + /// and deletes the key, simulating a concurrent withdrawal that removes the + /// row between this writer's read and its write. + struct DisappearOnConflictEcKv { + inner: InMemoryEcKv, + conflicts_remaining: std::sync::Mutex, + } + + impl DisappearOnConflictEcKv { + fn new(conflicts: u32) -> Self { + Self { + inner: InMemoryEcKv::new("disappear-store"), + conflicts_remaining: std::sync::Mutex::new(conflicts), + } + } + fn seed_live(&self, ec_id: &str) { + let (body, meta) = + KvIdentityGraph::serialize_entry(&live_entry(), self.inner.store_name()) + .expect("should serialize seeded entry"); + self.inner + .insert( + ec_id, + EcKvWrite { + body: &body, + metadata: &meta, + ttl: ENTRY_TTL, + mode: EcKvWriteMode::Add, + }, + ) + .expect("should seed live entry"); + } + } + + impl EcKvStore for DisappearOnConflictEcKv { + fn store_name(&self) -> &str { + self.inner.store_name() + } + fn lookup(&self, key: &str) -> Result, Report> { + self.inner.lookup(key) + } + fn insert( + &self, + key: &str, + write: EcKvWrite<'_>, + ) -> Result> { + if matches!(write.mode, EcKvWriteMode::IfGenerationMatch(_)) { + let mut remaining = self + .conflicts_remaining + .lock() + .expect("should lock conflict counter"); + if *remaining > 0 { + *remaining -= 1; + self.inner.delete(key).expect("should delete on conflict"); + return Ok(EcKvWriteOutcome::PreconditionFailed); + } + } + self.inner.insert(key, write) + } + fn count_keys_with_prefix( + &self, + prefix: &str, + limit: u32, + ) -> Result> { + self.inner.count_keys_with_prefix(prefix, limit) + } + fn delete(&self, key: &str) -> Result<(), Report> { + self.inner.delete(key) + } + } + + fn snapshot_ec_id() -> String { + format!("{}.ABC123", "a".repeat(64)) + } + + #[test] + fn snapshot_upsert_with_generation_writes_without_reading() { + let lookups = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let graph = KvIdentityGraph::new(CountingEcKv::new(lookups.clone())); + let ec_id = snapshot_ec_id(); + graph.create(&ec_id, &live_entry()).expect("should seed"); + let snapshot = EcKvSnapshot::Present { + ec_id: ec_id.clone(), + entry: Box::new(live_entry()), + generation: Some(1), + }; + let updates = [PartnerIdUpdate::new("ssp_x", "uid-1")]; + + let outcome = graph.upsert_partner_ids_from_snapshot(&ec_id, &updates, snapshot); + + assert_eq!( + lookups.load(std::sync::atomic::Ordering::Relaxed), + 0, + "a usable generation must avoid the initial read" + ); + assert_eq!( + outcome + .entry_for(&ec_id) + .and_then(|entry| entry.ids.get("ssp_x")) + .map(|id| id.uid.as_str()), + Some("uid-1") + ); + assert_eq!(outcome.generation_for(&ec_id), None); + } + + #[test] + fn snapshot_upsert_unchanged_updates_preserve_generation() { + let kv = KvIdentityGraph::in_memory("test_store"); + let ec_id = snapshot_ec_id(); + let mut seeded = live_entry(); + apply_partner_id_updates(&mut seeded, &[PartnerIdUpdate::new("ssp_x", "uid-1")]); + kv.create(&ec_id, &seeded).expect("should seed"); + let snapshot = kv.load_snapshot(&ec_id); + assert_eq!(snapshot.generation_for(&ec_id), Some(1)); + + let outcome = kv.upsert_partner_ids_from_snapshot( + &ec_id, + &[PartnerIdUpdate::new("ssp_x", "uid-1")], + snapshot, + ); + + assert_eq!( + outcome.generation_for(&ec_id), + Some(1), + "an unchanged merge preserves the usable generation and performs no write" + ); + } + + #[test] + fn snapshot_upsert_refreshes_unavailable_generation_exactly_once() { + let lookups = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let graph = KvIdentityGraph::new(CountingEcKv::new(lookups.clone())); + let ec_id = snapshot_ec_id(); + graph.create(&ec_id, &live_entry()).expect("should seed"); + // Finalize-written style snapshot: entry known, generation unavailable. + let snapshot = EcKvSnapshot::Present { + ec_id: ec_id.clone(), + entry: Box::new(live_entry()), + generation: None, + }; + let updates = [PartnerIdUpdate::new("ssp_x", "uid-1")]; + + let outcome = graph.upsert_partner_ids_from_snapshot(&ec_id, &updates, snapshot); + + assert_eq!( + lookups.load(std::sync::atomic::Ordering::Relaxed), + 1, + "an unavailable generation refreshes exactly once before CAS" + ); + assert!(outcome + .entry_for(&ec_id) + .is_some_and(|e| e.ids.contains_key("ssp_x"))); + } + + #[test] + fn snapshot_upsert_cas_conflict_remerges_concurrent_data() { + let graph = KvIdentityGraph::new(ConflictInjectingEcKv::new(1, true)); + let ec_id = snapshot_ec_id(); + graph.create(&ec_id, &live_entry()).expect("should seed"); + let snapshot = graph.load_snapshot(&ec_id); + let updates = [PartnerIdUpdate::new("ssp_x", "uid-1")]; + + let outcome = graph.upsert_partner_ids_from_snapshot(&ec_id, &updates, snapshot); + + let entry = outcome + .entry_for(&ec_id) + .expect("should persist re-merged entry"); + assert_eq!( + entry.ids.get("ssp_x").map(|id| id.uid.as_str()), + Some("uid-1"), + "conflict must re-merge our update onto the concurrently revived row" + ); + assert!(entry.consent.ok, "concurrent revive keeps the row live"); + } + + #[test] + fn snapshot_upsert_rejects_tombstone() { + let kv = KvIdentityGraph::in_memory("test_store"); + let ec_id = snapshot_ec_id(); + kv.create(&ec_id, &KvEntry::tombstone(1000)) + .expect("should seed tombstone"); + let snapshot = kv.load_snapshot(&ec_id); + let updates = [PartnerIdUpdate::new("ssp_x", "uid-1")]; + + let outcome = kv.upsert_partner_ids_from_snapshot(&ec_id, &updates, snapshot); + + assert!( + outcome + .entry_for(&ec_id) + .is_some_and(|entry| entry.ids.is_empty()), + "a tombstone must reject partner enrichment" + ); + let (stored, _) = kv + .get(&ec_id) + .expect("should read store") + .expect("tombstone should remain"); + assert!(stored.ids.is_empty(), "no update should reach the store"); + } + + #[test] + fn snapshot_upsert_store_failure_returns_failed_not_request_local() { + let graph = KvIdentityGraph::new(WriteFailingEcKv::new()); + let ec_id = snapshot_ec_id(); + let snapshot = EcKvSnapshot::Present { + ec_id: ec_id.clone(), + entry: Box::new(live_entry()), + generation: Some(1), + }; + let updates = [PartnerIdUpdate::new("ssp_x", "uid-1")]; + + let outcome = graph.upsert_partner_ids_from_snapshot(&ec_id, &updates, snapshot); + + assert!( + matches!(outcome, EcKvSnapshot::Failed { .. }), + "a store write failure must not claim request-local IDs were persisted" + ); + } + + #[test] + fn tombstone_existing_from_snapshot_retries_cas_conflict() { + let graph = KvIdentityGraph::new(ConflictInjectingEcKv::new(1, false)); + let ec_id = snapshot_ec_id(); + graph.create(&ec_id, &live_entry()).expect("should seed"); + let snapshot = graph.load_snapshot(&ec_id); + + let outcome = graph.tombstone_existing_from_snapshot(&ec_id, snapshot); + + assert!( + outcome + .entry_for(&ec_id) + .is_some_and(|entry| !entry.consent.ok), + "should retry the conflict and persist the tombstone" + ); + } + + #[test] + fn tombstone_existing_from_snapshot_store_failure_returns_failed() { + let graph = KvIdentityGraph::new(WriteFailingEcKv::new()); + let ec_id = snapshot_ec_id(); + let snapshot = EcKvSnapshot::Present { + ec_id: ec_id.clone(), + entry: Box::new(live_entry()), + generation: Some(1), + }; + + let outcome = graph.tombstone_existing_from_snapshot(&ec_id, snapshot); + + assert!(matches!(outcome, EcKvSnapshot::Failed { .. })); + } + + #[test] + fn tombstone_existing_from_snapshot_noop_when_row_disappears_on_retry() { + let store = DisappearOnConflictEcKv::new(1); + store.seed_live(&snapshot_ec_id()); + let graph = KvIdentityGraph::new(store); + let ec_id = snapshot_ec_id(); + let snapshot = graph.load_snapshot(&ec_id); + + let outcome = graph.tombstone_existing_from_snapshot(&ec_id, snapshot); + + assert!( + matches!(outcome, EcKvSnapshot::Missing { .. }), + "a row that disappears during retry becomes a no-op" + ); + assert!( + graph.get(&ec_id).expect("should read store").is_none(), + "must not recreate the disappeared key" + ); + } } diff --git a/crates/trusted-server-core/src/ec/mod.rs b/crates/trusted-server-core/src/ec/mod.rs index 408ea9b3..4423eb3e 100644 --- a/crates/trusted-server-core/src/ec/mod.rs +++ b/crates/trusted-server-core/src/ec/mod.rs @@ -74,9 +74,71 @@ use crate::platform::RuntimeServices; use crate::settings::Settings; use device::DeviceSignals; -use self::kv::KvIdentityGraph; +use self::kv::{CreateIfAbsentOutcome, KvIdentityGraph}; use self::kv_types::KvEntry; +/// Request-scoped view of one EC identity-graph lookup. +/// +/// The state distinguishes an authoritative miss from a store failure and +/// binds persisted entry data to the EC ID that was actually read or written. +#[derive(Debug, Clone, Default, PartialEq)] +pub enum EcKvSnapshot { + /// No identity-graph lookup has been attempted for this request. + #[default] + NotRead, + /// The store authoritatively reported that this EC ID does not exist. + Missing { ec_id: String }, + /// Persisted entry data, optionally with a generation usable for CAS. + Present { + ec_id: String, + entry: Box, + generation: Option, + }, + /// The lookup failed, so absence is not authoritative. + Failed { ec_id: String }, +} + +impl EcKvSnapshot { + /// Returns whether this state was produced for `ec_id`. + #[must_use] + pub fn belongs_to(&self, ec_id: &str) -> bool { + match self { + Self::NotRead => false, + Self::Missing { ec_id: snapshot_id } + | Self::Present { + ec_id: snapshot_id, .. + } + | Self::Failed { ec_id: snapshot_id } => snapshot_id == ec_id, + } + } + + /// Returns the persisted entry only when the snapshot belongs to `ec_id`. + #[must_use] + pub fn entry_for(&self, ec_id: &str) -> Option<&KvEntry> { + match self { + Self::Present { + ec_id: snapshot_id, + entry, + .. + } if snapshot_id == ec_id => Some(entry.as_ref()), + _ => None, + } + } + + /// Returns a usable CAS generation only when the snapshot belongs to `ec_id`. + #[must_use] + pub fn generation_for(&self, ec_id: &str) -> Option { + match self { + Self::Present { + ec_id: snapshot_id, + generation, + .. + } if snapshot_id == ec_id => *generation, + _ => None, + } + } +} + pub use generation::{ ec_hash, generate_ec_id, is_valid_ec_hash, is_valid_ec_id, normalize_ec_id_for_kv, }; @@ -160,6 +222,10 @@ pub struct EcContext { /// Set via [`EcContext::set_device_signals`] before /// [`EcContext::generate_if_needed`] is called. device_signals: Option, + /// Request-scoped persisted identity-graph state for the active EC ID. + kv_snapshot: EcKvSnapshot, + /// Whether this request may rotate an orphaned EC identity. + recovery_eligible: bool, } impl EcContext { @@ -239,6 +305,8 @@ impl EcContext { client_ip, geo_info: geo_info.cloned(), device_signals: None, + kv_snapshot: EcKvSnapshot::NotRead, + recovery_eligible: false, }) } @@ -278,12 +346,10 @@ impl EcContext { }) })?; - let ec_id = generation::generate_ec_id(settings, client_ip)?; - log::info!("Generated new EC ID: {}", log_id(&ec_id)); - self.ec_value = Some(ec_id); - self.ec_generated = true; - - if let (Some(graph), Some(ec_value)) = (kv, self.ec_value.as_deref()) { + const MAX_CREATE_ATTEMPTS: usize = 5; + for attempt in 0..MAX_CREATE_ATTEMPTS { + let ec_id = generation::generate_ec_id(settings, client_ip)?; + log::info!("Generated new EC ID: {}", log_id(&ec_id)); let now = current_timestamp(); let mut entry = KvEntry::new( &self.consent, @@ -296,20 +362,45 @@ impl EcContext { .as_ref() .map(DeviceSignals::to_kv_device); - if let Err(err) = graph.create_or_revive(ec_value, &entry) { - log::error!( - "Failed to create or revive EC entry for id '{}' after generation: {err:?}", - log_id(ec_value), - ); - self.ec_value = None; - self.ec_generated = false; - return Err(err.change_context(TrustedServerError::EdgeCookie { - message: "Failed to persist generated EC ID to KV identity graph".to_string(), - })); + if let Some(graph) = kv { + match graph.create_if_absent(&ec_id, &entry) { + Ok(CreateIfAbsentOutcome::Written) => { + self.kv_snapshot = EcKvSnapshot::Present { + ec_id: ec_id.clone(), + entry: Box::new(entry), + generation: None, + }; + } + Ok(CreateIfAbsentOutcome::AlreadyExists) => { + log::warn!( + "Generated EC ID collision on attempt {}/{MAX_CREATE_ATTEMPTS}", + attempt + 1 + ); + continue; + } + Err(err) => { + log::error!( + "Failed to create EC entry for id '{}' after generation: {err:?}", + log_id(&ec_id), + ); + return Err(err.change_context(TrustedServerError::EdgeCookie { + message: "Failed to persist generated EC ID to KV identity graph" + .to_string(), + })); + } + } } + + self.ec_value = Some(ec_id); + self.ec_generated = true; + return Ok(()); } - Ok(()) + Err(Report::new(TrustedServerError::EdgeCookie { + message: format!( + "Failed to allocate a unique EC ID after {MAX_CREATE_ATTEMPTS} attempts" + ), + })) } /// Returns the EC ID value, if present (either from request or generated). @@ -382,6 +473,35 @@ impl EcContext { self.geo_info.as_ref() } + /// Returns the request-scoped identity-graph snapshot. + #[must_use] + pub fn kv_snapshot(&self) -> &EcKvSnapshot { + &self.kv_snapshot + } + + /// Replaces the request-scoped identity-graph snapshot. + pub fn set_kv_snapshot(&mut self, snapshot: EcKvSnapshot) { + self.kv_snapshot = snapshot; + } + + /// Marks a real-browser document navigation as eligible for orphan recovery. + pub fn set_recovery_eligible(&mut self, eligible: bool) { + self.recovery_eligible = eligible; + } + + /// Returns whether orphan recovery is allowed for this request. + #[must_use] + pub fn recovery_eligible(&self) -> bool { + self.recovery_eligible + } + + /// Replaces an orphaned active ID after its new backing row is persisted. + pub(crate) fn replace_with_generated(&mut self, ec_id: String, snapshot: EcKvSnapshot) { + self.ec_value = Some(ec_id); + self.ec_generated = true; + self.kv_snapshot = snapshot; + } + /// Returns whether EC creation is permitted by consent for this request. #[must_use] pub fn ec_allowed(&self) -> bool { @@ -427,6 +547,8 @@ impl EcContext { client_ip: None, geo_info: None, device_signals: None, + kv_snapshot: EcKvSnapshot::NotRead, + recovery_eligible: false, } } @@ -447,6 +569,8 @@ impl EcContext { client_ip, geo_info: None, device_signals: None, + kv_snapshot: EcKvSnapshot::NotRead, + recovery_eligible: false, } } @@ -470,6 +594,8 @@ impl EcContext { client_ip: None, geo_info: None, device_signals: None, + kv_snapshot: EcKvSnapshot::NotRead, + recovery_eligible: false, } } } @@ -493,9 +619,146 @@ pub(crate) fn current_timestamp() -> u64 { #[cfg(test)] mod tests { use super::*; + use crate::consent::jurisdiction::Jurisdiction; + use crate::consent::types::{ConsentContext, ConsentSource}; + use crate::ec::kv_backend::test_support::InMemoryEcKv; + use crate::ec::kv_backend::{ + EcKvLookup, EcKvStore, EcKvWrite, EcKvWriteMode, EcKvWriteOutcome, + }; use crate::platform::test_support::noop_services; use crate::test_support::tests::create_test_settings; + /// [`EcKvStore`] wrapper whose first `collisions` `Add` writes report a + /// precondition failure, forcing generation to retry with a fresh suffix. + struct AddCollidingEcKv { + inner: InMemoryEcKv, + collisions_remaining: std::sync::Mutex, + } + + impl AddCollidingEcKv { + fn new(collisions: u32) -> Self { + Self { + inner: InMemoryEcKv::new("add-colliding-store"), + collisions_remaining: std::sync::Mutex::new(collisions), + } + } + } + + impl EcKvStore for AddCollidingEcKv { + fn store_name(&self) -> &str { + self.inner.store_name() + } + fn lookup(&self, key: &str) -> Result, Report> { + self.inner.lookup(key) + } + fn insert( + &self, + key: &str, + write: EcKvWrite<'_>, + ) -> Result> { + if matches!(write.mode, EcKvWriteMode::Add) { + let mut remaining = self + .collisions_remaining + .lock() + .expect("should lock collision counter"); + if *remaining > 0 { + *remaining -= 1; + return Ok(EcKvWriteOutcome::PreconditionFailed); + } + } + self.inner.insert(key, write) + } + fn count_keys_with_prefix( + &self, + prefix: &str, + limit: u32, + ) -> Result> { + self.inner.count_keys_with_prefix(prefix, limit) + } + fn delete(&self, key: &str) -> Result<(), Report> { + self.inner.delete(key) + } + } + + fn granting_consent() -> ConsentContext { + ConsentContext { + jurisdiction: Jurisdiction::NonRegulated, + source: ConsentSource::Cookie, + ..Default::default() + } + } + + #[test] + fn generate_if_needed_retries_id_collision_then_persists() { + let settings = create_test_settings(); + let mut ec = + EcContext::new_for_test_with_ip(None, granting_consent(), Some("192.0.2.5".to_owned())); + let graph = KvIdentityGraph::new(AddCollidingEcKv::new(2)); + + ec.generate_if_needed(&settings, Some(&graph)) + .expect("should generate after bounded collisions"); + + assert!(ec.ec_value().is_some(), "should allocate a fresh EC ID"); + assert!(ec.ec_generated(), "should mark the EC as generated"); + assert!( + matches!(ec.kv_snapshot(), EcKvSnapshot::Present { .. }), + "generation should seed a present snapshot" + ); + } + + #[test] + fn generate_if_needed_errors_after_collision_exhaustion() { + let settings = create_test_settings(); + let mut ec = + EcContext::new_for_test_with_ip(None, granting_consent(), Some("192.0.2.6".to_owned())); + // Collide on every attempt so the bounded retry is exhausted. + let graph = KvIdentityGraph::new(AddCollidingEcKv::new(u32::MAX)); + + let result = ec.generate_if_needed(&settings, Some(&graph)); + + assert!(result.is_err(), "should fail after exhausting attempts"); + assert!( + ec.ec_value().is_none() && !ec.ec_generated(), + "must not activate an EC ID it could not persist" + ); + } + + #[test] + fn default_ec_context_is_recovery_ineligible_and_unread() { + let ec = EcContext::default(); + assert!( + !ec.recovery_eligible(), + "a default context must not authorize orphan recovery" + ); + assert!( + matches!(ec.kv_snapshot(), EcKvSnapshot::NotRead), + "a default context must carry no identity-graph state" + ); + } + + #[test] + fn read_from_request_does_not_authorize_recovery_from_navigation_headers() { + // Non-Fastly adapters build EC context through the shared read path and + // never call `set_recovery_eligible`. Navigation headers alone must not + // authorize orphan recovery or seed KV state. + let settings = create_test_settings(); + let ec_id = valid_ec_id("b", "CkEc01"); + let cookie = format!("ts-ec={ec_id}"); + let req = create_test_request(&[("cookie", &cookie), ("sec-fetch-dest", "document")]); + + let ec = EcContext::read_from_request(&settings, &req, &noop_services()) + .expect("should read EC context"); + + assert!( + !ec.recovery_eligible(), + "the shared read path must never authorize recovery from headers" + ); + assert!( + matches!(ec.kv_snapshot(), EcKvSnapshot::NotRead), + "the shared read path must leave the snapshot unread" + ); + } + fn create_test_request(headers: &[(&str, &str)]) -> Request { let mut builder = Request::builder().method("GET").uri("http://example.com"); for &(key, value) in headers { @@ -511,6 +774,51 @@ mod tests { format!("{}.{suffix}", prefix_char.repeat(64)) } + #[test] + fn kv_snapshot_distinguishes_non_present_states() { + assert!(EcKvSnapshot::NotRead.entry_for("ec-1").is_none()); + assert!(EcKvSnapshot::Missing { + ec_id: "ec-1".to_owned() + } + .entry_for("ec-1") + .is_none()); + assert!(EcKvSnapshot::Failed { + ec_id: "ec-1".to_owned() + } + .entry_for("ec-1") + .is_none()); + } + + #[test] + fn kv_snapshot_present_state_is_bound_to_ec_id() { + let consent = ConsentContext::default(); + let entry = KvEntry::new(&consent, None, 1_000, "example.com"); + let snapshot = EcKvSnapshot::Present { + ec_id: "ec-1".to_owned(), + entry: Box::new(entry.clone()), + generation: Some(7), + }; + + assert_eq!(snapshot.entry_for("ec-1"), Some(&entry)); + assert_eq!(snapshot.generation_for("ec-1"), Some(7)); + assert!(snapshot.entry_for("ec-2").is_none()); + assert_eq!(snapshot.generation_for("ec-2"), None); + } + + #[test] + fn kv_snapshot_retains_persisted_entry_without_generation() { + let consent = ConsentContext::default(); + let entry = KvEntry::new(&consent, None, 1_000, "example.com"); + let snapshot = EcKvSnapshot::Present { + ec_id: "ec-1".to_owned(), + entry: Box::new(entry.clone()), + generation: None, + }; + + assert_eq!(snapshot.entry_for("ec-1"), Some(&entry)); + assert_eq!(snapshot.generation_for("ec-1"), None); + } + #[test] fn read_from_request_ignores_header_ec() { let settings = create_test_settings(); diff --git a/crates/trusted-server-core/src/ec/prebid_eids.rs b/crates/trusted-server-core/src/ec/prebid_eids.rs index 22620e59..0007408c 100644 --- a/crates/trusted-server-core/src/ec/prebid_eids.rs +++ b/crates/trusted-server-core/src/ec/prebid_eids.rs @@ -119,6 +119,28 @@ pub fn ingest_eid_cookies( ingest_eid_cookies_with_writer(eids_cookie, sharedid_cookie, ec_id, kv, registry); } +/// Collects validated request-local partner updates without performing KV I/O. +pub(crate) fn collect_eid_cookie_updates( + eids_cookie: Option<&str>, + sharedid_cookie: Option<&str>, + registry: &PartnerRegistry, +) -> Vec { + if registry.is_empty() { + return Vec::new(); + } + + let mut updates = Vec::new(); + if let Some(cookie) = eids_cookie { + updates.extend(collect_prebid_eid_updates(cookie, registry)); + } + if let Some(cookie) = sharedid_cookie { + if let Some(update) = collect_sharedid_update(cookie, registry) { + updates.push(update); + } + } + dedupe_partner_updates(updates) +} + /// Parses a `ts-eids` cookie value and writes matched partner UIDs to KV. /// /// `cookie_value` is the raw base64-encoded cookie value, already extracted @@ -142,21 +164,7 @@ fn ingest_eid_cookies_with_writer( writer: &dyn PartnerIdBulkWriter, registry: &PartnerRegistry, ) { - if registry.is_empty() { - return; - } - - let mut updates = Vec::new(); - if let Some(cookie) = eids_cookie { - updates.extend(collect_prebid_eid_updates(cookie, registry)); - } - if let Some(cookie) = sharedid_cookie { - if let Some(update) = collect_sharedid_update(cookie, registry) { - updates.push(update); - } - } - - let updates = dedupe_partner_updates(updates); + let updates = collect_eid_cookie_updates(eids_cookie, sharedid_cookie, registry); if updates.is_empty() { return; } @@ -599,6 +607,39 @@ mod tests { ); } + #[test] + fn collect_eid_cookie_updates_merges_prebid_and_sharedid_without_kv() { + let registry = make_registry(vec![("id5", "id5-sync.com"), ("sharedid", "sharedid.org")]); + let eids_cookie = encode_json(&json!([ + {"source": "id5-sync.com", "uids": [{"id": "ID5_abc", "atype": 1}]} + ])); + + let updates = collect_eid_cookie_updates(Some(&eids_cookie), Some(" shared-1 "), ®istry); + + assert_eq!( + updates.len(), + 2, + "should collect prebid and sharedId matches" + ); + assert!(updates.contains(&PartnerIdUpdate::new("id5-sync.com", "ID5_abc"))); + assert!(updates.contains(&PartnerIdUpdate::new("sharedid.org", "shared-1"))); + } + + #[test] + fn collect_eid_cookie_updates_empty_registry_returns_no_updates() { + let registry = PartnerRegistry::empty(); + let eids_cookie = encode_json(&json!([ + {"source": "id5-sync.com", "uids": [{"id": "ID5_abc", "atype": 1}]} + ])); + + let updates = collect_eid_cookie_updates(Some(&eids_cookie), Some("shared-1"), ®istry); + + assert!( + updates.is_empty(), + "an empty registry matches no partners and touches no KV" + ); + } + #[test] fn dedupe_partner_updates_uses_last_partner_value() { let updates = vec![ diff --git a/crates/trusted-server-core/src/ec/pull_sync.rs b/crates/trusted-server-core/src/ec/pull_sync.rs index fa096d59..89df8fa4 100644 --- a/crates/trusted-server-core/src/ec/pull_sync.rs +++ b/crates/trusted-server-core/src/ec/pull_sync.rs @@ -20,7 +20,7 @@ use crate::platform::{ use crate::settings::Settings; use super::generation::{ec_hash, is_valid_ec_id}; -use super::kv::KvIdentityGraph; +use super::kv::{KvIdentityGraph, PartnerIdUpdate}; use super::kv_types::KvEntry; use super::rate_limiter::RateLimiter; use super::registry::{PartnerConfig, PartnerRegistry}; @@ -28,11 +28,13 @@ use super::registry::{PartnerConfig, PartnerRegistry}; // `current_timestamp` is defined in the parent `ec` module. use super::current_timestamp; use super::EcContext; +use super::EcKvSnapshot; /// Inputs needed to dispatch pull sync after response flush. #[derive(Debug, Clone)] pub struct PullSyncContext { ec_id: String, + snapshot: EcKvSnapshot, } impl PullSyncContext { @@ -69,7 +71,8 @@ pub fn build_pull_sync_context(ec_context: &EcContext) -> Option entry.map(|(entry, _)| entry), - Err(err) => { - log::warn!( - "Pull sync: failed to read identity graph for '{}': {err:?}", - super::log_id(context.ec_id()) - ); - return; - } + let Some(kv_entry) = context.snapshot.entry_for(context.ec_id()) else { + return; }; + if !kv_entry.consent.ok { + return; + } let mut pull_partners = registry.pull_enabled_partners(); @@ -123,9 +122,10 @@ pub fn dispatch_pull_sync( let max_concurrency = settings.ec.pull_sync_concurrency.max(1); let mut in_flight: Vec = Vec::new(); + let mut updates = Vec::new(); for partner in pull_partners { - if !is_partner_pull_eligible(partner, kv_entry.as_ref()) { + if !is_partner_pull_eligible(partner, Some(kv_entry)) { continue; } @@ -213,11 +213,24 @@ pub fn dispatch_pull_sync( }); if in_flight.len() >= max_concurrency { - drain_pull_batch(kv, context.ec_id(), &mut in_flight, services); + drain_pull_batch(&mut in_flight, services, &mut updates); } } - drain_pull_batch(kv, context.ec_id(), &mut in_flight, services); + drain_pull_batch(&mut in_flight, services, &mut updates); + if !updates.is_empty() { + let outcome = kv.upsert_partner_ids_from_snapshot( + context.ec_id(), + &updates, + context.snapshot.clone(), + ); + if matches!(outcome, EcKvSnapshot::Failed { .. }) { + log::warn!( + "Pull sync: failed to persist partner updates for '{}'", + super::log_id(context.ec_id()) + ); + } + } } fn is_partner_pull_eligible(partner: &PartnerConfig, kv_entry: Option<&KvEntry>) -> bool { @@ -286,10 +299,9 @@ fn pull_rate_limit_key(source_domain: &str, ec_id: &str) -> String { } fn drain_pull_batch( - kv: &KvIdentityGraph, - ec_id: &str, in_flight: &mut Vec, services: &RuntimeServices, + updates: &mut Vec, ) { for pending in in_flight.drain(..) { let source_domain = pending.source_domain; @@ -311,13 +323,7 @@ fn drain_pull_batch( continue; }; - if let Err(err) = kv.upsert_partner_id(ec_id, &source_domain, &uid) { - log::warn!( - "Pull sync: failed to upsert partner '{}' for ec_id '{}': {err:?}", - source_domain, - super::log_id(ec_id) - ); - } + updates.push(PartnerIdUpdate::new(source_domain, uid)); } } @@ -710,4 +716,190 @@ mod tests { "hour 1 rotation should move beta to front" ); } + + // ----------------------------------------------------------------------- + // Snapshot-driven eligibility and request-wide aggregation + // ----------------------------------------------------------------------- + + use crate::error::TrustedServerError; + use crate::platform::test_support::{build_services_with_http_client, StubHttpClient}; + use crate::settings::EcPartner; + use crate::test_support::tests::create_test_settings; + use error_stack::Report; + use std::sync::Arc; + + struct AllowAllRateLimiter; + + impl RateLimiter for AllowAllRateLimiter { + fn exceeded( + &self, + _key: &str, + _hourly_limit: u32, + ) -> Result> { + Ok(false) + } + } + + fn pull_enabled_ec_partner(source_domain: &str) -> EcPartner { + EcPartner { + name: format!("Partner {source_domain}"), + source_domain: source_domain.to_owned(), + openrtb_atype: EcPartner::default_openrtb_atype(), + bidstream_enabled: true, + api_token: Redacted::new(format!("{source_domain}-api-token-32-bytes-minimum")), + batch_rate_limit: EcPartner::default_batch_rate_limit(), + pull_sync_enabled: true, + pull_sync_url: Some(format!("https://{source_domain}/sync")), + pull_sync_allowed_domains: vec![source_domain.to_owned()], + pull_sync_ttl_sec: 3600, + pull_sync_rate_limit: 100, + ts_pull_token: Some(Redacted::new("outbound-token".to_owned())), + } + } + + fn snapshot_ec_id() -> String { + format!("{}.ABC123", "a".repeat(64)) + } + + fn seed_present_snapshot(graph: &KvIdentityGraph, ec_id: &str) -> EcKvSnapshot { + let mut entry = KvEntry::tombstone(1000); + entry.consent.ok = true; + graph.create(ec_id, &entry).expect("should seed live entry"); + graph.load_snapshot(ec_id) + } + + #[test] + fn dispatch_pull_sync_aggregates_batches_into_one_bulk_write() { + let mut settings = create_test_settings(); + // Force one partner per concurrency batch so responses span batches. + settings.ec.pull_sync_concurrency = 1; + let registry = PartnerRegistry::from_config(&[ + pull_enabled_ec_partner("alpha.example.com"), + pull_enabled_ec_partner("beta.example.com"), + ]) + .expect("should build pull registry"); + + let graph = KvIdentityGraph::in_memory("pull_store"); + let ec_id = snapshot_ec_id(); + let snapshot = seed_present_snapshot(&graph, &ec_id); + + let stub = Arc::new(StubHttpClient::new()); + // One JSON response per partner, drained across two concurrency batches. + stub.push_response(200, br#"{"uid":"synced-uid"}"#.to_vec()); + stub.push_response(200, br#"{"uid":"synced-uid"}"#.to_vec()); + let services = build_services_with_http_client(stub.clone()); + + let context = PullSyncContext { + ec_id: ec_id.clone(), + snapshot, + }; + dispatch_pull_sync( + &settings, + &graph, + ®istry, + &AllowAllRateLimiter, + &context, + &services, + ); + + let (entry, generation) = graph + .get(&ec_id) + .expect("should read store") + .expect("entry should exist"); + assert_eq!( + entry.ids.get("alpha.example.com").map(|id| id.uid.as_str()), + Some("synced-uid"), + "first partner UID should persist" + ); + assert_eq!( + entry.ids.get("beta.example.com").map(|id| id.uid.as_str()), + Some("synced-uid"), + "second partner UID should persist" + ); + assert_eq!( + generation, 2, + "two partner responses across batches must persist in exactly one bulk write" + ); + } + + #[test] + fn dispatch_pull_sync_skips_non_present_snapshots() { + let mut settings = create_test_settings(); + settings.ec.pull_sync_concurrency = 4; + let registry = + PartnerRegistry::from_config(&[pull_enabled_ec_partner("alpha.example.com")]) + .expect("should build registry"); + let graph = KvIdentityGraph::in_memory("pull_store"); + let ec_id = snapshot_ec_id(); + let stub = Arc::new(StubHttpClient::new()); + let services = build_services_with_http_client(stub.clone()); + + for snapshot in [ + EcKvSnapshot::NotRead, + EcKvSnapshot::Missing { + ec_id: ec_id.clone(), + }, + EcKvSnapshot::Failed { + ec_id: ec_id.clone(), + }, + ] { + let context = PullSyncContext { + ec_id: ec_id.clone(), + snapshot, + }; + dispatch_pull_sync( + &settings, + &graph, + ®istry, + &AllowAllRateLimiter, + &context, + &services, + ); + } + + assert!( + stub.recorded_backend_names().is_empty(), + "not-read, missing, and failed snapshots must not dispatch pull sync" + ); + assert!( + graph.get(&ec_id).expect("should read store").is_none(), + "no snapshot state should create a missing root" + ); + } + + #[test] + fn dispatch_pull_sync_skips_tombstone_snapshot() { + let mut settings = create_test_settings(); + settings.ec.pull_sync_concurrency = 4; + let registry = + PartnerRegistry::from_config(&[pull_enabled_ec_partner("alpha.example.com")]) + .expect("should build registry"); + let graph = KvIdentityGraph::in_memory("pull_store"); + let ec_id = snapshot_ec_id(); + let snapshot = EcKvSnapshot::Present { + ec_id: ec_id.clone(), + entry: Box::new(KvEntry::tombstone(1000)), + generation: Some(1), + }; + let stub = Arc::new(StubHttpClient::new()); + let services = build_services_with_http_client(stub.clone()); + + let context = PullSyncContext { + ec_id: ec_id.clone(), + snapshot, + }; + dispatch_pull_sync( + &settings, + &graph, + ®istry, + &AllowAllRateLimiter, + &context, + &services, + ); + + assert!( + stub.recorded_backend_names().is_empty(), + "a tombstone snapshot must not dispatch pull sync" + ); + } } diff --git a/crates/trusted-server-core/src/http_util.rs b/crates/trusted-server-core/src/http_util.rs index 74830ff9..dde955bd 100644 --- a/crates/trusted-server-core/src/http_util.rs +++ b/crates/trusted-server-core/src/http_util.rs @@ -256,6 +256,14 @@ fn detect_request_scheme( // 4. Check Fastly-SSL header. On the `EdgeZero` path this is injected from // authoritative Fastly TLS metadata after spoofable headers are stripped, // so it is reliable. On direct or legacy paths it can be spoofed by clients. + // + // Layering wart: this is a vendor-specific header name living in + // platform-neutral core. It is only a fallback — signal #1 above + // (`ClientInfo::tls_protocol`) is the neutral path adapters populate. The + // `fastly-ssl` fallback (plus its entry in `SPOOFABLE_FORWARDED_HEADERS` + // and the origin-forwarding strip in `publisher::rewrite_origin_request`) + // should be replaced by a platform-neutral scheme signal in a separate + // change, after confirming the legacy path is covered by `ClientInfo`. if let Some(ssl) = req.headers().get("fastly-ssl") { if let Ok(ssl_str) = ssl.to_str() { if ssl_str == "1" || ssl_str.to_lowercase() == "true" { diff --git a/crates/trusted-server-core/src/publisher.rs b/crates/trusted-server-core/src/publisher.rs index 6b0ea3a5..7d87d66e 100644 --- a/crates/trusted-server-core/src/publisher.rs +++ b/crates/trusted-server-core/src/publisher.rs @@ -755,6 +755,55 @@ fn mediator_placeholder_request() -> Request { .expect("MEDIATOR_PLACEHOLDER_URL should be a valid URI") } +/// Copies the downstream request head while leaving its body available to the caller. +fn request_head_snapshot(req: &Request) -> Request { + let mut snapshot = Request::new(EdgeBody::empty()); + *snapshot.method_mut() = req.method().clone(); + *snapshot.uri_mut() = req.uri().clone(); + *snapshot.version_mut() = req.version(); + *snapshot.headers_mut() = req.headers().clone(); + snapshot +} + +fn should_preload_ec_snapshot( + is_navigation: bool, + is_get: bool, + has_ec_id: bool, + has_kv: bool, +) -> bool { + is_navigation && is_get && has_ec_id && has_kv +} + +/// Rewrites a downstream request into an outbound publisher-origin request. +/// +/// Restricts advertised encodings to those the rewrite pipeline can handle, +/// strips the internal `fastly-ssl` scheme signal so it never leaks to the +/// backend, and retargets the URI and `Host` header at the origin. +fn rewrite_origin_request( + req: &mut Request, + target_uri: Uri, + origin_host_header: &str, +) -> Result<(), Report> { + restrict_accept_encoding(req); + // Layering wart: `fastly-ssl` is a vendor-specific header name. Core only + // knows it because the Fastly adapter re-injects it from trusted TLS + // metadata and `detect_scheme` (see `http_util`) still reads it as a + // fallback scheme signal. The neutral signal (`ClientInfo::tls_protocol`) + // is already the primary path, so this strip — and the whole `fastly-ssl` + // coupling in core — should move behind a platform-neutral header in a + // separate change. Until then we strip it here so the edge signal never + // reaches publisher backends. + req.headers_mut().remove("fastly-ssl"); + *req.uri_mut() = target_uri; + req.headers_mut().insert( + header::HOST, + HeaderValue::from_str(origin_host_header).change_context(TrustedServerError::Proxy { + message: "invalid publisher origin host header".to_string(), + })?, + ); + Ok(()) +} + /// Build a minimal [`AuctionContext`] for the collect phase. /// /// See [`AuctionContext::request`]: the orchestrator's collect path runs @@ -1305,7 +1354,7 @@ pub async fn handle_publisher_request( kv: Option<&KvIdentityGraph>, ec_context: &mut EcContext, auction: AuctionDispatch<'_>, - mut req: Request, + req: Request, ) -> Result> { log::debug!("Proxying request to publisher_origin"); @@ -1343,7 +1392,11 @@ pub async fn handle_publisher_request( ); let consent_context = ec_context.consent().clone(); - let ec_id = ec_context.ec_value().filter(|_| ec_allowed); + let ec_id_owned = ec_context + .ec_value() + .filter(|_| ec_allowed) + .map(str::to_owned); + let ec_id = ec_id_owned.as_deref(); let cookie_jar = handle_request_cookies(&req)?; let geo = ec_context.geo_info().cloned(); @@ -1450,6 +1503,34 @@ pub async fn handle_publisher_request( .map(|co| co.price_granularity) .unwrap_or_default(); + let auction_client_request = request_head_snapshot(&req); + let mut origin_request = Some(req); + let should_preload_ec = + should_preload_ec_snapshot(is_navigation, is_get, ec_id.is_some(), kv.is_some()); + let mut pending_origin = None; + if should_preload_ec && services.http_client().supports_concurrent_fanout() { + let mut origin_req = origin_request.take().ok_or_else(|| { + Report::new(TrustedServerError::Proxy { + message: "publisher origin request was already consumed".to_owned(), + }) + })?; + rewrite_origin_request(&mut origin_req, target_uri.clone(), &origin_host_header)?; + pending_origin = Some( + services + .http_client() + .send_async(PlatformHttpRequest::new(origin_req, backend_name.clone())) + .await + .change_context(TrustedServerError::Proxy { + message: "Failed to start publisher origin request".to_string(), + })?, + ); + } + if should_preload_ec { + if let (Some(graph), Some(ec_id)) = (kv, ec_id) { + ec_context.set_kv_snapshot(graph.load_snapshot(ec_id)); + } + } + // Dispatch SSP bid requests while req still has the original client headers // (User-Agent, x-forwarded-for, cookies, etc.). The borrow ends when // dispatch_auction returns — DispatchedAuction holds no lifetime — so req @@ -1477,7 +1558,8 @@ pub async fn handle_publisher_request( ec_id, &consent_context, &request_info, - req.headers() + auction_client_request + .headers() .get("user-agent") .and_then(|v| v.to_str().ok()), ); @@ -1486,7 +1568,7 @@ pub async fn handle_publisher_request( &AuctionEidTargeting { cookie_jar: cookie_jar.as_ref(), ec_id, - kv, + kv_snapshot: ec_context.kv_snapshot(), partner_registry: auction.registry, ec_context, services, @@ -1496,7 +1578,7 @@ pub async fn handle_publisher_request( ); let auction_context = AuctionContext { settings, - request: &req, + request: &auction_client_request, timeout_ms: auction_timeout_ms, provider_responses: None, services, @@ -1582,29 +1664,31 @@ pub async fn handle_publisher_request( } ); - // Only advertise encodings the rewrite pipeline can decode and re-encode. - restrict_accept_encoding(&mut req); - // Strip the internal `fastly-ssl` scheme signal before forwarding to the - // origin. On the EdgeZero path the entry point re-injects this header from - // trusted Fastly TLS metadata so in-process scheme detection works; the - // legacy path never sets it. Either way it is an internal edge signal that - // must not leak to publisher backends. - req.headers_mut().remove("fastly-ssl"); - *req.uri_mut() = target_uri; - req.headers_mut().insert( - header::HOST, - HeaderValue::from_str(&origin_host_header).change_context(TrustedServerError::Proxy { - message: "invalid publisher origin host header".to_string(), - })?, - ); + // Rewrite the origin request only on the eager path where it was not already + // started via `send_async`. The concurrent path rewrote and dispatched it + // above, leaving `origin_request` empty. + if let Some(req) = origin_request.as_mut() { + rewrite_origin_request(req, target_uri, &origin_host_header)?; + } // SSP requests are already racing through the platform HTTP client, so // origin TTFB tracks origin latency rather than the auction timeout. - let mut response = match services - .http_client() - .send(PlatformHttpRequest::new(req, backend_name)) - .await - { + let origin_result = if let Some(pending) = pending_origin { + services.http_client().wait(pending).await + } else { + services + .http_client() + .send(PlatformHttpRequest::new( + origin_request.take().ok_or_else(|| { + Report::new(TrustedServerError::Proxy { + message: "publisher origin request was already consumed".to_owned(), + }) + })?, + backend_name, + )) + .await + }; + let mut response = match origin_result { Ok(platform_response) => platform_response.response, Err(err) => { if let Some(dispatched) = dispatched_auction.take() { @@ -1792,7 +1876,7 @@ pub(crate) struct MatchedSlotsContext<'a> { struct AuctionEidTargeting<'a> { cookie_jar: Option<&'a CookieJar>, ec_id: Option<&'a str>, - kv: Option<&'a KvIdentityGraph>, + kv_snapshot: &'a crate::ec::EcKvSnapshot, partner_registry: Option<&'a PartnerRegistry>, ec_context: &'a EcContext, services: &'a RuntimeServices, @@ -1821,7 +1905,7 @@ fn apply_auction_eids_and_device( None }; let kv_eids = resolve_auction_eids( - targeting.kv, + targeting.kv_snapshot, targeting.partner_registry, targeting.ec_context, ); @@ -2236,6 +2320,10 @@ pub async fn handle_page_bids( let request_info = crate::http_util::RequestInfo::from_request(&req, services.client_info()); let ec_id = ec_context.ec_value().filter(|_| ec_context.ec_allowed()); + let page_bids_kv_snapshot = match (kv, ec_id) { + (Some(graph), Some(ec_id)) => graph.load_snapshot(ec_id), + _ => crate::ec::EcKvSnapshot::NotRead, + }; let consent_context = ec_context.consent(); let geo = ec_context.geo_info().cloned(); let cookie_jar = handle_request_cookies(&req)?; @@ -2305,7 +2393,7 @@ pub async fn handle_page_bids( &AuctionEidTargeting { cookie_jar: cookie_jar.as_ref(), ec_id, - kv, + kv_snapshot: &page_bids_kv_snapshot, partner_registry: auction.registry, ec_context, services, @@ -2437,7 +2525,41 @@ mod tests { use flate2::write::GzEncoder; use super::*; + + #[test] + fn request_head_snapshot_preserves_downstream_shape_without_body() { + let request = Request::builder() + .method(Method::GET) + .uri("https://publisher.example/article?x=1") + .header(header::HOST, "publisher.example") + .header("fastly-ssl", "1") + .header(header::USER_AGENT, "test-browser") + .body(EdgeBody::from("ignored-body")) + .expect("should build request"); + + let snapshot = request_head_snapshot(&request); + + assert_eq!(snapshot.method(), Method::GET); + assert_eq!(snapshot.uri(), request.uri()); + assert_eq!(snapshot.headers(), request.headers()); + assert!( + matches!(snapshot.body(), EdgeBody::Once(bytes) if bytes.is_empty()), + "snapshot should never duplicate the request body" + ); + } + + #[test] + fn ec_snapshot_preload_requires_navigation_get_ec_and_kv() { + assert!(should_preload_ec_snapshot(true, true, true, true)); + assert!(!should_preload_ec_snapshot(false, true, true, true)); + assert!(!should_preload_ec_snapshot(true, false, true, true)); + assert!(!should_preload_ec_snapshot(true, true, false, true)); + assert!(!should_preload_ec_snapshot(true, true, true, false)); + } use crate::auction::types::{AdFormat, AdSlot, MediaType}; + use crate::consent::ConsentContext; + use crate::ec::kv_backend::test_support::InMemoryEcKv; + use crate::ec::kv_backend::{EcKvLookup, EcKvStore, EcKvWrite, EcKvWriteOutcome}; use crate::integrations::IntegrationRegistry; use crate::platform::test_support::{ build_services_with_http_client, noop_services, StubHttpClient, @@ -2447,6 +2569,157 @@ mod tests { use http::{header, Method, Request as HttpRequest, StatusCode}; use std::sync::Arc; + /// [`EcKvStore`] that records how many HTTP calls the shared stub client had + /// made at the moment of each identity-graph lookup. This exposes the + /// interleaving between origin dispatch and the EC KV read so scheduling + /// tests can prove origin starts before the lookup only for concurrent + /// clients. + struct OrderRecordingKv { + inner: InMemoryEcKv, + http: Arc, + http_calls_at_lookup: Arc, + lookups: Arc, + } + + impl EcKvStore for OrderRecordingKv { + fn store_name(&self) -> &str { + self.inner.store_name() + } + fn lookup(&self, _key: &str) -> Result, Report> { + self.lookups.fetch_add(1, Ordering::SeqCst); + self.http_calls_at_lookup + .store(self.http.recorded_backend_names().len(), Ordering::SeqCst); + // Report a miss: the scheduling assertions only care about ordering. + Ok(None) + } + fn insert( + &self, + key: &str, + write: EcKvWrite<'_>, + ) -> Result> { + self.inner.insert(key, write) + } + fn count_keys_with_prefix( + &self, + prefix: &str, + limit: u32, + ) -> Result> { + self.inner.count_keys_with_prefix(prefix, limit) + } + fn delete(&self, key: &str) -> Result<(), Report> { + self.inner.delete(key) + } + } + + fn scheduling_consent() -> ConsentContext { + ConsentContext { + jurisdiction: crate::consent::jurisdiction::Jurisdiction::NonRegulated, + ..Default::default() + } + } + + fn navigation_request() -> Request { + HttpRequest::builder() + .method(Method::GET) + .uri("https://publisher.example/article") + .header(header::HOST, "publisher.example") + .header("sec-fetch-dest", "document") + .body(EdgeBody::empty()) + .expect("should build navigation request") + } + + /// Drives one EC-capable navigation and returns + /// `(lookups, http_calls_at_first_lookup, result_is_ok)`. + async fn run_scheduling_probe( + concurrent_fanout: bool, + queue_origin: bool, + ) -> (usize, usize, bool) { + let settings = create_test_settings(); + let http = Arc::new(StubHttpClient::new()); + http.set_concurrent_fanout(concurrent_fanout); + if queue_origin { + http.push_response(200, b"ok".to_vec()); + } + let lookups = Arc::new(AtomicUsize::new(0)); + let http_calls_at_lookup = Arc::new(AtomicUsize::new(0)); + let graph = KvIdentityGraph::new(OrderRecordingKv { + inner: InMemoryEcKv::new("order-store"), + http: Arc::clone(&http), + http_calls_at_lookup: Arc::clone(&http_calls_at_lookup), + lookups: Arc::clone(&lookups), + }); + let services = build_services_with_http_client( + Arc::clone(&http) as Arc + ); + let ec_id = format!("{}.CkId01", "b".repeat(64)); + let mut ec_context = EcContext::new_for_test_with_ip( + Some(ec_id), + scheduling_consent(), + Some("203.0.113.7".to_owned()), + ); + assert!( + ec_context.ec_allowed() && ec_context.ec_value().is_some(), + "test precondition: an active, consent-allowed EC must exist" + ); + let orchestrator = AuctionOrchestrator::new(settings.auction.clone()); + + let result = handle_publisher_request( + &settings, + &services, + Some(&graph), + &mut ec_context, + AuctionDispatch { + orchestrator: &orchestrator, + slots: &[], + registry: None, + }, + navigation_request(), + ) + .await; + + ( + lookups.load(Ordering::SeqCst), + http_calls_at_lookup.load(Ordering::SeqCst), + result.is_ok(), + ) + } + + #[tokio::test] + async fn concurrent_client_starts_origin_before_ec_lookup() { + let (lookups, http_calls_at_lookup, ok) = run_scheduling_probe(true, true).await; + + assert!(ok, "should proxy the origin response"); + assert_eq!(lookups, 1, "should perform exactly one EC lookup"); + assert_eq!( + http_calls_at_lookup, 1, + "a concurrent client must start the origin before its EC KV lookup" + ); + } + + #[tokio::test] + async fn eager_client_reads_ec_before_starting_origin() { + let (lookups, http_calls_at_lookup, ok) = run_scheduling_probe(false, true).await; + + assert!(ok, "should proxy the origin response"); + assert_eq!(lookups, 1, "should perform exactly one EC lookup"); + assert_eq!( + http_calls_at_lookup, 0, + "an eager client must not start the origin before its EC KV lookup" + ); + } + + #[tokio::test] + async fn concurrent_origin_start_failure_skips_ec_and_auction_work() { + // No origin response is queued, so the concurrent `send_async` fails. + let (lookups, _http_calls, ok) = run_scheduling_probe(true, false).await; + + assert!(!ok, "origin-start failure should surface as an error"); + assert_eq!( + lookups, 0, + "origin-start failure must occur before any EC KV work" + ); + } + struct ChunkedReader { chunks: std::collections::VecDeque>, read_count: Arc, diff --git a/docs/superpowers/plans/2026-07-10-kv-eid-request-snapshot-ec-recovery.md b/docs/superpowers/plans/2026-07-10-kv-eid-request-snapshot-ec-recovery.md new file mode 100644 index 00000000..71dc8ba8 --- /dev/null +++ b/docs/superpowers/plans/2026-07-10-kv-eid-request-snapshot-ec-recovery.md @@ -0,0 +1,396 @@ +# KV EID Request Snapshot and EC Recovery Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Reuse one EC KV snapshot across publisher auction, finalize, and pull sync; overlap Fastly origin work with the lookup; and safely rotate orphaned EC cookies to a newly backed identity. + +**Architecture:** A cloneable, EC-ID-bound snapshot carries persisted `KvEntry` state and an optional CAS generation through `EcContext` and Fastly response extensions. Publisher navigations preload it after a truly asynchronous origin start, finalize consumes and updates it, and pull sync uses the finalized state and request-wide bulk persistence. Missing rows rotate only during real-browser document navigation; withdrawal becomes existing-key-only. + +**Tech Stack:** Rust 2024, `error-stack`, EdgeZero HTTP abstractions, Fastly KV generation/CAS, existing core and adapter test support. + +--- + +## File Map + +- Modify `crates/trusted-server-core/src/ec/mod.rs`: request-scoped snapshot type, EC-ID binding, navigation/recovery state, and normal-generation snapshot seeding. +- Modify `crates/trusted-server-core/src/ec/kv.rs`: snapshot-aware bulk mutation, existing-key-only tombstones, persisted-state outcomes, and CAS retry tests. +- Modify `crates/trusted-server-core/src/ec/prebid_eids.rs`: separate validated update collection from persistence so orphan creation can include IDs atomically. +- Modify `crates/trusted-server-core/src/ec/finalize.rs`: consume/return snapshot state, rotate authoritative misses, and preserve withdrawal ordering. +- Modify `crates/trusted-server-core/src/auction/endpoints.rs`: resolve server-side auction EIDs from a snapshot instead of performing KV I/O. +- Modify `crates/trusted-server-core/src/publisher.rs`: snapshot original request head, conditionally start origin first, preload snapshot, dispatch auction, and await origin. +- Modify `crates/trusted-server-core/src/ec/pull_sync.rs`: consume finalized snapshot and aggregate successful partner results across all HTTP batches before one bulk persistence operation. +- Modify `crates/trusted-server-adapter-fastly/src/app.rs`: carry navigation and snapshot state through `EcRequestState`/`EcFinalizeState` and publisher dispatch. +- Modify `crates/trusted-server-adapter-fastly/src/main.rs`: finalize mutable EC state, pass updated snapshot to post-send pull sync, and keep #880 concerns out. +- Modify adapter call sites/tests only as required by shared signature changes. + +### Task 1: Define EC KV Snapshot Semantics + +**Files:** +- Modify: `crates/trusted-server-core/src/ec/mod.rs` +- Modify: `crates/trusted-server-core/src/ec/kv.rs` +- Test: `crates/trusted-server-core/src/ec/mod.rs` +- Test: `crates/trusted-server-core/src/ec/kv.rs` + +- [ ] **Step 1: Write failing snapshot-state tests** + +Add tests proving that a snapshot distinguishes not-read, missing, failed, and present; a present snapshot is bound to one EC ID; a successful mutation can retain an entry without a usable generation; and a different active EC ID cannot consume stale state. + +- [ ] **Step 2: Run the focused tests and verify RED** + +Run: `cargo test -p trusted-server-core ec::tests::kv_snapshot` + +Expected: compilation/test failure because the snapshot API does not exist. + +- [ ] **Step 3: Implement the minimal snapshot type and accessors** + +Use an enum whose present state contains `ec_id`, `KvEntry`, and `Option`. Keep the type cloneable and do not store `Report`. Add helpers for ID-safe entry/generation access and state replacement. + +- [ ] **Step 4: Write failing authoritative-creation tests** + +Cover create-if-absent Written, collision, and store error outcomes. Cover generation collision retry, bounded collision exhaustion, and rollback with no request-local candidate exposed as persisted. + +- [ ] **Step 5: Run authoritative-creation tests and verify RED** + +Run: `cargo test -p trusted-server-core ec::kv::tests::create_if_absent` + +Run: `cargo test -p trusted-server-core ec::tests::generate_collision` + +Expected: fail because the Add-only outcome and bounded retry do not exist. + +- [ ] **Step 6: Implement Add-only creation and generation seeding** + +Introduce a `KvIdentityGraph` create-if-absent outcome that preserves `Written` versus collision while propagating store errors. Replace generation-time `create_or_revive` use with that Add-only outcome. Seed the exact candidate entry only after `Written`; on collision generate a different fresh suffix and retry within a small bound. Never revive a tombstone or claim a colliding request-local candidate was persisted. Do not change cookie emission timing. + +- [ ] **Step 7: Run focused tests and verify GREEN** + +Run: `cargo test -p trusted-server-core ec::tests::kv_snapshot` + +Run: `cargo test -p trusted-server-core ec::kv::tests::create_if_absent` + +Expected: snapshot tests and Written/collision/store-error creation tests pass. + +- [ ] **Step 8: Run existing EC generation tests** + +Run: `cargo test -p trusted-server-core ec::tests::generate` + +Expected: existing generation and failure rollback tests pass. + +### Task 2: Add Snapshot-Aware KV Mutations + +**Files:** +- Modify: `crates/trusted-server-core/src/ec/kv.rs` +- Test: `crates/trusted-server-core/src/ec/kv.rs` + +- [ ] **Step 1: Write failing bulk-mutation tests** + +Cover: supplied generation avoids an initial read; unchanged updates preserve generation; successful writes return persisted entry with unavailable generation; unavailable generation refreshes exactly once; CAS conflict rereads and re-merges; tombstone rejects updates; missing never creates a root; store failure returns failed state rather than request-local data. + +- [ ] **Step 2: Run focused tests and verify RED** + +Run: `cargo test -p trusted-server-core ec::kv::tests::snapshot` + +Expected: failure because snapshot-aware mutation APIs do not exist. + +- [ ] **Step 3: Implement snapshot-aware bulk upsert** + +Refactor the existing merge/CAS loop behind one API accepting an initial snapshot. Preserve `upsert_partner_id_if_exists` semantics for batch sync and keep compatibility wrappers only where necessary. + +- [ ] **Step 4: Write failing conditional-tombstone tests** + +Cover present row, missing row, CAS conflict, disappearance during retry, and store failure. Assert missing/failed cases perform no insert. + +- [ ] **Step 5: Implement existing-key-only tombstones** + +Use the carried generation when available, refresh when unavailable, and retry conflicts without unconditional overwrite. Retain the 24-hour TTL and empty tombstone entry. + +- [ ] **Step 6: Run focused and module tests** + +Run: `cargo test -p trusted-server-core ec::kv::tests` + +Expected: all KV tests pass. + +### Task 3: Separate EID Collection from Persistence + +**Files:** +- Modify: `crates/trusted-server-core/src/ec/prebid_eids.rs` +- Test: `crates/trusted-server-core/src/ec/prebid_eids.rs` + +- [ ] **Step 1: Write failing collection tests** + +Prove one helper returns validated, registry-matched, deduplicated updates from `ts-eids` and `sharedId` without touching KV. + +- [ ] **Step 2: Run focused tests and verify RED** + +Run: `cargo test -p trusted-server-core ec::prebid_eids::tests::collect` + +- [ ] **Step 3: Extract the collection API** + +Reuse existing parsing and validation. Keep logging best-effort and keep raw identifiers out of logs. + +- [ ] **Step 4: Route existing ingestion through the helper** + +Preserve current public behavior while allowing finalize to apply updates to a new orphan-recovery entry before `Add`. + +- [ ] **Step 5: Run module tests and verify GREEN** + +Run: `cargo test -p trusted-server-core ec::prebid_eids::tests` + +### Task 4: Implement Finalize Recovery and Persisted Outcomes + +**Files:** +- Modify: `crates/trusted-server-core/src/ec/finalize.rs` +- Modify: `crates/trusted-server-core/src/ec/mod.rs` +- Test: `crates/trusted-server-core/src/ec/finalize.rs` + +- [ ] **Step 1: Write failing persisted-mutation finalize tests** + +Cover returning-user update from a supplied generation, unchanged update, CAS failure returning failed state, and successful update returning the persisted entry without generation. Cover NotRead returning EIDs performing exactly one lazy lookup/mutation, NotRead never rotating, and a failed lazy lookup degrading without retry. + +- [ ] **Step 2: Run the persisted-mutation tests and verify RED** + +Run: `cargo test -p trusted-server-core ec::finalize::tests::snapshot` + +- [ ] **Step 3: Implement persisted finalize mutation and verify GREEN** + +Use the snapshot-aware KV API and return only authoritative persisted state. + +Run: `cargo test -p trusted-server-core ec::finalize::tests::snapshot` + +- [ ] **Step 4: Write failing orphan-recovery tests** + +Cover authoritative miss on real-browser navigation rotating to a fresh EC, applying request EIDs before one `Add`, emitting a cookie only after success, bounded ID collision retry, KV failure not rotating, tombstone not rotating, NotRead not rotating, and subresource/non-browser requests not rotating. + +- [ ] **Step 5: Run orphan tests RED, implement rotation, and verify GREEN** + +Run before and after implementation: `cargo test -p trusted-server-core ec::finalize::tests::orphan` + +Make finalization update request EC state only after the replacement entry is durably added. Return the authoritative persisted snapshot for post-send consumers. + +- [ ] **Step 6: Write failing withdrawal tests** + +Assert browser cookie expiry remains immediate while present/missing/failed snapshot states follow the conditional tombstone contract. Cover cookie EC differing from active EC: use the carried snapshot only for its matching ID, independently look up/CAS the other, and never create either missing ID. + +- [ ] **Step 7: Run withdrawal tests RED, implement, and verify GREEN** + +Run before and after implementation: `cargo test -p trusted-server-core ec::finalize::tests::withdrawal` + +- [ ] **Step 8: Run complete finalize and EC tests** + +Run: `cargo test -p trusted-server-core ec::finalize::tests` + +Run: `cargo test -p trusted-server-core ec::tests` + +### Task 5: Thread Snapshot and Browser Eligibility Through Call Sites + +**Files:** +- Modify: `crates/trusted-server-core/src/publisher.rs` +- Modify: `crates/trusted-server-adapter-fastly/src/app.rs` +- Modify: `crates/trusted-server-adapter-axum/src/app.rs` +- Modify: `crates/trusted-server-adapter-cloudflare/src/app.rs` +- Modify: `crates/trusted-server-adapter-spin/src/app.rs` + +- [ ] **Step 1: Write failing handler-contract tests** + +Require an explicit mutable snapshot and real-browser eligibility input at the publisher boundary. Prove non-Fastly adapters use NotRead/no-KV state and cannot authorize recovery merely from navigation headers. + +- [ ] **Step 2: Run focused tests and verify RED** + +Run: `cargo test -p trusted-server-core publisher::tests::ec_snapshot_contract` + +- [ ] **Step 3: Thread the contract through every call site** + +Update all publisher handler invocations together so the workspace remains compilable. Fastly supplies request state; other adapters pass explicit NotRead/no-KV and recovery-ineligible `false`. Do not change auction resolution or scheduling yet. + +- [ ] **Step 4: Check all adapter compilation** + +Run: `cargo check-fastly` + +Run: `cargo check-axum` + +Run: `cargo check-cloudflare` + +Run: `cargo check-spin` + +Expected: all call sites compile before auction behavior changes. + +### Task 6: Resolve Auction EIDs and Schedule Origin from the Snapshot + +**Files:** +- Modify: `crates/trusted-server-core/src/auction/endpoints.rs` +- Modify: `crates/trusted-server-core/src/publisher.rs` +- Test: `crates/trusted-server-core/src/auction/endpoints.rs` +- Test: `crates/trusted-server-core/src/publisher.rs` + +- [ ] **Step 1: Write failing snapshot-resolution tests** + +Cover present live entry, missing, failed, not-read, mismatched EC ID, tombstone, denied consent, and absent registry. Assert resolution performs no KV operation. Update the separate page-bids endpoint to perform its own single explicit snapshot load before resolution; do not retain a compatibility KV lookup inside `resolve_auction_eids`. + +- [ ] **Step 2: Run focused tests and verify RED** + +Run: `cargo test -p trusted-server-core auction::endpoints::tests::resolve_auction_eids` + +- [ ] **Step 3: Replace every graph lookup call surface with snapshot resolution** + +Keep client-EID merge and consent gating unchanged. Remove the discarded-generation lookup from auction resolution. Make page-bids snapshot ownership explicit and local because it is outside the publisher-navigation lifecycle. + +- [ ] **Step 4: Run auction endpoint tests and verify GREEN** + +Run: `cargo test -p trusted-server-core auction::endpoints::tests` + +- [ ] **Step 5: Write failing request-head snapshot tests** + +Assert bidders see the original method, URI, Host, scheme signal, and selected headers after the real request is rewritten and consumed by origin dispatch. Assert provider outbound sanitization tests remain unchanged. + +- [ ] **Step 6: Write failing concurrent-order and preload-gate tests** + +Record events proving `origin_start` precedes `kv_lookup` and `auction_dispatch`, then `origin_wait` follows dispatch for a truly concurrent client. Also prove real-browser document navigations preload when auctions are disabled, no slots match, or consent denies auction; non-document requests do not preload; and non-EC requests retain the ordinary origin path. + +- [ ] **Step 7: Write failing eager-client ordering test** + +Prove a client reporting no concurrent fan-out and needing a snapshot performs `kv_lookup -> auction_dispatch -> origin_execute`, never completing the eager origin before auction dispatch. + +- [ ] **Step 8: Write failing error-path tests** + +Cover origin-start failure causing no KV/auction work, auction dispatch failure still returning origin, and origin completion failure emitting one abandoned-auction terminal event. + +- [ ] **Step 9: Run focused tests and verify RED** + +Run: `cargo test -p trusted-server-core publisher::tests::origin_auction_order` + +- [ ] **Step 10: Implement capability-aware scheduling** + +Extract small helpers rather than duplicating auction construction. For concurrent clients, start origin first, preload/refresh while it is in flight, dispatch from the bodyless request head, then await origin. For eager clients, preload first, dispatch auction, then execute origin. Apply the explicit navigation/browser/active-EC/graph gates from the spec even when auction work is skipped. + +- [ ] **Step 11: Run endpoint and publisher tests and verify GREEN** + +Run: `cargo test -p trusted-server-core publisher::tests` + +### Task 7: Thread Finalize Outcome Through Fastly + +**Files:** +- Modify: `crates/trusted-server-adapter-fastly/src/app.rs` +- Modify: `crates/trusted-server-adapter-fastly/src/main.rs` +- Test: `crates/trusted-server-adapter-fastly/src/app.rs` +- Test: `crates/trusted-server-adapter-fastly/src/main.rs` + +- [ ] **Step 1: Write failing state-threading tests** + +Assert publisher navigation state carries snapshot and navigation eligibility into `EcFinalizeState`; named/subresource routes cannot request orphan rotation; normal generated state is preserved; and cookie/active EC mismatch retains enough state for two-ID withdrawal handling. + +- [ ] **Step 2: Run focused Fastly tests and verify RED** + +Run: `cargo test-fastly ec_finalize_state` + +- [ ] **Step 3: Implement state threading and mutable finalize outcome** + +Pass the snapshot into `handle_publisher_request`, pop mutable finalize state in `main.rs`, apply the returned EC context/snapshot, send the response, and pass finalized state to pull sync. + +- [ ] **Step 4: Run focused Fastly tests and verify GREEN** + +Run: `cargo test-fastly ec_finalize_state` + +### Task 8: Reuse Finalized State in Pull Sync + +**Files:** +- Modify: `crates/trusted-server-core/src/ec/pull_sync.rs` +- Modify: `crates/trusted-server-adapter-fastly/src/main.rs` +- Test: `crates/trusted-server-core/src/ec/pull_sync.rs` + +- [ ] **Step 1: Write failing eligibility tests** + +Prove a present finalized snapshot selects only missing partners without an initial KV read; failed/missing/not-read/tombstone state dispatches no pull work in #851. Do not add #880 completeness-marker or partner-set precheck behavior. + +- [ ] **Step 2: Write failing request-wide aggregation tests** + +Provide successful partner responses across multiple concurrency batches and assert one final bulk update. Cover a usable generation, a finalize-written unavailable generation requiring exactly one refresh read, CAS conflict re-merge, tombstone on refresh, and persistence failure. + +- [ ] **Step 3: Run focused tests and verify RED** + +Run: `cargo test -p trusted-server-core ec::pull_sync::tests` + +- [ ] **Step 4: Implement snapshot-aware pull sync** + +Separate network-batch draining from KV persistence. Accumulate deduplicated updates for the whole request, persist once after all responses, and keep the current rate limits, allowlist, tokens, response bounds, and best-effort logging. + +- [ ] **Step 5: Run pull-sync tests and verify GREEN** + +Run: `cargo test -p trusted-server-core ec::pull_sync::tests` + +### Task 9: Run Regression Suites + +**Files:** +- Modify as required: `crates/trusted-server-adapter-axum/src/app.rs` +- Modify as required: `crates/trusted-server-adapter-cloudflare/src/app.rs` +- Modify as required: `crates/trusted-server-adapter-spin/src/app.rs` +- Modify as required: shared tests and test support + +- [ ] **Step 1: Compile all adapters and fix only signature fallout** + +Run: `cargo check-fastly` + +Run: `cargo check-axum` + +Run: `cargo check-cloudflare` + +Run: `cargo check-spin` + +- [ ] **Step 2: Run adapter test suites** + +Run: `cargo test-fastly` + +Run: `cargo test-axum` + +Run: `cargo test-cloudflare` + +Run: `cargo test-spin` + +- [ ] **Step 3: Run cross-adapter parity tests** + +Run: `cargo test --manifest-path crates/trusted-server-integration-tests/Cargo.toml --test parity` + +- [ ] **Step 4: Run formatting** + +Run: `cargo fmt --all -- --check` + +- [ ] **Step 5: Run target-matched clippy** + +Run: `cargo clippy-fastly` + +Run: `cargo clippy-axum` + +Run: `cargo clippy-cloudflare` + +Run: `cargo clippy-cloudflare-wasm` + +Run: `cargo clippy-spin-native` + +Run: `cargo clippy-spin-wasm` + +- [ ] **Step 6: Inspect the final diff** + +Run: `git diff --check` + +Run: `git status --short` + +Confirm no #880 completeness marker, partner fingerprint, or unrelated refactor entered the branch. + +### Task 10: Final Code Review + +**Files:** +- Review all modified production and test files + +- [ ] **Step 1: Review against the approved spec** + +Check every success criterion and privacy invariant against code and tests. + +- [ ] **Step 2: Review concurrency and cost accounting** + +Manually trace returning user, first generation, orphan miss, KV failure, withdrawal, finalize write followed by pull write, no auction, eager adapter, and concurrent adapter. + +- [ ] **Step 3: Re-run any test affected by review fixes** + +Use the smallest target-specific command first, then repeat the relevant adapter suite. + +- [ ] **Step 4: Commit implementation in coherent units** + +Use small commits aligned with snapshot/KV primitives, publisher scheduling, finalize recovery, and pull-sync reuse. Do not amend the reviewed design commit. diff --git a/docs/superpowers/specs/2026-07-10-kv-eid-request-snapshot-ec-recovery-design.md b/docs/superpowers/specs/2026-07-10-kv-eid-request-snapshot-ec-recovery-design.md new file mode 100644 index 00000000..cc08e008 --- /dev/null +++ b/docs/superpowers/specs/2026-07-10-kv-eid-request-snapshot-ec-recovery-design.md @@ -0,0 +1,268 @@ +# KV EID Request Snapshot and EC Recovery Design + +## Objective + +Remove redundant EC identity-graph reads from the publisher-navigation hot path, +overlap Fastly origin latency with KV lookup and auction dispatch, and recover +from an orphaned `ts-ec` cookie without allowing an untrusted cookie to create an +arbitrary identity-graph root. + +## Confirmed Problems + +An eligible Fastly publisher navigation currently performs a synchronous KV +lookup while decorating the auction before the publisher origin starts. EC +finalization can perform a second pre-send read-modify-write lookup while +ingesting `ts-eids` or `sharedId`. Post-send pull sync performs another lookup +for eligibility and then one read-modify-write lookup per successful partner. + +When a request contains a valid-looking EC cookie but its KV row is missing, +generation is skipped because an EC ID is already active. Browser EID ingestion +and pull sync both reject the missing root, leaving the request in a permanent +degraded state until the cookie expires. + +The cookie and live KV entry currently both have a one-year lifetime and neither +is refreshed on ordinary returning visits. The concrete correctness defect is +therefore missing-row recovery, not a general long-term mismatch between two +rolling expiration policies. + +## Scope + +This change covers the publisher-navigation EC path on Fastly, shared core EC +snapshot and mutation primitives, response finalization, and pull sync. It does +not change batch-sync root-creation policy, introduce sampled sliding TTL +refresh, change the EC cookie wire format, or refactor unrelated adapters. + +GitHub issue #880 separately owns avoiding pull-sync reads when no partners are +enabled or when a browser-side completeness marker proves the current partner +set is complete. This design does not add that early no-partner optimization, +the completeness marker, partner-set fingerprinting, or marker validation. Its +pull-sync changes consume the single request-scoped snapshot owned by #851 and +leave a clear pre-dispatch boundary where #880 can independently skip work +without introducing another EC cache. + +## Request-Scoped KV Snapshot + +Introduce a snapshot type tied to one EC ID. It must distinguish: + +- lookup not attempted; +- authoritative missing row; +- present row with `KvEntry` and an optional usable generation; +- lookup failure. + +The failed state is a marker rather than a clone of `Report`, because the state +must travel through cloneable response extensions. The original error is logged +at the lookup boundary. Consumers degrade without retrying a failed lookup on +the hot path. + +The EC ID association prevents a snapshot from being reused after orphan +recovery rotates the active ID. A present snapshot with a generation supplies +the initial CAS input to the next update. The current Fastly insert API reports +only written or precondition failed; it does not return the generation produced +by a successful write. Consequently, a successful mutation retains the updated +in-memory entry but marks its generation unavailable. The next writer can still +use that entry for read-only decisions, but must perform one refresh lookup +before another CAS. A CAS precondition failure likewise causes a fresh lookup, +merge, and bounded retry. + +## Origin and Auction Scheduling + +Before consuming the downstream publisher request, build a bodyless auction +request snapshot containing the original method, URI, version, and headers. +This is an in-process compatibility view, not an outbound request: providers +continue to receive the same client-facing request shape through +`AuctionContext` and must not observe the origin-rewritten URI or Host header. +The existing provider-specific outbound allowlist remains authoritative. For +example, Prebid copies only selected browser headers and applies its configured +consent-cookie forwarding policy; internal and hop-by-hop headers are not newly +forwarded by this snapshot. Consent-denied auctions do not dispatch at all. + +When an EC-capable publisher request needs a snapshot and +`PlatformHttpClient::supports_concurrent_fanout()` is true: + +1. Build the base auction request and client-request snapshot. +2. Rewrite and start the publisher-origin request with `send_async`. +3. Read the EC KV row once while the origin is in flight. +4. If auction-eligible, decorate and dispatch the auction using the + client-request snapshot. +5. Await the pending origin request. + +For eager implementations where `send_async` completes the upstream request +before returning, retain dispatch-before-origin ordering. Cloudflare and Spin +must not wait for the complete origin response before starting their auction. +On Fastly, real-browser document navigations with an active EC and configured +graph preload the snapshot even when auctions are disabled, no slots match, or +consent prevents auction dispatch. Their lookup still occurs after the origin +has started, so orphan detection and finalization remain available without +placing the read before origin start. Non-document publisher requests do not +preload and cannot trigger orphan rotation. Routes that do not proxy a +publisher origin leave the snapshot not-read; finalization may perform its +existing lazy lookup for EID ingestion, while orphan recovery remains restricted +to real-browser document navigations and explicit withdrawal remains +route-independent. Non-EC publisher requests retain the ordinary origin send +path. + +Normal `generate_if_needed` creation seeds a generation-unavailable present +snapshot with the exact `KvEntry` successfully added to KV. It does not +pre-apply EID cookies in the generation layer. On a document navigation, the +single origin-overlapped preload refreshes that snapshot and obtains the +generation needed by finalize, so ordinary first-generation behavior joins the +same read lifecycle without an extra pre-send lookup. + +Origin-start failure returns the existing proxy error without performing KV or +auction work. Auction dispatch failure remains best-effort and does not prevent +the already-started origin response from being returned. Origin failure after +auction dispatch preserves the existing abandoned-auction telemetry behavior +and emits its terminal event exactly once. + +## Auction EID Resolution + +Auction resolution accepts the request snapshot rather than a KV graph. A +present live entry resolves registered partner IDs. Missing, failed, not-read, +or tombstone state produces no server-side EIDs and does not fail the auction. +Client-provided EIDs continue through the existing merge and consent gate. + +## Orphaned Cookie Recovery + +An incoming EC ID cannot be authenticated in full. The HMAC prefix is shared by +all IDs behind the same normalized IP, while the suffix is random and unsigned. +Consequently, neither format validation nor HMAC-prefix validation authorizes +recreating the incoming key. + +On an authoritative missing snapshot during consent-granted, real-browser +document-navigation EC finalization: + +1. Generate a fresh EC ID using the current generation path. +2. Parse, validate, deduplicate, and apply request EID-cookie updates to the new + in-memory `KvEntry` before persistence. +3. Atomically add that complete entry in one write. +4. Only after the add succeeds, replace the active EC ID and emit the new + cookie. +5. Return a snapshot associated with the replacement ID and updated entry. Its + generation is unavailable because `Add` does not return the new token. + +KV lookup failure is not a miss and must not rotate the cookie. A present +tombstone must never be revived. Explicit withdrawal runs before recovery and +continues to expire the cookie and write the authoritative tombstone. + +If creating the replacement row fails, retain fail-closed behavior: keep the +original active context for withdrawal bookkeeping, return a failed snapshot, +do not emit a replacement cookie, and do not create partner mappings. A random +suffix collision is handled by generating another fresh ID and retrying a small +bounded number of times; no existing key is overwritten or revived. + +Batch sync and generic partner upsert APIs continue to reject missing roots. +Only the trusted browser finalization flow can initiate orphan recovery. +Subresources and non-document integration requests never rotate an orphaned +cookie. + +## Withdrawal Semantics + +The invariant that an unverified incoming ID cannot create a root also applies +to withdrawal tombstones. The current unconditional tombstone overwrite can +create a new key for a forged cookie and must become existing-key-only: + +- an authoritative missing row is a no-op after expiring the browser cookie; +- a present row is tombstoned with its generation; +- a CAS conflict rereads and retries against the latest existing row; +- a row that disappears during retry is a no-op; +- a lookup failure expires the browser cookie but performs no KV write. + +This preserves withdrawal for real entries without turning arbitrary cookie +values into stored tombstones. Tombstone writes retain their 24-hour TTL and +empty identity payload. + +## Finalization Contract + +EC finalization accepts the request snapshot and returns an outcome containing +the current EC context and updated snapshot. Returning-user EID ingestion uses +the carried entry and generation for its first CAS attempt. It rereads only on +CAS conflict. After a successful write, it returns the updated entry with no +usable generation because the backend does not expose the new token. + +The updated in-memory entry must include partner IDs written during finalization +so post-send pull sync does not dispatch a partner that was just populated. +Cookie and cache-privacy behavior remains centralized in the existing finalize +and entry-point layers. + +Mutation outcomes contain only state known to be persisted. An unchanged merge +returns the original entry and generation. A successful write returns the +persisted updated entry with no generation. A store error or exhausted CAS +retry returns a failed snapshot rather than claiming request-local updates were +stored. Pull sync does not dispatch from failed state. + +## Pull Sync + +Pull sync receives the finalized snapshot and uses it for eligibility without +an initial KV lookup. Missing, failed, not-read, or tombstone snapshots do not +dispatch pull sync. + +Valid partner responses are collected across every HTTP concurrency batch into +one request-wide set of `PartnerIdUpdate` values and merged in one final bulk +CAS operation. Draining a network batch never writes KV. When the finalized snapshot still +has a usable generation, the uncontended case performs one write and no +additional read. When finalization already wrote the row, pull sync uses its +updated entry for eligibility, collects responses, then performs one refresh +lookup to obtain the new generation before its bulk CAS. A conflict rereads the +latest entry, rejects a tombstone, re-merges every collected update, and retries +within the existing bound. Pull sync never creates a missing root. + +Rate limiting, URL allowlisting, bearer-token handling, response-size limits, +UID validation, concurrency limits, and best-effort post-send behavior remain +unchanged. + +## Adapter Surface + +Fastly owns the configured EC KV graph and carries the snapshot through +`EcRequestState` and `EcFinalizeState`. The graph itself is rebuilt at the entry +point as it is today; only cloneable entry data and generation travel in response +extensions. + +Axum, Cloudflare, and Spin currently call the shared publisher path without an +EC KV graph. They must continue compiling and retain their existing auction and +origin behavior. Scheduling tests cover both concurrent and eager HTTP-client +implementations so future adapters cannot accidentally regress auction timing. + +## Error and Privacy Invariants + +- A KV error never becomes an authoritative miss. +- No unverified incoming EC ID can create a KV root. +- A cookie is emitted only after its backing row exists. +- Tombstones are never converted to live entries by enrichment. +- CAS conflicts re-merge rather than overwrite concurrent data. +- Consent-denied requests expose no EC or EID data. +- Post-send failures never change the client response. +- Logs use redacted EC identifiers through the existing `log_id` helper. + +## Test Strategy + +Use strict red-green-refactor cycles. Add focused unit tests for snapshot state, +snapshot-bound EID resolution, CAS reuse and conflict retry, orphan rotation, +failure versus miss, tombstone protection, and bulk pull updates. Include a +finalize-write-then-pull test proving eligibility uses the updated entry and the +pull writer refreshes the unavailable generation exactly once. Cover successful +pull responses spanning multiple HTTP concurrency batches and prove they still +produce one request-wide KV merge. Add publisher scheduling tests using +recording HTTP/KV collaborators to prove origin starts before the lookup only +for truly concurrent clients and that bidders see the original downstream +request. Cover origin-start failure with no KV/auction work, auction dispatch +failure followed by a successful origin response, and origin completion failure +with exactly one abandoned-auction terminal event. Add withdrawal tests for +present, missing, conflicting, and failed snapshot states, plus mutation-failure +tests proving unpersisted request-local IDs never appear in returned snapshots. + +Run targeted core tests after each behavior change, followed by adapter suites, +formatting, and target-matched clippy according to `CLAUDE.md`. + +## Success Criteria + +- An eligible Fastly navigation starts the publisher origin before its EC KV + lookup and SSP dispatch. +- Auction, finalize, and pull eligibility share one normal-path KV read. A + subsequent pull write may require one generation-refresh read when finalize + already wrote the row. +- Successful pull-sync responses use one bulk write rather than one RMW cycle + per partner. +- Orphaned cookies recover through a newly generated, KV-backed EC ID. +- KV errors and tombstones remain fail-closed. +- Eager adapters do not delay auction dispatch behind origin completion. +- All target-specific tests, formatting checks, and clippy checks pass.