From 5c7151601de5f0a9e6161c29667a435d3b5aaf1b Mon Sep 17 00:00:00 2001 From: Christian Date: Mon, 13 Jul 2026 14:53:01 -0500 Subject: [PATCH] Group batch sync mappings by EC ID --- .../trusted-server-core/src/ec/batch_sync.rs | 479 ++++++++++++++++-- crates/trusted-server-core/src/ec/kv.rs | 24 + docs/guide/api-reference.md | 28 +- docs/guide/ec-setup-guide.md | 2 +- docs/guide/edge-cookies.md | 2 +- ...-13-issue-882-group-batch-sync-by-ec-id.md | 259 ++++++++++ 6 files changed, 755 insertions(+), 39 deletions(-) create mode 100644 docs/superpowers/plans/2026-07-13-issue-882-group-batch-sync-by-ec-id.md diff --git a/crates/trusted-server-core/src/ec/batch_sync.rs b/crates/trusted-server-core/src/ec/batch_sync.rs index 0e0f3b90..eef91f0c 100644 --- a/crates/trusted-server-core/src/ec/batch_sync.rs +++ b/crates/trusted-server-core/src/ec/batch_sync.rs @@ -2,15 +2,18 @@ //! //! Partners send authenticated batch ID sync requests via Bearer token. //! Each mapping associates an `ec_id` (`{64hex}.{6alnum}`) -//! with the partner's user ID. Mappings are individually validated and -//! written to the KV identity graph, with per-mapping rejection reasons -//! reported in the response. +//! with the partner's user ID. Mappings are individually validated, then valid +//! mappings are grouped by normalized EC ID before one call to the KV update +//! path per group. +//! Responses still report outcomes per original mapping index. //! //! Mapping timestamps are retained in the request schema for client //! compatibility, but the EC identity graph no longer stores per-partner sync -//! timestamps. Valid mappings therefore use idempotent last-write-wins -//! semantics: unchanged UIDs are accepted without a write; different UIDs -//! replace the stored value regardless of timestamp. +//! timestamps. The last valid mapping in request order supplies each group's +//! UID; unchanged UIDs are accepted without a write, and different UIDs replace +//! the stored value regardless of timestamp. + +use std::collections::HashMap; use edgezero_core::body::Body as EdgeBody; use error_stack::{Report, ResultExt}; @@ -88,6 +91,17 @@ struct MappingError { reason: &'static str, } +/// Valid mappings sharing one normalized EC ID. +/// +/// `indexes` stays in request order, while the groups themselves stay ordered +/// by first valid occurrence. The lookup map used to locate this structure is +/// never used for processing order. +struct MappingGroup { + ec_id: String, + partner_uid: String, + indexes: Vec, +} + // --------------------------------------------------------------------------- // Handler // --------------------------------------------------------------------------- @@ -179,19 +193,28 @@ fn content_length_exceeds_limit(req: &Request, max_body_size: usize) - .is_some_and(|content_length| content_length > max_body_size) } +/// Validates all mappings, then processes each normalized EC ID once. +/// +/// Successful and eligibility outcomes fan out to every valid input in a +/// group. On infrastructure failure, the failing group and all unprocessed +/// valid groups are rejected as unavailable; already validated invalid inputs +/// keep their specific errors. Errors are sorted by original input index. fn process_mappings( writer: &dyn BatchSyncWriter, partner_id: &str, mappings: &[SyncMapping], ) -> (usize, Vec) { - let mut accepted: usize = 0; let mut errors = Vec::new(); + let mut groups: Vec = Vec::new(); + let mut group_indexes: HashMap = HashMap::new(); - for (idx, mapping) in mappings.iter().enumerate() { + // Validate all inputs before beginning KV work. The vector preserves group + // order; the map only locates an existing group in constant time. + for (index, mapping) in mappings.iter().enumerate() { let ec_id = normalize_ec_id_for_kv(&mapping.ec_id); if !is_valid_ec_id(&ec_id) { errors.push(MappingError { - index: idx, + index, reason: REASON_INVALID_EC_ID, }); continue; @@ -199,42 +222,57 @@ fn process_mappings( if mapping.partner_uid.trim().is_empty() || mapping.partner_uid.len() > MAX_UID_LENGTH { errors.push(MappingError { - index: idx, + index, reason: REASON_INVALID_PARTNER_UID, }); continue; } - match writer.upsert_partner_id_if_exists(&ec_id, partner_id, &mapping.partner_uid) { + + if let Some(&group_index) = group_indexes.get(&ec_id) { + let group = &mut groups[group_index]; + group.partner_uid.clone_from(&mapping.partner_uid); + group.indexes.push(index); + } else { + group_indexes.insert(ec_id.clone(), groups.len()); + groups.push(MappingGroup { + ec_id, + partner_uid: mapping.partner_uid.clone(), + indexes: vec![index], + }); + } + } + + let mut accepted = 0; + for (group_index, group) in groups.iter().enumerate() { + match writer.upsert_partner_id_if_exists(&group.ec_id, partner_id, &group.partner_uid) { Ok(UpsertResult::Written | UpsertResult::Unchanged) => { - accepted += 1; + accepted += group.indexes.len(); } Ok(UpsertResult::NotFound | UpsertResult::ConsentWithdrawn) => { - errors.push(MappingError { - index: idx, + errors.extend(group.indexes.iter().map(|&index| MappingError { + index, reason: REASON_INELIGIBLE, - }); + })); } Err(err) => { log::warn!( - "Batch sync KV write failed for index {idx} (ec_id '{}'): {err:?}", - log_id(&mapping.ec_id), + "Batch sync KV write failed for group starting at index {} (ec_id '{}'): {err:?}", + group.indexes[0], + log_id(&group.ec_id), ); - errors.push(MappingError { - index: idx, - reason: REASON_KV_UNAVAILABLE, - }); - // Abort remaining mappings on infrastructure failure. - for remaining_idx in (idx + 1)..mappings.len() { - errors.push(MappingError { - index: remaining_idx, + for unavailable_group in &groups[group_index..] { + errors.extend(unavailable_group.indexes.iter().map(|&index| MappingError { + index, reason: REASON_KV_UNAVAILABLE, - }); + })); } break; } } } + errors.sort_by_key(|error| error.index); + debug_assert_eq!(accepted + errors.len(), mappings.len()); (accepted, errors) } @@ -301,29 +339,47 @@ mod tests { } } + #[derive(Clone, Debug, PartialEq, Eq)] + struct WriterCall { + ec_id: String, + partner_id: String, + uid: String, + } + struct MockWriter { results: std::cell::RefCell>>>, + calls: std::cell::RefCell>, } impl MockWriter { fn new(results: Vec>>) -> Self { Self { results: std::cell::RefCell::new(results.into()), + calls: std::cell::RefCell::new(Vec::new()), } } + + fn calls(&self) -> Vec { + self.calls.borrow().clone() + } } impl BatchSyncWriter for MockWriter { fn upsert_partner_id_if_exists( &self, - _ec_id: &str, - _partner_id: &str, - _uid: &str, + ec_id: &str, + partner_id: &str, + uid: &str, ) -> Result> { + self.calls.borrow_mut().push(WriterCall { + ec_id: ec_id.to_owned(), + partner_id: partner_id.to_owned(), + uid: uid.to_owned(), + }); self.results .borrow_mut() .pop_front() - .expect("should provide mock result for each mapping") + .expect("should provide mock result for each group") } } @@ -361,6 +417,14 @@ mod tests { .expect("should build authorized batch request") } + fn response_json(response: Response) -> serde_json::Value { + let body = response + .into_body() + .into_bytes() + .expect("should contain batch-sync response"); + serde_json::from_slice(&body).expect("should serialize batch-sync response") + } + fn test_registry() -> PartnerRegistry { let partners = vec![make_test_partner( "ssp.example.com", @@ -578,8 +642,12 @@ mod tests { Ok(UpsertResult::NotFound), Ok(UpsertResult::ConsentWithdrawn), ]); - let ec_id = format!("{}.ABC123", "a".repeat(64)); - let mappings = vec![mapping(&ec_id, "uid-1", 100), mapping(&ec_id, "uid-2", 101)]; + let missing_ec_id = format!("{}.ABC123", "a".repeat(64)); + let withdrawn_ec_id = format!("{}.ABC123", "b".repeat(64)); + let mappings = vec![ + mapping(&missing_ec_id, "uid-1", 100), + mapping(&withdrawn_ec_id, "uid-2", 101), + ]; let (accepted, errors) = process_mappings(&writer, "partner", &mappings); @@ -589,26 +657,28 @@ mod tests { assert_eq!(errors[0].reason, REASON_INELIGIBLE); assert_eq!(errors[1].index, 1); assert_eq!(errors[1].reason, REASON_INELIGIBLE); + assert_eq!(writer.calls().len(), 2, "should exercise both outcomes"); } #[test] - fn process_mappings_counts_unchanged_as_accepted() { + fn process_mappings_fans_out_unchanged_to_group_members() { let writer = MockWriter::new(vec![Ok(UpsertResult::Unchanged)]); let ec_id = format!("{}.ABC123", "a".repeat(64)); - let mappings = vec![mapping(&ec_id, "uid-1", 100)]; + let mappings = vec![mapping(&ec_id, "uid-1", 100), mapping(&ec_id, "uid-1", 101)]; let (accepted, errors) = process_mappings(&writer, "partner", &mappings); - assert_eq!(accepted, 1, "should count unchanged mappings as accepted"); + assert_eq!(accepted, 2, "should accept every unchanged group member"); assert!( errors.is_empty(), "should report no errors for unchanged mappings" ); + assert_eq!(writer.calls().len(), 1, "should call once for the group"); } #[test] fn process_mappings_does_not_order_by_timestamp() { - let writer = MockWriter::new(vec![Ok(UpsertResult::Written), Ok(UpsertResult::Written)]); + let writer = MockWriter::new(vec![Ok(UpsertResult::Written)]); let ec_id = format!("{}.ABC123", "a".repeat(64)); let mappings = vec![ mapping(&ec_id, "uid-new", 200), @@ -622,5 +692,342 @@ mod tests { "timestamps are compatibility fields and should not reject older mappings" ); assert!(errors.is_empty(), "should accept valid mappings"); + assert_eq!( + writer.calls(), + vec![WriterCall { + ec_id, + partner_id: "partner".to_owned(), + uid: "uid-old".to_owned(), + }], + "should persist the last valid UID with one writer call" + ); + } + + #[test] + fn process_mappings_groups_normalized_ids_in_first_occurrence_order() { + let writer = MockWriter::new(vec![Ok(UpsertResult::Written), Ok(UpsertResult::Unchanged)]); + let ec_id_a = format!("{}.ABC123", "a".repeat(64)); + let ec_id_a_upper = format!("{}.ABC123", "A".repeat(64)); + let ec_id_b = format!("{}.ABC123", "b".repeat(64)); + let mappings = vec![ + mapping(&ec_id_a, "a-first", 1), + mapping(&ec_id_b, "b-only", 2), + mapping(&ec_id_a_upper, "a-last", 3), + ]; + + let (accepted, errors) = process_mappings(&writer, "partner", &mappings); + + assert_eq!(accepted, 3, "should accept every valid group member"); + assert!(errors.is_empty(), "should report no errors"); + assert_eq!( + writer.calls(), + vec![ + WriterCall { + ec_id: ec_id_a, + partner_id: "partner".to_owned(), + uid: "a-last".to_owned(), + }, + WriterCall { + ec_id: ec_id_b, + partner_id: "partner".to_owned(), + uid: "b-only".to_owned(), + }, + ], + "should make one ordered call per normalized EC ID" + ); + } + + #[test] + fn process_mappings_keeps_suffix_case_distinct() { + let writer = MockWriter::new(vec![Ok(UpsertResult::Written), Ok(UpsertResult::Written)]); + let upper_suffix = format!("{}.ABC123", "a".repeat(64)); + let mixed_suffix = format!("{}.AbC123", "a".repeat(64)); + let mappings = vec![ + mapping(&upper_suffix, "upper", 1), + mapping(&mixed_suffix, "mixed", 2), + ]; + + let (accepted, errors) = process_mappings(&writer, "partner", &mappings); + + assert_eq!(accepted, 2, "should accept both distinct EC IDs"); + assert!(errors.is_empty(), "should report no errors"); + assert_eq!( + writer.calls(), + vec![ + WriterCall { + ec_id: upper_suffix, + partner_id: "partner".to_owned(), + uid: "upper".to_owned(), + }, + WriterCall { + ec_id: mixed_suffix, + partner_id: "partner".to_owned(), + uid: "mixed".to_owned(), + }, + ], + "normalization must preserve suffix case" + ); + } + + #[test] + fn process_mappings_invalid_duplicate_does_not_replace_last_valid_uid() { + let writer = MockWriter::new(vec![Ok(UpsertResult::Written)]); + let ec_id = format!("{}.ABC123", "a".repeat(64)); + let mappings = vec![ + mapping(&ec_id, "first", 1), + mapping(&ec_id, "last", 2), + mapping(&ec_id, " ", 3), + ]; + + let (accepted, errors) = process_mappings(&writer, "partner", &mappings); + + assert_eq!(accepted, 2, "should accept valid group members"); + assert_eq!(errors.len(), 1, "should retain the invalid UID error"); + assert_eq!(errors[0].index, 2, "should retain original error index"); + assert_eq!(errors[0].reason, REASON_INVALID_PARTNER_UID); + assert_eq!( + writer.calls()[0].uid, + "last", + "invalid duplicates must not replace the final valid UID" + ); + } + + #[test] + fn process_mappings_fans_out_ineligible_outcomes_to_group_members() { + let writer = MockWriter::new(vec![ + Ok(UpsertResult::NotFound), + Ok(UpsertResult::ConsentWithdrawn), + ]); + let ec_id_a = format!("{}.ABC123", "a".repeat(64)); + let ec_id_b = format!("{}.ABC123", "b".repeat(64)); + let mappings = vec![ + mapping(&ec_id_a, "a-1", 1), + mapping(&ec_id_b, "b-1", 2), + mapping(&ec_id_a, "a-2", 3), + mapping(&ec_id_b, "b-2", 4), + ]; + + let (accepted, errors) = process_mappings(&writer, "partner", &mappings); + + assert_eq!(accepted, 0, "should reject all ineligible group members"); + assert_eq!( + errors.len(), + mappings.len(), + "should account for every input" + ); + assert_eq!( + errors.iter().map(|error| error.index).collect::>(), + vec![0, 1, 2, 3], + "should sort errors by input index" + ); + assert!( + errors.iter().all(|error| error.reason == REASON_INELIGIBLE), + "should fan out ineligible outcomes" + ); + assert_eq!(writer.calls().len(), 2, "should call once per group"); + } + + #[test] + fn process_mappings_aborts_by_group_and_preserves_sorted_accounting() { + let writer = MockWriter::new(vec![ + Ok(UpsertResult::Written), + Err(Report::new(TrustedServerError::KvStore { + store_name: "ec_store".to_owned(), + message: "down".to_owned(), + })), + ]); + let ec_id_a = format!("{}.ABC123", "a".repeat(64)); + let ec_id_b = format!("{}.ABC123", "b".repeat(64)); + let ec_id_c = format!("{}.ABC123", "c".repeat(64)); + let mappings = vec![ + mapping("invalid", "bad-id", 1), + mapping(&ec_id_a, "a-first", 2), + mapping(&ec_id_b, "b-only", 3), + mapping(&ec_id_a, "a-last", 4), + mapping(&ec_id_c, "c-only", 5), + mapping(&ec_id_c, "", 6), + ]; + + let (accepted, errors) = process_mappings(&writer, "partner", &mappings); + + assert_eq!( + accepted, 2, + "successful groups should accept every member, including later duplicates" + ); + assert_eq!( + errors + .iter() + .map(|error| (error.index, error.reason)) + .collect::>(), + vec![ + (0, REASON_INVALID_EC_ID), + (2, REASON_KV_UNAVAILABLE), + (4, REASON_KV_UNAVAILABLE), + (5, REASON_INVALID_PARTNER_UID), + ], + "should preserve validation errors and fan out failed/unprocessed groups in input order" + ); + assert_eq!( + accepted + errors.len(), + mappings.len(), + "should account for every input exactly once" + ); + assert_eq!( + writer.calls().len(), + 2, + "should stop after the failing group" + ); + } + + #[test] + fn handle_batch_sync_reports_grouped_success_and_rejection_counts() { + let registry = test_registry(); + let limiter = MockRateLimiter { + should_exceed: false, + }; + let ec_id_a = format!("{}.ABC123", "a".repeat(64)); + let ec_id_b = format!("{}.ABC123", "b".repeat(64)); + let success_writer = MockWriter::new(vec![Ok(UpsertResult::Written)]); + let success_body = format!( + r#"{{"mappings":[{{"ec_id":"{ec_id_a}","partner_uid":"one","timestamp":1}},{{"ec_id":"{ec_id_a}","partner_uid":"two","timestamp":2}}]}}"# + ); + let success_response = handle_batch_sync_with_writer( + &success_writer, + ®istry, + &limiter, + authorized_batch_request(&success_body), + ) + .expect("should return success response"); + assert_eq!(success_response.status(), StatusCode::OK); + let success_body = success_response + .into_body() + .into_bytes() + .expect("should contain grouped success response"); + let success_json: serde_json::Value = serde_json::from_slice(&success_body) + .expect("should serialize grouped success response"); + assert_eq!(success_json["accepted"], 2); + assert_eq!(success_json["rejected"], 0); + + let rejected_writer = MockWriter::new(vec![Ok(UpsertResult::NotFound)]); + let rejected_body = format!( + r#"{{"mappings":[{{"ec_id":"{ec_id_b}","partner_uid":"one","timestamp":1}},{{"ec_id":"{ec_id_b}","partner_uid":"two","timestamp":2}}]}}"# + ); + let rejected_response = handle_batch_sync_with_writer( + &rejected_writer, + ®istry, + &limiter, + authorized_batch_request(&rejected_body), + ) + .expect("should return multi-status response"); + assert_eq!(rejected_response.status(), StatusCode::MULTI_STATUS); + let rejected_body = rejected_response + .into_body() + .into_bytes() + .expect("should contain grouped multi-status response"); + let rejected_json: serde_json::Value = serde_json::from_slice(&rejected_body) + .expect("should serialize grouped multi-status response"); + assert_eq!(rejected_json["accepted"], 0); + assert_eq!(rejected_json["rejected"], 2); + assert_eq!( + rejected_json["errors"], + serde_json::json!([ + {"index": 0, "reason": REASON_INELIGIBLE}, + {"index": 1, "reason": REASON_INELIGIBLE}, + ]) + ); + } + + #[test] + fn handle_batch_sync_reports_validation_errors_without_writer_calls() { + let writer = MockWriter::new(vec![]); + let registry = test_registry(); + let limiter = MockRateLimiter { + should_exceed: false, + }; + let valid_ec_id = format!("{}.ABC123", "a".repeat(64)); + let body = format!( + r#"{{"mappings":[{{"ec_id":"invalid","partner_uid":"one","timestamp":1}},{{"ec_id":"{valid_ec_id}","partner_uid":"","timestamp":2}}]}}"# + ); + + let response = handle_batch_sync_with_writer( + &writer, + ®istry, + &limiter, + authorized_batch_request(&body), + ) + .expect("should return validation response"); + + assert_eq!(response.status(), StatusCode::MULTI_STATUS); + let response = response_json(response); + assert_eq!(response["accepted"], 0); + assert_eq!(response["rejected"], 2); + assert_eq!( + response["errors"], + serde_json::json!([ + {"index": 0, "reason": REASON_INVALID_EC_ID}, + {"index": 1, "reason": REASON_INVALID_PARTNER_UID}, + ]) + ); + assert!( + writer.calls().is_empty(), + "invalid-only requests should not call the writer" + ); + } + + #[test] + fn handle_batch_sync_reports_grouped_infrastructure_failure() { + let writer = MockWriter::new(vec![ + Ok(UpsertResult::Written), + Err(Report::new(TrustedServerError::KvStore { + store_name: "ec_store".to_owned(), + message: "down".to_owned(), + })), + ]); + let registry = test_registry(); + let limiter = MockRateLimiter { + should_exceed: false, + }; + let ec_id_a = format!("{}.ABC123", "a".repeat(64)); + let ec_id_b = format!("{}.ABC123", "b".repeat(64)); + let ec_id_c = format!("{}.ABC123", "c".repeat(64)); + let body = format!( + r#"{{"mappings":[{{"ec_id":"{ec_id_a}","partner_uid":"a-first","timestamp":1}},{{"ec_id":"{ec_id_b}","partner_uid":"b","timestamp":2}},{{"ec_id":"{ec_id_a}","partner_uid":"a-last","timestamp":3}},{{"ec_id":"{ec_id_c}","partner_uid":"c","timestamp":4}}]}}"# + ); + + let response = handle_batch_sync_with_writer( + &writer, + ®istry, + &limiter, + authorized_batch_request(&body), + ) + .expect("should return infrastructure failure response"); + + assert_eq!(response.status(), StatusCode::MULTI_STATUS); + let response = response_json(response); + assert_eq!(response["accepted"], 2); + assert_eq!(response["rejected"], 2); + assert_eq!( + response["errors"], + serde_json::json!([ + {"index": 1, "reason": REASON_KV_UNAVAILABLE}, + {"index": 3, "reason": REASON_KV_UNAVAILABLE}, + ]) + ); + assert_eq!( + writer.calls(), + vec![ + WriterCall { + ec_id: ec_id_a, + partner_id: "ssp.example.com".to_owned(), + uid: "a-last".to_owned(), + }, + WriterCall { + ec_id: ec_id_b, + partner_id: "ssp.example.com".to_owned(), + uid: "b".to_owned(), + }, + ], + "should stop after the failing group and accept A's later duplicate" + ); } } diff --git a/crates/trusted-server-core/src/ec/kv.rs b/crates/trusted-server-core/src/ec/kv.rs index 667bd0d6..846f63e4 100644 --- a/crates/trusted-server-core/src/ec/kv.rs +++ b/crates/trusted-server-core/src/ec/kv.rs @@ -1446,6 +1446,30 @@ mod tests { assert_eq!(second, UpsertResult::Unchanged); } + #[test] + fn upsert_partner_id_if_exists_retries_cas_conflict() { + let graph = KvIdentityGraph::new(ConflictInjectingEcKv::new(1, false)); + let ec_id = format!("{}.ABC123", "a".repeat(64)); + graph + .create(&ec_id, &live_entry()) + .expect("should seed live entry"); + + let result = graph + .upsert_partner_id_if_exists(&ec_id, "ssp_x", "uid-1") + .expect("should retry and write after one generation conflict"); + + assert_eq!(result, UpsertResult::Written); + let (entry, _) = graph + .get(&ec_id) + .expect("should read persisted entry") + .expect("should retain entry after conflict"); + assert_eq!( + entry.ids.get("ssp_x").map(|id| id.uid.as_str()), + Some("uid-1"), + "should persist requested UID after retry" + ); + } + #[test] fn upsert_partner_id_if_exists_rejects_tombstone() { let kv = KvIdentityGraph::in_memory("test_store"); diff --git a/docs/guide/api-reference.md b/docs/guide/api-reference.md index 9195138a..909177c3 100644 --- a/docs/guide/api-reference.md +++ b/docs/guide/api-reference.md @@ -157,10 +157,36 @@ re-evaluated within the recheck window). ### POST /\_ts/api/v1/batch-sync -Server-to-server batch sync endpoint for writing EC ID to partner UID mappings. Mapping timestamps are retained in the request schema for compatibility, but they no longer order writes because EC identity entries do not store per-partner sync timestamps. Valid mappings use idempotent last-write-wins semantics. +Server-to-server batch sync endpoint for writing EC ID to partner UID mappings. Mapping timestamps are retained in the request schema for compatibility, but they no longer order writes because EC identity entries do not store per-partner sync timestamps. **Auth:** Bearer token (`Authorization: Bearer `) +**Batch processing behavior:** + +- Every mapping is validated before any KV update. Validation errors retain their + original input index. +- Valid mappings are grouped by normalized EC ID: only the 64-character hex + prefix is lowercased; the six-character suffix remains case-sensitive. +- Groups are processed in first-valid-occurrence order, with one call to the + CAS-protected KV update path per distinct normalized EC ID. Within a group, the + last valid `partner_uid` in request order is persisted. An invalid mapping + never replaces a group's final UID. +- A successful or unchanged update accepts every valid mapping in its group. + Missing and withdrawn EC entries reject every valid mapping in their group as + `ineligible`. +- If a KV infrastructure failure occurs, every valid mapping in the failing + group and each unprocessed valid group is rejected as `kv_unavailable`; no + later group is updated. Already processed groups keep their outcomes, and + validation errors are preserved. +- `errors` is sorted by original input index. Therefore each input has exactly + one outcome and `accepted + rejected` equals the number of submitted + mappings. The endpoint returns `200 OK` only when all mappings are accepted; + otherwise it returns `207 Multi-Status`. + +Groupwise failure behavior is intentional: for `A(valid), B(valid), A(valid)`, +if A's group succeeds and B's group has an infrastructure failure, both A +mappings are accepted even though the second A appears after B in the input. + **Request Body:** ```json diff --git a/docs/guide/ec-setup-guide.md b/docs/guide/ec-setup-guide.md index afd962b6..1422b572 100644 --- a/docs/guide/ec-setup-guide.md +++ b/docs/guide/ec-setup-guide.md @@ -93,7 +93,7 @@ Look for: Endpoint: `POST /_ts/api/v1/batch-sync` -Important: request field is `ec_id` (full `{64hex}.{6alnum}` value). The `timestamp` field remains required for API compatibility, but it no longer orders writes because EC identity entries do not store per-partner sync timestamps. Valid mappings are idempotent last-write-wins: unchanged UIDs are accepted without a write, and different UIDs replace the stored value. +Important: request field is `ec_id` (full `{64hex}.{6alnum}` value). The `timestamp` field remains required for API compatibility, but it no longer orders writes because EC identity entries do not store per-partner sync timestamps. Within one request, valid mappings are grouped by normalized EC ID and the last valid UID for each group is applied once; unchanged UIDs are accepted without a write. Group outcomes are reported for every original mapping, and infrastructure failures abort the remaining groups. See the [API Reference](/guide/api-reference) for the complete accounting and failure contract. ```bash BATCH_UID="${PARTNER_UID}-batch" diff --git a/docs/guide/edge-cookies.md b/docs/guide/edge-cookies.md index ce060a60..5442a9f2 100644 --- a/docs/guide/edge-cookies.md +++ b/docs/guide/edge-cookies.md @@ -261,7 +261,7 @@ Configure EC settings in `trusted-server.toml`. See the full [Configuration Refe - Newly generated ECs receive `Set-Cookie: ts-ec=...`. - When consent is blocked but not explicitly withdrawn, Trusted Server strips EC response headers for that request but leaves any existing `ts-ec` cookie intact; cookie expiry and tombstones happen only on explicit withdrawal. - `/_ts/api/v1/identify` is read-oriented and returns identity enrichment for the authenticated partner. It computes `cluster_size` only when the EC entry does not already store one. -- `/_ts/api/v1/batch-sync` writes mappings into the EC identity graph. Mapping timestamps are retained for API compatibility but no longer order writes; valid mappings use idempotent last-write-wins semantics. +- `/_ts/api/v1/batch-sync` validates every input, groups valid mappings by normalized EC ID, and applies the last valid UID for each group once. Group outcomes still account for every original input; infrastructure failures reject the failing and remaining groups. Mapping timestamps remain required for API compatibility but do not order writes. See the [API Reference](/guide/api-reference) for the complete contract. - Pull sync fills missing partner UIDs only. Existing partner UIDs are not periodically refreshed because EC entries no longer store per-partner sync timestamps. ## Next Steps diff --git a/docs/superpowers/plans/2026-07-13-issue-882-group-batch-sync-by-ec-id.md b/docs/superpowers/plans/2026-07-13-issue-882-group-batch-sync-by-ec-id.md new file mode 100644 index 00000000..f3f78345 --- /dev/null +++ b/docs/superpowers/plans/2026-07-13-issue-882-group-batch-sync-by-ec-id.md @@ -0,0 +1,259 @@ +# Issue #882 — Group Batch Sync by EC ID Implementation Plan + +## Metadata + +- **Issue:** [#882 — Group S2S batch-sync mappings by EC ID before KV updates](https://github.com/IABTechLab/trusted-server/issues/882) +- **Stack base:** Draft PR #901, branch `fix/idempotent-ec-withdrawal-tombstones` +- **Implementation branch:** `perf/group-batch-sync-by-ec-id` +- **Status:** Implemented and verified +- **Date:** 2026-07-13 + +## Goal + +Reduce S2S batch-sync KV work from one read-modify-write per valid mapping to +one call of the existing CAS-protected writer per distinct valid normalized EC +ID, while preserving deterministic last-valid-write-wins behavior and complete +per-input response accounting. + +## Approved Behavioral Contract + +1. Validate every input before making any writer call. +2. Group only valid mappings by normalized EC ID. Normalization continues to + lowercase only the 64-character hash prefix; the six-character suffix + remains case-sensitive. +3. Process groups in the order of their first valid input occurrence. A lookup + map may locate groups, but only an insertion-ordered vector may determine + writer-call order. +4. The last valid mapping for a group in request order supplies the one partner + UID sent to the writer. Invalid mappings never join a group or replace its + final UID. The required `timestamp` field remains non-ordering. +5. Call the existing `upsert_partner_id_if_exists` writer once per group in the + uncontended normal path. Its existing bounded CAS loop remains responsible + for same-key conflicts. +6. `Written` and `Unchanged` accept every valid input in the group. + `NotFound` and `ConsentWithdrawn` reject every valid input in the group as + `ineligible`. +7. On infrastructure failure, reject every input in the failing group and all + unprocessed valid groups as `kv_unavailable`, then stop making writer calls. + Already processed groups retain their outcomes, and prevalidated invalid + inputs retain their specific errors. +8. Sort all errors by original input index before responding. Exactly one + outcome must exist per input, so `accepted + rejected == mappings.len()`. +9. Preserve response fields, reason strings, and status behavior: `200 OK` when + all mappings are accepted and `207 Multi-Status` when any are rejected. + +The group boundary intentionally supersedes the old positional abort boundary. +For inputs `A(valid), B(valid), A(valid)`, if group A succeeds and group B then +fails, both A inputs are accepted even though the second A appears after B's +first input. + +## Existing Behavior and Constraints + +- `process_mappings` currently validates and writes sequentially, causing a + writer call for every valid mapping and making duplicate EC IDs contend with + earlier writes from the same request. +- `KvIdentityGraph::upsert_partner_id_if_exists` already implements existing-key + only behavior, tombstone rejection, unchanged-value detection, and bounded + CAS retries. Grouping must reuse it rather than introduce a bulk KV API. +- The request limit is 1,000 mappings. An owned group vector plus an EC-ID to + group-index lookup map is bounded and appropriate for this limit. +- Authentication, rate limiting, request body limits, request/response schemas, + and adapter routing are outside this issue. + +## File Map + +### Modify + +- `crates/trusted-server-core/src/ec/batch_sync.rs` + - Add deterministic validation/grouping and fan-out processing. + - Upgrade the writer mock to record calls. + - Add grouped ordering, accounting, outcome, abort, and HTTP tests. + - Document the internal grouped contract. +- `crates/trusted-server-core/src/ec/kv.rs` + - Add test-only coverage proving the unchanged conditional writer retries a + generation conflict and persists the requested UID. +- `docs/guide/api-reference.md` + - Document validation-before-write, grouping, last-valid-wins, outcome fan-out, + error ordering, and infrastructure-abort semantics. + +### Add + +- `docs/superpowers/plans/2026-07-13-issue-882-group-batch-sync-by-ec-id.md` + - Record the reviewed plan and verification contract. + +No dependency, configuration, adapter, JavaScript, public schema, KV API, or +normalization change is expected. + +## Implementation Tasks + +### Task 1 — Establish failing grouped-processing tests + +- [x] Extend `MockWriter` to record ordered `(ec_id, partner_id, uid)` calls + while retaining queued outcomes. +- [x] Add duplicate and hash-prefix case-variant tests proving one writer call + per normalized EC ID. +- [x] Add an interleaved-group test proving first-valid-occurrence processing + order independent of lookup-map iteration order. +- [x] Add conflicting-UID coverage proving the last valid input wins and an + invalid duplicate cannot override it. +- [x] Add group fan-out tests for `Written`, `Unchanged`, `NotFound`, and + `ConsentWithdrawn`. +- [x] Run `cargo test-fastly batch_sync` and confirm the new call-count and + last-valid-wins tests fail before implementation. + +### Task 2 — Implement validation and deterministic grouping + +- [x] Add a private group representation containing the normalized EC ID, final + valid partner UID, and all valid original indexes. +- [x] Validate the complete request before any writer call, retaining validation + errors at their original indexes. +- [x] Maintain a lookup map only for group location and a separate vector as the + authoritative first-occurrence order. +- [x] Process each group once through the existing `BatchSyncWriter`. +- [x] Fan out success and eligibility outcomes to every valid group member. +- [x] Sort all errors by original input index before returning. +- [x] Assert complete accounting in grouped tests. + +### Task 3 — Preserve and test infrastructure-abort semantics + +- [x] Add mixed validation/success/failure tests proving validation errors are + retained and the failing plus unprocessed groups become `kv_unavailable`. +- [x] Prove no writer calls occur after the failing group. +- [x] Add the explicit `A, B, A` test proving a successful early group accepts + its later-positioned duplicate when B fails. +- [x] Assert final errors are sorted by input index and every input receives + exactly one outcome. +- [x] Keep infrastructure failures encoded in the response rather than returned + as a handler error. + +### Task 4 — Verify the unchanged CAS boundary and HTTP compatibility + +- [x] Add a test using `ConflictInjectingEcKv` that seeds a live entry, injects + one generation conflict, and proves `upsert_partner_id_if_exists` retries, + returns `Written`, and persists the requested UID. +- [x] Do not modify production KV behavior, retry limits, or backend APIs. +- [x] Add handler-level grouped tests for `200 OK`, `207 Multi-Status`, serialized + counts/errors, and `accepted + rejected == mappings.len()`. +- [x] Retain existing auth, rate-limit, body-size, and timestamp tests. + +### Task 5 — Document the public contract + +- [x] Update module/function commentary in `batch_sync.rs`. +- [x] Expand the batch-sync API reference with normalized grouping, + first-occurrence processing, last-valid-wins, group outcome fan-out, + sorted errors, and groupwise infrastructure abort behavior. +- [x] Explicitly document the intentional later-positioned duplicate case. +- [x] Keep the required-but-non-ordering timestamp guidance. + +### Task 6 — Review and full verification + +- [x] Run independent correctness/performance and test-quality reviews. +- [x] Apply only fixes required by issue scope. +- [x] Mark this plan implemented only after every check below passes. + +## Acceptance Mapping + +| Issue requirement | Planned evidence | +| ------------------------------------------------ | ------------------------------------------------------------------ | +| Work bounded by distinct valid normalized EC IDs | Recording-writer call count with duplicate and case-variant inputs | +| Validation errors retain indexes | Mixed validation/group processing test | +| Last valid UID wins | Conflicting UID test including an invalid duplicate | +| Missing and withdrawn fan out consistently | Duplicate `NotFound` and `ConsentWithdrawn` tests | +| Accepted/rejected counts remain compatible | Unit and handler complete-accounting assertions | +| Deterministic processing | Interleaved groups and recorded call order | +| Infrastructure abort is explicit | Failure test covering prior, failing, and unprocessed groups | +| Errors remain in input order | Mixed validation/eligibility/infrastructure sorted-error assertion | +| CAS conflicts remain safe | Real conditional-writer conflict-injection test | +| Contract is documented | Updated module docs and API reference | + +## Test Design Notes + +- Use at least two distinct normalized EC IDs when queuing distinct mock outcomes; + duplicate inputs now deliberately consume only one outcome. +- Test uppercase/lowercase variants only in the hash prefix. Do not imply the + suffix is case-insensitive. +- Include invalid EC ID and invalid UID mappings before and after valid members + to expose accidental positional processing and final-UID replacement. +- Record writer arguments, not only call count, to prove normalized keys, + partner identity, final UID, and deterministic group order. +- For every processing test, either directly assert or share a helper asserting + `accepted + errors.len() == mappings.len()`. +- The CAS test covers retries inside the existing writer. The batch writer trait + must still be called once for that EC group; backend read/write attempts are + intentionally not equated with trait-call count. + +## Verification Contract + +Run focused checks while implementing: + +```bash +cargo test-fastly batch_sync +cargo test-fastly upsert_partner_id_if_exists_retries_cas_conflict +cargo fmt --all -- --check +cd docs && npx prettier --check guide/api-reference.md superpowers/plans/2026-07-13-issue-882-group-batch-sync-by-ec-id.md +``` + +Run the complete PR gates before marking the plan verified: + +```bash +cargo fmt --all -- --check + +cargo test-fastly +cargo test-axum +cargo test-cloudflare +cargo test-spin +cargo test --manifest-path crates/trusted-server-integration-tests/Cargo.toml --test parity + +cargo clippy-fastly +cargo clippy-axum +cargo clippy-cloudflare +cargo clippy-cloudflare-wasm +cargo clippy-spin-native +cargo clippy-spin-wasm + +cd crates/trusted-server-js/lib +npx vitest run +npm run format +cd ../../.. + +cd docs +npm run format +cd .. + +cargo build --package trusted-server-adapter-fastly --release --target wasm32-wasip1 +git diff --check +``` + +## Risks and Mitigations + +- **Nondeterministic map iteration:** never process groups by iterating the + lookup map; use the insertion-ordered vector. +- **Incorrect normalization:** continue using `normalize_ec_id_for_kv`; do not + lowercase the suffix. +- **Invalid mapping overrides final UID:** update a group's UID only after both + EC ID and partner UID validation pass. +- **Double or missing accounting:** fan out exactly once per valid group member, + retain exactly one validation error per invalid input, and assert totals. +- **Abort fan-out overwrites validation:** unprocessed fan-out traverses only + already validated group member indexes. +- **Mock masks excess calls:** make missing queued results fail and explicitly + assert recorded calls. +- **Scope creep into KV design:** production `kv.rs` is unchanged; only add a + regression test for its existing retry contract. + +## Non-Goals + +- No new KV bulk API, cross-key transaction, parallel group processing, or + retry policy. +- No coalescing across separate HTTP requests. +- No change to authentication, rate limits, body limits, maximum batch size, + timestamps, wire schemas, reason strings, or EC normalization rules. +- No change to root creation, tombstone, or consent semantics. + +## Review and Verification Record + +- Independent plan review: approved (no blockers) +- Focused tests: passed (22 batch-sync tests plus CAS conflict regression) +- Independent code review: approved after resolving all test/documentation findings +- Full cross-adapter verification: passed +- Release WASM build: passed