From 7e2c7e5920a4a7cbbc80ead1e887795b1f454516 Mon Sep 17 00:00:00 2001
From: prk-Jr <prakashsagwara@gmail.com>
Date: Tue, 1 Sep 2026 20:36:02 +0530
Subject: [PATCH 01/11] Harden the Edge Cookie withdrawal write path

Tombstone only an identity the graph already holds. The marker exists to
stop later reads of a real row, so writing one for an identifier that was
never issued enforces nothing while still consuming a write and a row, and
the identifier arrives in a client-supplied cookie.

Confirm existence with the list API rather than a lookup. A lookup is
eventually consistent, so a stale miss would discard a genuine withdrawal;
the list is strongly consistent. Reject anything that is not a well-formed
EC ID before querying, since this is a prefix query and an empty or
truncated value would match unrelated keys.

When the list cannot answer, re-check with a lookup instead of writing
regardless. Eventual consistency yields false negatives, never false
positives, so a hit is proof the identity exists while no fabricated
identifier can produce one. If neither can answer, report the withdrawal
unconfirmed and write nothing; the browser cookie is expired either way and
remains the primary enforcement.

Split the unusable-consent branch out of ec_finalize_response and route the
per-identity result through one place, so an unconfirmed identity is logged
as a fault while an unknown one is not.
---
 crates/trusted-server-core/src/ec/admin.rs    |  17 +-
 crates/trusted-server-core/src/ec/finalize.rs | 194 +++++++++--
 crates/trusted-server-core/src/ec/kv.rs       | 309 +++++++++++++++++-
 3 files changed, 483 insertions(+), 37 deletions(-)

diff --git a/crates/trusted-server-core/src/ec/admin.rs b/crates/trusted-server-core/src/ec/admin.rs
index 6219af7a9..38033c533 100644
--- a/crates/trusted-server-core/src/ec/admin.rs
+++ b/crates/trusted-server-core/src/ec/admin.rs
@@ -677,6 +677,8 @@ mod tests {
 
     use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64};
 
+    use crate::ec::kv::TombstoneOutcome;
+
     use super::*;
     use crate::ec::kv_backend::test_support::InMemoryEcKv;
     use crate::ec::kv_backend::{EcKvStore as _, EcKvWrite, EcKvWriteMode};
@@ -1114,9 +1116,18 @@ mod tests {
     #[test]
     fn reports_tombstone_entries() {
         let ec_id = test_ec_id();
-        let kv = KvIdentityGraph::in_memory("test-store");
-        kv.write_withdrawal_tombstone(&ec_id)
-            .expect("should write tombstone");
+        // Only an identity the store already holds can be tombstoned, so seed
+        // the live entry the withdrawal replaces.
+        let kv = kv_with_entry(
+            &ec_id,
+            &KvEntry::minimal("bidstream.example", "uid-live", 1_741_824_000),
+        );
+        assert_eq!(
+            kv.write_withdrawal_tombstone(&ec_id)
+                .expect("should write tombstone"),
+            TombstoneOutcome::Written,
+            "should tombstone the seeded identity"
+        );
         let req = get_request(&format!("/_ts/admin/ec/{ec_id}"));
 
         let response = handle_admin_ec_lookup(Some(&kv), &test_registry(), &req)
diff --git a/crates/trusted-server-core/src/ec/finalize.rs b/crates/trusted-server-core/src/ec/finalize.rs
index a553bb7a7..607ce690a 100644
--- a/crates/trusted-server-core/src/ec/finalize.rs
+++ b/crates/trusted-server-core/src/ec/finalize.rs
@@ -6,15 +6,17 @@
 use std::collections::HashSet;
 
 use edgezero_core::body::Body as EdgeBody;
+use error_stack::Report;
 use http::Response;
 
 use super::consent::{ec_consent_granted, ec_consent_withdrawn};
+use crate::error::TrustedServerError;
 use crate::settings::Settings;
 
 use super::EcContext;
 use super::cookies::{expire_ec_cookie, set_ec_cookie};
 use super::generation::is_valid_ec_id;
-use super::kv::KvIdentityGraph;
+use super::kv::{KvIdentityGraph, TombstoneOutcome};
 use super::log_id;
 use super::prebid_eids::ingest_eid_cookies;
 use super::registry::PartnerRegistry;
@@ -51,34 +53,14 @@ pub fn ec_finalize_response(
     let consent_withdrawn = ec_consent_withdrawn(ec_context.consent());
 
     if !consent_allows_ec {
-        // Always strip EC-specific response headers when consent is not
-        // currently usable for this request. This covers both explicit
-        // revocation and fail-closed cases such as missing geo or undecodable
-        // consent input.
-        clear_ec_headers_on_response(response, Some(registry));
-
-        // Only expire the browser cookie and tombstone the identity-graph row
-        // when the request carries an explicit withdrawal signal.
-        if consent_withdrawn && ec_context.cookie_was_present() {
-            expire_ec_cookie(settings, response);
-
-            // Compute once for the authoritative identity-graph tombstones.
-            let ids_to_withdraw = withdrawal_ec_ids(ec_context);
-
-            // The identity-graph tombstone is the authoritative withdrawal marker
-            // 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),
-                        );
-                    }
-                });
-            }
-        }
-
+        finalize_unusable_consent(
+            settings,
+            ec_context,
+            kv,
+            registry,
+            consent_withdrawn,
+            response,
+        );
         return;
     }
 
@@ -152,6 +134,79 @@ pub fn clear_ec_on_response(settings: &Settings, response: &mut Response<EdgeBod
     clear_ec_headers_on_response(response, None);
 }
 
+/// Finalizes a response whose consent does not currently permit an EC.
+///
+/// Covers explicit revocation and fail-closed cases alike, such as missing geo
+/// or undecodable consent input: EC response headers always come off. The
+/// browser cookie is expired and the identity-graph row tombstoned only when
+/// the request carries an explicit withdrawal signal, so a visitor who has
+/// simply not decided yet is not stripped of an identity they already hold.
+fn finalize_unusable_consent(
+    settings: &Settings,
+    ec_context: &EcContext,
+    kv: Option<&KvIdentityGraph>,
+    registry: &PartnerRegistry,
+    consent_withdrawn: bool,
+    response: &mut Response<EdgeBody>,
+) {
+    clear_ec_headers_on_response(response, Some(registry));
+
+    if !(consent_withdrawn && ec_context.cookie_was_present()) {
+        return;
+    }
+
+    expire_ec_cookie(settings, response);
+
+    // Compute once for the authoritative identity-graph tombstones.
+    let ids_to_withdraw = withdrawal_ec_ids(ec_context);
+
+    // The identity-graph tombstone is the authoritative withdrawal marker
+    // for subsequent EC behavior.
+    if let Some(graph) = kv {
+        apply_withdrawal_tombstones(&ids_to_withdraw, |ec_id| {
+            log_tombstone_outcome(ec_id, graph.write_withdrawal_tombstone(ec_id));
+        });
+    }
+}
+
+/// Records what happened to one withdrawal tombstone.
+///
+/// An unknown identity is expected traffic rather than a fault: the identifier
+/// comes from a client-supplied cookie, so it may name something this
+/// deployment never issued. An unconfirmed identity is different: the store
+/// could not answer, so a real row may have gone unmarked and that is logged as
+/// an error. The browser cookie is expired in every case, and that is the
+/// primary enforcement.
+fn log_tombstone_outcome(
+    ec_id: &str,
+    outcome: Result<TombstoneOutcome, Report<TrustedServerError>>,
+) {
+    match outcome {
+        Ok(TombstoneOutcome::Written) => {}
+        Ok(TombstoneOutcome::UnknownIdentity) => {
+            log::debug!(
+                "Skipping withdrawal tombstone for unknown EC ID '{}'",
+                log_id(ec_id),
+            );
+        }
+        Ok(TombstoneOutcome::Unconfirmed) => {
+            // Unlike an unknown identity, this is a fault worth surfacing: the
+            // store could not be read, so a real identity may have gone
+            // untombstoned for the batch-sync window.
+            log::error!(
+                "Could not confirm EC ID '{}' to tombstone it; the browser cookie is still expired",
+                log_id(ec_id),
+            );
+        }
+        Err(err) => {
+            log::error!(
+                "Failed to write withdrawal tombstone for EC ID '{}': {err:?}",
+                log_id(ec_id),
+            );
+        }
+    }
+}
+
 fn withdrawal_ec_ids(ec_context: &EcContext) -> HashSet<String> {
     let mut hashes = HashSet::new();
 
@@ -391,6 +446,87 @@ mod tests {
         );
     }
 
+    #[test]
+    fn finalize_withdrawal_does_not_create_a_row_for_an_unheld_identity() {
+        let settings = create_test_settings();
+        // The cookie value is chosen by the client, so a withdrawal naming an
+        // identity this deployment never issued must not put a row in the
+        // identity graph.
+        let ec_id = sample_ec_id("zz9999");
+        let consent = ConsentContext {
+            jurisdiction: Jurisdiction::UsState("CA".to_owned()),
+            gpc: true,
+            source: ConsentSource::Cookie,
+            ..Default::default()
+        };
+        let ec_context =
+            make_context_with_consent(Some(&ec_id), Some(&ec_id), true, false, consent);
+        let kv = KvIdentityGraph::in_memory("test-store");
+        let mut response = empty_response();
+        let registry = PartnerRegistry::from_config(&[]).expect("should build registry");
+
+        ec_finalize_response(
+            &settings,
+            &ec_context,
+            Some(&kv),
+            &registry,
+            None,
+            None,
+            &mut response,
+        );
+
+        assert!(
+            kv.get(&ec_id).expect("should read back").is_none(),
+            "should not write a tombstone for an identity that was never issued"
+        );
+        let set_cookie = get_header_str(&response, "set-cookie").unwrap_or_default();
+        assert!(
+            set_cookie.contains("Max-Age=0"),
+            "should still expire the browser cookie, which is the primary enforcement"
+        );
+    }
+
+    #[test]
+    fn finalize_withdrawal_tombstones_a_held_identity() {
+        let settings = create_test_settings();
+        let ec_id = sample_ec_id("held01");
+        let consent = ConsentContext {
+            jurisdiction: Jurisdiction::UsState("CA".to_owned()),
+            gpc: true,
+            source: ConsentSource::Cookie,
+            ..Default::default()
+        };
+        let ec_context =
+            make_context_with_consent(Some(&ec_id), Some(&ec_id), true, false, consent);
+        let kv = KvIdentityGraph::in_memory("test-store");
+        kv.create(
+            &ec_id,
+            &crate::ec::kv_types::KvEntry::minimal("p.example", "uid", 1),
+        )
+        .expect("should seed the identity");
+        let mut response = empty_response();
+        let registry = PartnerRegistry::from_config(&[]).expect("should build registry");
+
+        ec_finalize_response(
+            &settings,
+            &ec_context,
+            Some(&kv),
+            &registry,
+            None,
+            None,
+            &mut response,
+        );
+
+        let (entry, _) = kv
+            .get(&ec_id)
+            .expect("should read back")
+            .expect("should still hold the identity");
+        assert!(
+            !entry.consent.ok,
+            "a genuine withdrawal must still tombstone the identity"
+        );
+    }
+
     #[test]
     fn finalize_withdrawal_clears_cookie_and_headers() {
         let settings = create_test_settings();
diff --git a/crates/trusted-server-core/src/ec/kv.rs b/crates/trusted-server-core/src/ec/kv.rs
index 3572581ce..32f24c967 100644
--- a/crates/trusted-server-core/src/ec/kv.rs
+++ b/crates/trusted-server-core/src/ec/kv.rs
@@ -20,7 +20,7 @@ use error_stack::{Report, ResultExt};
 use crate::error::TrustedServerError;
 
 use super::current_timestamp;
-use super::generation::ec_hash;
+use super::generation::{ec_hash, is_valid_ec_id};
 use super::kv_backend::{EcKvLookup, EcKvStore, EcKvWrite, EcKvWriteMode, EcKvWriteOutcome};
 use super::kv_types::{KvEntry, KvMetadata, KvNetwork};
 use super::log_id;
@@ -123,6 +123,19 @@ impl fmt::Debug for KvIdentityGraph {
     }
 }
 
+/// Result of [`KvIdentityGraph::write_withdrawal_tombstone`].
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum TombstoneOutcome {
+    /// The identity was found and is now tombstoned.
+    Written,
+    /// No such identity is held, so there was nothing to mark withdrawn.
+    UnknownIdentity,
+    /// The store could not say whether the identity exists, so nothing was
+    /// written. The browser cookie is still expired by the caller, which is the
+    /// primary enforcement; only the batch-sync revocation marker is missing.
+    Unconfirmed,
+}
+
 impl KvIdentityGraph {
     /// Creates a new identity graph backed by the given store primitives.
     #[must_use]
@@ -621,6 +634,31 @@ impl KvIdentityGraph {
         )))
     }
 
+    /// Whether `ec_id` names a key this store actually holds.
+    ///
+    /// Uses the list API rather than [`Self::get`] deliberately. A lookup is
+    /// eventually consistent, so it can answer "missing" for an entry that was
+    /// just written; the list is strongly consistent and will not. That matters
+    /// wherever a false "missing" would discard something, because the identity
+    /// would be treated as one this deployment never issued.
+    ///
+    /// An EC ID is a fixed-width `{64 hex}.{6 alphanumeric}` string, so no other
+    /// key can carry a whole EC ID as a prefix and the count is exact. A value
+    /// that is not a well-formed EC ID is one this store can never hold, and is
+    /// reported missing without a store round trip. That check is not merely an
+    /// optimisation: this is a prefix query, so an empty or truncated value
+    /// would match unrelated keys, and an empty one would match every key.
+    ///
+    /// # Errors
+    ///
+    /// Returns [`TrustedServerError::KvStore`] on store error.
+    fn key_exists_confirmed(&self, ec_id: &str) -> Result<bool, Report<TrustedServerError>> {
+        if !is_valid_ec_id(ec_id) {
+            return Ok(false);
+        }
+        Ok(self.store.count_keys_with_prefix(ec_id, 1)? > 0)
+    }
+
     /// Writes a withdrawal tombstone for consent enforcement.
     ///
     /// Overwrites the entry with `consent.ok = false`, empty partner IDs,
@@ -630,6 +668,27 @@ impl KvIdentityGraph {
     /// The tombstone preserves consent enforcement for batch sync clients
     /// (`POST /_ts/api/v1/batch-sync`) during the 24-hour revocation window.
     ///
+    /// Only an identity this store already holds is tombstoned. The marker
+    /// exists to stop later reads of a real row, so writing one for an ID that
+    /// was never issued enforces nothing while still consuming a write and a
+    /// row; the identifier in a request is chosen by the client, so that write
+    /// would be the client's to trigger at will. Existence is confirmed with
+    /// [`Self::key_exists_confirmed`] so a freshly issued identity is never
+    /// mistaken for an unknown one.
+    ///
+    /// The check and the write are not one atomic operation: an entry that
+    /// expires between them is still tombstoned, briefly restoring a row that
+    /// had gone. That is deliberate — the write stays unconditional so a
+    /// withdrawal is not lost to a concurrent update — and it cannot be used to
+    /// create an identity, because the entry must have existed to pass the
+    /// check at all.
+    ///
+    /// When the list cannot answer, existence is re-checked with a lookup, which
+    /// cannot report an identity the store does not hold. If that cannot answer
+    /// either, nothing is written and [`TombstoneOutcome::Unconfirmed`] is
+    /// returned; the caller still expires the browser cookie, which is the
+    /// primary enforcement.
+    ///
     /// # Errors
     ///
     /// Returns [`TrustedServerError::KvStore`] on store error. Callers on
@@ -638,7 +697,30 @@ impl KvIdentityGraph {
     pub fn write_withdrawal_tombstone(
         &self,
         ec_id: &str,
-    ) -> Result<(), Report<TrustedServerError>> {
+    ) -> Result<TombstoneOutcome, Report<TrustedServerError>> {
+        match self.key_exists_confirmed(ec_id) {
+            Ok(true) => {}
+            Ok(false) => return Ok(TombstoneOutcome::UnknownIdentity),
+            Err(list_error) => {
+                // The list could not answer. Fall back to a lookup rather than
+                // writing blind: a lookup is eventually consistent, so it can
+                // miss a very recent write, but it can never invent a row. A
+                // hit is therefore proof the identity exists, while no
+                // fabricated identifier can produce one. Writing blind here
+                // would instead hand a caller the original unbounded write back
+                // whenever the store can be pushed into failing.
+                match self.get(ec_id) {
+                    Ok(Some(_)) => {
+                        log::warn!(
+                            "Confirmed EC ID '{}' by lookup after a list failure: {list_error:?}",
+                            log_id(ec_id),
+                        );
+                    }
+                    Ok(None) | Err(_) => return Ok(TombstoneOutcome::Unconfirmed),
+                }
+            }
+        }
+
         let entry = KvEntry::tombstone(current_timestamp());
         let (body, meta_str) = Self::serialize_entry(&entry, self.store_name())?;
 
@@ -649,7 +731,7 @@ impl KvIdentityGraph {
             TOMBSTONE_TTL,
             EcKvWriteMode::Overwrite,
         ) {
-            Ok(_) => Ok(()),
+            Ok(_) => Ok(TombstoneOutcome::Written),
             Err(report) => Err(report.change_context(TrustedServerError::KvStore {
                 store_name: self.store_name().to_owned(),
                 message: format!("Failed to write tombstone for key '{ec_id}'"),
@@ -1264,8 +1346,12 @@ mod tests {
         let ec_id = format!("{}.ABC123", "a".repeat(64));
         kv.create(&ec_id, &live_entry()).expect("should create");
 
-        kv.write_withdrawal_tombstone(&ec_id)
-            .expect("should write tombstone");
+        assert_eq!(
+            kv.write_withdrawal_tombstone(&ec_id)
+                .expect("should write tombstone"),
+            TombstoneOutcome::Written,
+            "should tombstone an identity the store holds"
+        );
 
         let (loaded, _) = kv
             .get(&ec_id)
@@ -1273,4 +1359,217 @@ mod tests {
             .expect("should find tombstone entry");
         assert!(!loaded.consent.ok, "should be withdrawn after tombstone");
     }
+
+    #[test]
+    fn write_withdrawal_tombstone_ignores_an_identity_the_store_does_not_hold() {
+        let kv = KvIdentityGraph::in_memory("test_store");
+        let ec_id = format!("{}.ABC123", "b".repeat(64));
+
+        assert_eq!(
+            kv.write_withdrawal_tombstone(&ec_id)
+                .expect("should resolve the withdrawal"),
+            TombstoneOutcome::UnknownIdentity,
+            "an identity that was never issued has nothing to withdraw"
+        );
+        assert!(
+            kv.get(&ec_id).expect("should read back").is_none(),
+            "should not create a row for an identity the store never held"
+        );
+    }
+
+    #[test]
+    fn withdrawing_many_unheld_identities_creates_no_rows() {
+        let kv = KvIdentityGraph::in_memory("test_store");
+        let hash = "c".repeat(64);
+
+        // The suffix is caller-supplied, so a shared hash prefix must not be
+        // enough to have a row written under it.
+        for suffix in ["aaaaaa", "bbbbbb", "cccccc", "dddddd"] {
+            assert_eq!(
+                kv.write_withdrawal_tombstone(&format!("{hash}.{suffix}"))
+                    .expect("should resolve the withdrawal"),
+                TombstoneOutcome::UnknownIdentity,
+                "suffix `{suffix}` was never issued"
+            );
+        }
+
+        assert_eq!(
+            kv.count_hash_prefix_keys(&hash)
+                .expect("should count the prefix"),
+            0,
+            "should hold no rows under a hash nothing was issued for"
+        );
+    }
+
+    #[test]
+    fn key_exists_confirmed_refuses_a_value_that_is_not_an_ec_id() {
+        let kv = KvIdentityGraph::in_memory("test_store");
+        kv.create(&format!("{}.ABC123", "e".repeat(64)), &live_entry())
+            .expect("should create");
+
+        // A prefix query would match every key for an empty value and unrelated
+        // keys for a truncated one, so neither may reach the store.
+        for value in [
+            "",
+            "e".repeat(64).as_str(),
+            "not-an-ec-id",
+            "E".repeat(64).as_str(),
+        ] {
+            assert!(
+                !kv.key_exists_confirmed(value)
+                    .expect("should resolve the check"),
+                "should not report `{value}` as held"
+            );
+        }
+    }
+
+    #[test]
+    fn write_withdrawal_tombstone_refuses_a_value_that_is_not_an_ec_id() {
+        let kv = KvIdentityGraph::in_memory("test_store");
+        kv.create(&format!("{}.ABC123", "f".repeat(64)), &live_entry())
+            .expect("should create");
+
+        assert_eq!(
+            kv.write_withdrawal_tombstone("")
+                .expect("should resolve the withdrawal"),
+            TombstoneOutcome::UnknownIdentity,
+            "an empty identifier must not be treated as a match-all prefix"
+        );
+        let (held, _) = kv
+            .get(&format!("{}.ABC123", "f".repeat(64)))
+            .expect("should read back")
+            .expect("should still hold the identity");
+        assert!(
+            held.consent.ok,
+            "should not have withdrawn an unrelated row"
+        );
+    }
+
+    /// Store double whose list API always fails while writes still work.
+    struct ListFailingEcKv {
+        inner: super::super::kv_backend::test_support::InMemoryEcKv,
+    }
+
+    impl ListFailingEcKv {
+        fn new() -> Self {
+            Self {
+                inner: super::super::kv_backend::test_support::InMemoryEcKv::new("test_store"),
+            }
+        }
+    }
+
+    impl EcKvStore for ListFailingEcKv {
+        fn store_name(&self) -> &str {
+            self.inner.store_name()
+        }
+
+        fn lookup(&self, key: &str) -> Result<Option<EcKvLookup>, Report<TrustedServerError>> {
+            self.inner.lookup(key)
+        }
+
+        fn insert(
+            &self,
+            key: &str,
+            write: EcKvWrite<'_>,
+        ) -> Result<EcKvWriteOutcome, Report<TrustedServerError>> {
+            self.inner.insert(key, write)
+        }
+
+        fn count_keys_with_prefix(
+            &self,
+            _prefix: &str,
+            _limit: u32,
+        ) -> Result<u32, Report<TrustedServerError>> {
+            Err(Report::new(TrustedServerError::KvStore {
+                store_name: "test_store".to_owned(),
+                message: "list unavailable".to_owned(),
+            }))
+        }
+
+        fn delete(&self, key: &str) -> Result<(), Report<TrustedServerError>> {
+            self.inner.delete(key)
+        }
+    }
+
+    #[test]
+    fn write_withdrawal_tombstone_falls_back_to_lookup_when_the_list_fails() {
+        // A store outage must not discard a genuine withdrawal, so a failed
+        // list is re-checked with a lookup, which can still see the row.
+        let kv = KvIdentityGraph::new(ListFailingEcKv::new());
+        let ec_id = format!("{}.ABC123", "9".repeat(64));
+        kv.create(&ec_id, &live_entry()).expect("should create");
+
+        assert_eq!(
+            kv.write_withdrawal_tombstone(&ec_id)
+                .expect("should still resolve the withdrawal"),
+            TombstoneOutcome::Written,
+            "a held identity should still be tombstoned when only the list fails"
+        );
+        let (loaded, _) = kv
+            .get(&ec_id)
+            .expect("should read entry back")
+            .expect("should find the tombstone");
+        assert!(!loaded.consent.ok, "should be withdrawn");
+    }
+
+    #[test]
+    fn a_failing_list_is_not_a_way_to_write_for_an_identity_that_was_never_issued() {
+        // The caller controls the identifier and can drive load, so a store
+        // failure must not become a route to the write this gate exists to
+        // prevent. A lookup cannot report a row that does not exist.
+        let kv = KvIdentityGraph::new(ListFailingEcKv::new());
+        let ec_id = format!("{}.ABC123", "8".repeat(64));
+
+        assert_eq!(
+            kv.write_withdrawal_tombstone(&ec_id)
+                .expect("should resolve the withdrawal"),
+            TombstoneOutcome::Unconfirmed,
+            "should refuse to write when existence cannot be established"
+        );
+        assert!(
+            kv.get(&ec_id).expect("should read back").is_none(),
+            "should not create a row while the store is degraded"
+        );
+    }
+
+    #[test]
+    fn key_exists_confirmed_does_not_match_a_longer_key_by_prefix() {
+        // The check is a prefix query, sound only while every EC ID is the same
+        // width. If that ever stops holding, a shorter identifier would match a
+        // longer unrelated row and a tombstone would be written for an identity
+        // that was never issued. This test pins the assumption.
+        let kv = KvIdentityGraph::in_memory("test_store");
+        let held = format!("{}.ABC123", "7".repeat(64));
+        kv.create(&held, &live_entry()).expect("should create");
+
+        let shorter = &held[..held.len() - 1];
+        assert!(
+            !kv.key_exists_confirmed(shorter)
+                .expect("should resolve the check"),
+            "a proper prefix of a stored key is a different identity"
+        );
+        assert_eq!(
+            kv.write_withdrawal_tombstone(shorter)
+                .expect("should resolve the withdrawal"),
+            TombstoneOutcome::UnknownIdentity,
+            "should not tombstone via a prefix match"
+        );
+    }
+
+    #[test]
+    fn key_exists_confirmed_distinguishes_held_identities() {
+        let kv = KvIdentityGraph::in_memory("test_store");
+        let held = format!("{}.ABC123", "d".repeat(64));
+        let sibling = format!("{}.ZZZ999", "d".repeat(64));
+        kv.create(&held, &live_entry()).expect("should create");
+
+        assert!(
+            kv.key_exists_confirmed(&held).expect("should check"),
+            "should confirm a held identity"
+        );
+        assert!(
+            !kv.key_exists_confirmed(&sibling).expect("should check"),
+            "a different suffix under the same hash is a different identity"
+        );
+    }
 }

From 3573fb952bc3a92de3e75e9e60312261a6972cdf Mon Sep 17 00:00:00 2001
From: prk-Jr <prakashsagwara@gmail.com>
Date: Wed, 2 Sep 2026 10:00:11 +0530
Subject: [PATCH 02/11] Log why a withdrawal could not be confirmed

The degraded path dropped both the list and lookup errors, leaving a
store outage undiagnosable. Log both, and use the raw lookup so a
corrupt-but-present row is not read as absent.

Add a test pinning the fixed-width assumption the prefix check relies on.
---
 crates/trusted-server-core/src/ec/kv.rs | 67 +++++++++++++++++++++++--
 1 file changed, 62 insertions(+), 5 deletions(-)

diff --git a/crates/trusted-server-core/src/ec/kv.rs b/crates/trusted-server-core/src/ec/kv.rs
index 32f24c967..bee320973 100644
--- a/crates/trusted-server-core/src/ec/kv.rs
+++ b/crates/trusted-server-core/src/ec/kv.rs
@@ -691,9 +691,11 @@ impl KvIdentityGraph {
     ///
     /// # Errors
     ///
-    /// Returns [`TrustedServerError::KvStore`] on store error. Callers on
-    /// the browser path should log at `error` level and continue — cookie
-    /// deletion is the primary enforcement mechanism.
+    /// Returns [`TrustedServerError::KvStore`] when the tombstone write itself
+    /// fails. A failure to determine whether the identity exists is reported as
+    /// [`TombstoneOutcome::Unconfirmed`] rather than an error, since nothing was
+    /// written. Callers on the browser path should log at `error` level and
+    /// continue — cookie deletion is the primary enforcement mechanism.
     pub fn write_withdrawal_tombstone(
         &self,
         ec_id: &str,
@@ -709,14 +711,35 @@ impl KvIdentityGraph {
                 // fabricated identifier can produce one. Writing blind here
                 // would instead hand a caller the original unbounded write back
                 // whenever the store can be pushed into failing.
-                match self.get(ec_id) {
+                // The raw form is used deliberately: a row whose body no longer
+                // deserializes is still a row, and presence is all that matters
+                // here, so a corrupt entry must not read as absent.
+                match self.lookup_raw(ec_id) {
                     Ok(Some(_)) => {
                         log::warn!(
                             "Confirmed EC ID '{}' by lookup after a list failure: {list_error:?}",
                             log_id(ec_id),
                         );
                     }
-                    Ok(None) | Err(_) => return Ok(TombstoneOutcome::Unconfirmed),
+                    Ok(None) => {
+                        log::error!(
+                            "Cannot confirm EC ID '{}', so a withdrawal may go unrecorded \
+                             for the batch-sync window: the list failed and a lookup found \
+                             nothing. List error: {list_error:?}",
+                            log_id(ec_id),
+                        );
+                        return Ok(TombstoneOutcome::Unconfirmed);
+                    }
+                    Err(lookup_error) => {
+                        log::error!(
+                            "Cannot confirm EC ID '{}', so a withdrawal may go unrecorded \
+                             for the batch-sync window: neither the list nor a lookup could \
+                             be read. List error: {list_error:?}. Lookup error: \
+                             {lookup_error:?}",
+                            log_id(ec_id),
+                        );
+                        return Ok(TombstoneOutcome::Unconfirmed);
+                    }
                 }
             }
         }
@@ -1532,6 +1555,40 @@ mod tests {
         );
     }
 
+    /// Tripwire for [`KvIdentityGraph::key_exists_confirmed`].
+    ///
+    /// That check is a prefix query, and it is only exact because every value
+    /// [`is_valid_ec_id`] accepts is the same width — no accepted identifier can
+    /// be a proper prefix of another. Nothing in the type system enforces that,
+    /// so this test states the dependency: if the accepted grammar is ever
+    /// widened to variable-length identifiers, such as a provider envelope like
+    /// `{code}~value`, this fails and whoever widened it has to revisit the
+    /// prefix query rather than discovering a false positive in production.
+    #[test]
+    fn the_prefix_check_depends_on_ec_ids_being_fixed_width() {
+        let accepted = format!("{}.ABC123", "a".repeat(64));
+        assert!(is_valid_ec_id(&accepted), "the built-in shape is accepted");
+        assert_eq!(
+            accepted.len(),
+            71,
+            "an accepted identifier is a fixed 71 bytes"
+        );
+
+        // Anything longer or shorter, including a provider envelope, must be
+        // refused while the existence check relies on prefix semantics.
+        for widened in [
+            format!("hmac~{accepted}"),
+            format!("51dd~{}", "opaque-value-of-some-other-length"),
+            format!("{accepted}trailing"),
+            "a".repeat(64),
+        ] {
+            assert!(
+                !is_valid_ec_id(&widened),
+                "a variable-length identifier would break the prefix query: {widened}"
+            );
+        }
+    }
+
     #[test]
     fn key_exists_confirmed_does_not_match_a_longer_key_by_prefix() {
         // The check is a prefix query, sound only while every EC ID is the same

From 92273b155d923dbc86510feb0108ec1af43c06dc Mon Sep 17 00:00:00 2001
From: prk-Jr <prakashsagwara@gmail.com>
Date: Wed, 2 Sep 2026 10:29:50 +0530
Subject: [PATCH 03/11] Check for the exact key instead of counting a prefix

Counting keys by prefix reported a held identity whenever any longer key
started with the one asked for, so a withdrawal for an identity that was
never issued still wrote a row. Add an exact, strongly consistent
`key_exists` to the store and use it.

This also drops the dependency on the identifier grammar: the check no
longer cares what shape an identifier takes, only whether that key is
present. Redact the key in the Fastly lookup error, matching the list error.
Carry the reason for an unconfirmed withdrawal so it is logged once.
---
 .../src/ec_kv.rs                              |  27 ++-
 crates/trusted-server-core/src/ec/finalize.rs |  49 ++++-
 crates/trusted-server-core/src/ec/kv.rs       | 184 ++++++++----------
 .../trusted-server-core/src/ec/kv_backend.rs  |  20 ++
 4 files changed, 173 insertions(+), 107 deletions(-)

diff --git a/crates/trusted-server-adapter-fastly/src/ec_kv.rs b/crates/trusted-server-adapter-fastly/src/ec_kv.rs
index 229c6c2d3..697f9feba 100644
--- a/crates/trusted-server-adapter-fastly/src/ec_kv.rs
+++ b/crates/trusted-server-adapter-fastly/src/ec_kv.rs
@@ -42,6 +42,12 @@ impl FastlyEcKvStore {
     }
 }
 
+/// Keys to request when checking for one exact key.
+///
+/// Only an exact match matters, but a page of one could be filled by a longer
+/// key sharing the prefix, so a small page is requested and scanned.
+const EXACT_MATCH_LIST_LIMIT: u32 = 10;
+
 impl EcKvStore for FastlyEcKvStore {
     fn store_name(&self) -> &str {
         &self.store_name
@@ -56,7 +62,7 @@ impl EcKvStore for FastlyEcKvStore {
                 return Err(
                     Report::new(err).change_context(TrustedServerError::KvStore {
                         store_name: self.store_name.clone(),
-                        message: format!("Failed to read key '{key}'"),
+                        message: format!("Failed to read key '{}'", key.get(..8).unwrap_or(key),),
                     }),
                 );
             }
@@ -128,6 +134,25 @@ impl EcKvStore for FastlyEcKvStore {
         Ok(count)
     }
 
+    fn key_exists(&self, key: &str) -> Result<bool, Report<TrustedServerError>> {
+        let store = self.open_store()?;
+        // The list is strongly consistent, unlike a lookup, but it matches by
+        // prefix — so the returned keys are compared for equality rather than
+        // counted. A longer key carrying this one as a prefix is a different
+        // identity and must not answer for it.
+        let page = store
+            .build_list()
+            .prefix(key)
+            .limit(EXACT_MATCH_LIST_LIMIT)
+            .execute()
+            .change_context(TrustedServerError::KvStore {
+                store_name: self.store_name.clone(),
+                message: format!("Failed to check key '{}'", key.get(..8).unwrap_or(key),),
+            })?;
+
+        Ok(page.keys().iter().any(|listed| listed == key))
+    }
+
     fn delete(&self, key: &str) -> Result<(), Report<TrustedServerError>> {
         let store = self.open_store()?;
         store
diff --git a/crates/trusted-server-core/src/ec/finalize.rs b/crates/trusted-server-core/src/ec/finalize.rs
index 607ce690a..e6834d9ca 100644
--- a/crates/trusted-server-core/src/ec/finalize.rs
+++ b/crates/trusted-server-core/src/ec/finalize.rs
@@ -189,12 +189,14 @@ fn log_tombstone_outcome(
                 log_id(ec_id),
             );
         }
-        Ok(TombstoneOutcome::Unconfirmed) => {
+        Ok(TombstoneOutcome::Unconfirmed { reason }) => {
             // Unlike an unknown identity, this is a fault worth surfacing: the
             // store could not be read, so a real identity may have gone
-            // untombstoned for the batch-sync window.
+            // untombstoned for the batch-sync window. Reported once, here,
+            // with the detail the store layer handed back.
             log::error!(
-                "Could not confirm EC ID '{}' to tombstone it; the browser cookie is still expired",
+                "Could not confirm EC ID '{}' to tombstone it, so a withdrawal may go \
+                 unrecorded; the browser cookie is still expired. {reason}",
                 log_id(ec_id),
             );
         }
@@ -527,6 +529,47 @@ mod tests {
         );
     }
 
+    #[test]
+    fn withdrawal_still_expires_the_cookie_when_the_store_is_unavailable() {
+        // Cookie expiry is the primary enforcement, so it has to survive a
+        // store that cannot answer at all — the case where the identity-graph
+        // marker is exactly what goes missing.
+        let settings = create_test_settings();
+        let ec_id = sample_ec_id("dead01");
+        let consent = ConsentContext {
+            jurisdiction: Jurisdiction::UsState("CA".to_owned()),
+            gpc: true,
+            source: ConsentSource::Cookie,
+            ..Default::default()
+        };
+        let ec_context =
+            make_context_with_consent(Some(&ec_id), Some(&ec_id), true, false, consent);
+        let kv = KvIdentityGraph::failing("test-store");
+        let mut response = empty_response();
+        set_header(&mut response, "x-ts-ec", "stale");
+        let registry = PartnerRegistry::from_config(&[]).expect("should build registry");
+
+        ec_finalize_response(
+            &settings,
+            &ec_context,
+            Some(&kv),
+            &registry,
+            None,
+            None,
+            &mut response,
+        );
+
+        let set_cookie = get_header_str(&response, "set-cookie").unwrap_or_default();
+        assert!(
+            set_cookie.contains("Max-Age=0"),
+            "should expire the EC cookie even when the store is unavailable: {set_cookie}"
+        );
+        assert!(
+            get_header(&response, "x-ts-ec").is_none(),
+            "should still strip EC response headers"
+        );
+    }
+
     #[test]
     fn finalize_withdrawal_clears_cookie_and_headers() {
         let settings = create_test_settings();
diff --git a/crates/trusted-server-core/src/ec/kv.rs b/crates/trusted-server-core/src/ec/kv.rs
index bee320973..a43f599ba 100644
--- a/crates/trusted-server-core/src/ec/kv.rs
+++ b/crates/trusted-server-core/src/ec/kv.rs
@@ -20,7 +20,7 @@ use error_stack::{Report, ResultExt};
 use crate::error::TrustedServerError;
 
 use super::current_timestamp;
-use super::generation::{ec_hash, is_valid_ec_id};
+use super::generation::ec_hash;
 use super::kv_backend::{EcKvLookup, EcKvStore, EcKvWrite, EcKvWriteMode, EcKvWriteOutcome};
 use super::kv_types::{KvEntry, KvMetadata, KvNetwork};
 use super::log_id;
@@ -123,8 +123,15 @@ impl fmt::Debug for KvIdentityGraph {
     }
 }
 
+/// Longest identifier this store will look for.
+///
+/// Bounds the work a caller-supplied cookie can ask for. It is not a format
+/// check: the existence check is exact, so it does not depend on the shape of
+/// an identifier.
+const MAX_EC_ID_LEN: usize = 256;
+
 /// Result of [`KvIdentityGraph::write_withdrawal_tombstone`].
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[derive(Debug, Clone, PartialEq, Eq)]
 pub enum TombstoneOutcome {
     /// The identity was found and is now tombstoned.
     Written,
@@ -133,7 +140,10 @@ pub enum TombstoneOutcome {
     /// The store could not say whether the identity exists, so nothing was
     /// written. The browser cookie is still expired by the caller, which is the
     /// primary enforcement; only the batch-sync revocation marker is missing.
-    Unconfirmed,
+    ///
+    /// Carries why, so the caller can log it once rather than each layer
+    /// reporting the same incident.
+    Unconfirmed { reason: String },
 }
 
 impl KvIdentityGraph {
@@ -636,27 +646,24 @@ impl KvIdentityGraph {
 
     /// Whether `ec_id` names a key this store actually holds.
     ///
-    /// Uses the list API rather than [`Self::get`] deliberately. A lookup is
-    /// eventually consistent, so it can answer "missing" for an entry that was
-    /// just written; the list is strongly consistent and will not. That matters
-    /// wherever a false "missing" would discard something, because the identity
-    /// would be treated as one this deployment never issued.
+    /// Delegates to an exact, strongly consistent check. Counting keys by
+    /// prefix would not do: another key may carry this one as a prefix and
+    /// would answer for it, and a read that may lag would report a freshly
+    /// written identity as absent, discarding a genuine withdrawal.
     ///
-    /// An EC ID is a fixed-width `{64 hex}.{6 alphanumeric}` string, so no other
-    /// key can carry a whole EC ID as a prefix and the count is exact. A value
-    /// that is not a well-formed EC ID is one this store can never hold, and is
-    /// reported missing without a store round trip. That check is not merely an
-    /// optimisation: this is a prefix query, so an empty or truncated value
-    /// would match unrelated keys, and an empty one would match every key.
+    /// A value longer than any identifier this deployment could hold is
+    /// refused without a store round trip. That bound is about cost, not
+    /// correctness — the check is exact either way, so it does not care what
+    /// shape an identifier takes.
     ///
     /// # Errors
     ///
     /// Returns [`TrustedServerError::KvStore`] on store error.
     fn key_exists_confirmed(&self, ec_id: &str) -> Result<bool, Report<TrustedServerError>> {
-        if !is_valid_ec_id(ec_id) {
+        if ec_id.is_empty() || ec_id.len() > MAX_EC_ID_LEN {
             return Ok(false);
         }
-        Ok(self.store.count_keys_with_prefix(ec_id, 1)? > 0)
+        self.store.key_exists(ec_id)
     }
 
     /// Writes a withdrawal tombstone for consent enforcement.
@@ -683,11 +690,13 @@ impl KvIdentityGraph {
     /// create an identity, because the entry must have existed to pass the
     /// check at all.
     ///
-    /// When the list cannot answer, existence is re-checked with a lookup, which
-    /// cannot report an identity the store does not hold. If that cannot answer
-    /// either, nothing is written and [`TombstoneOutcome::Unconfirmed`] is
-    /// returned; the caller still expires the browser cookie, which is the
-    /// primary enforcement.
+    /// When the exact check cannot answer, existence is re-checked with a
+    /// lookup. A lookup may lag, so it can miss a very recent write, and it can
+    /// return a row already deleted or expired at the primary — but it cannot
+    /// report an identifier this deployment never issued, which is what the
+    /// gate is for. If that cannot answer either, nothing is written and
+    /// [`TombstoneOutcome::Unconfirmed`] is returned with the reason; the caller
+    /// still expires the browser cookie, which is the primary enforcement.
     ///
     /// # Errors
     ///
@@ -704,41 +713,30 @@ impl KvIdentityGraph {
             Ok(true) => {}
             Ok(false) => return Ok(TombstoneOutcome::UnknownIdentity),
             Err(list_error) => {
-                // The list could not answer. Fall back to a lookup rather than
-                // writing blind: a lookup is eventually consistent, so it can
-                // miss a very recent write, but it can never invent a row. A
-                // hit is therefore proof the identity exists, while no
-                // fabricated identifier can produce one. Writing blind here
-                // would instead hand a caller the original unbounded write back
-                // whenever the store can be pushed into failing.
-                // The raw form is used deliberately: a row whose body no longer
-                // deserializes is still a row, and presence is all that matters
-                // here, so a corrupt entry must not read as absent.
+                // The check could not answer. Fall back to a raw lookup rather
+                // than writing blind: a lookup may lag, so it can miss a very
+                // recent write, but no identifier this deployment never issued
+                // can appear in it. Writing blind would instead restore the
+                // unconditional write whenever the store can be made to fail.
+                //
+                // The raw form is deliberate: a row whose body no longer
+                // deserializes is still a row, and presence is all that matters.
                 match self.lookup_raw(ec_id) {
-                    Ok(Some(_)) => {
-                        log::warn!(
-                            "Confirmed EC ID '{}' by lookup after a list failure: {list_error:?}",
-                            log_id(ec_id),
-                        );
-                    }
+                    Ok(Some(_)) => {}
                     Ok(None) => {
-                        log::error!(
-                            "Cannot confirm EC ID '{}', so a withdrawal may go unrecorded \
-                             for the batch-sync window: the list failed and a lookup found \
-                             nothing. List error: {list_error:?}",
-                            log_id(ec_id),
-                        );
-                        return Ok(TombstoneOutcome::Unconfirmed);
+                        return Ok(TombstoneOutcome::Unconfirmed {
+                            reason: format!(
+                                "existence check failed and a lookup found nothing: {list_error:?}"
+                            ),
+                        });
                     }
                     Err(lookup_error) => {
-                        log::error!(
-                            "Cannot confirm EC ID '{}', so a withdrawal may go unrecorded \
-                             for the batch-sync window: neither the list nor a lookup could \
-                             be read. List error: {list_error:?}. Lookup error: \
-                             {lookup_error:?}",
-                            log_id(ec_id),
-                        );
-                        return Ok(TombstoneOutcome::Unconfirmed);
+                        return Ok(TombstoneOutcome::Unconfirmed {
+                            reason: format!(
+                                "neither the existence check nor a lookup could be read: \
+                                 {list_error:?}; {lookup_error:?}"
+                            ),
+                        });
                     }
                 }
             }
@@ -1091,6 +1089,10 @@ mod tests {
             self.inner.count_keys_with_prefix(prefix, limit)
         }
 
+        fn key_exists(&self, key: &str) -> Result<bool, Report<TrustedServerError>> {
+            self.inner.key_exists(key)
+        }
+
         fn delete(&self, key: &str) -> Result<(), Report<TrustedServerError>> {
             self.inner.delete(key)
         }
@@ -1509,6 +1511,13 @@ mod tests {
             }))
         }
 
+        fn key_exists(&self, _key: &str) -> Result<bool, Report<TrustedServerError>> {
+            Err(Report::new(TrustedServerError::KvStore {
+                store_name: "test_store".to_owned(),
+                message: "list unavailable".to_owned(),
+            }))
+        }
+
         fn delete(&self, key: &str) -> Result<(), Report<TrustedServerError>> {
             self.inner.delete(key)
         }
@@ -1543,10 +1552,12 @@ mod tests {
         let kv = KvIdentityGraph::new(ListFailingEcKv::new());
         let ec_id = format!("{}.ABC123", "8".repeat(64));
 
-        assert_eq!(
-            kv.write_withdrawal_tombstone(&ec_id)
-                .expect("should resolve the withdrawal"),
-            TombstoneOutcome::Unconfirmed,
+        assert!(
+            matches!(
+                kv.write_withdrawal_tombstone(&ec_id)
+                    .expect("should resolve the withdrawal"),
+                TombstoneOutcome::Unconfirmed { .. }
+            ),
             "should refuse to write when existence cannot be established"
         );
         assert!(
@@ -1555,61 +1566,28 @@ mod tests {
         );
     }
 
-    /// Tripwire for [`KvIdentityGraph::key_exists_confirmed`].
-    ///
-    /// That check is a prefix query, and it is only exact because every value
-    /// [`is_valid_ec_id`] accepts is the same width — no accepted identifier can
-    /// be a proper prefix of another. Nothing in the type system enforces that,
-    /// so this test states the dependency: if the accepted grammar is ever
-    /// widened to variable-length identifiers, such as a provider envelope like
-    /// `{code}~value`, this fails and whoever widened it has to revisit the
-    /// prefix query rather than discovering a false positive in production.
-    #[test]
-    fn the_prefix_check_depends_on_ec_ids_being_fixed_width() {
-        let accepted = format!("{}.ABC123", "a".repeat(64));
-        assert!(is_valid_ec_id(&accepted), "the built-in shape is accepted");
-        assert_eq!(
-            accepted.len(),
-            71,
-            "an accepted identifier is a fixed 71 bytes"
-        );
-
-        // Anything longer or shorter, including a provider envelope, must be
-        // refused while the existence check relies on prefix semantics.
-        for widened in [
-            format!("hmac~{accepted}"),
-            format!("51dd~{}", "opaque-value-of-some-other-length"),
-            format!("{accepted}trailing"),
-            "a".repeat(64),
-        ] {
-            assert!(
-                !is_valid_ec_id(&widened),
-                "a variable-length identifier would break the prefix query: {widened}"
-            );
-        }
-    }
-
     #[test]
-    fn key_exists_confirmed_does_not_match_a_longer_key_by_prefix() {
-        // The check is a prefix query, sound only while every EC ID is the same
-        // width. If that ever stops holding, a shorter identifier would match a
-        // longer unrelated row and a tombstone would be written for an identity
-        // that was never issued. This test pins the assumption.
+    fn a_longer_key_does_not_answer_for_the_identity_it_starts_with() {
         let kv = KvIdentityGraph::in_memory("test_store");
-        let held = format!("{}.ABC123", "7".repeat(64));
-        kv.create(&held, &live_entry()).expect("should create");
+        let ec_id = format!("{}.ABC123", "7".repeat(64));
+        // Only a longer key exists. The identity itself was never issued, so a
+        // check that matched by prefix would report it as held and tombstone it.
+        kv.create(&format!("{ec_id}trailing"), &live_entry())
+            .expect("should create");
 
-        let shorter = &held[..held.len() - 1];
         assert!(
-            !kv.key_exists_confirmed(shorter)
-                .expect("should resolve the check"),
-            "a proper prefix of a stored key is a different identity"
+            !kv.key_exists_confirmed(&ec_id).expect("should check"),
+            "a longer key is a different identity"
         );
         assert_eq!(
-            kv.write_withdrawal_tombstone(shorter)
+            kv.write_withdrawal_tombstone(&ec_id)
                 .expect("should resolve the withdrawal"),
             TombstoneOutcome::UnknownIdentity,
-            "should not tombstone via a prefix match"
+            "should not tombstone an identity the store never held"
+        );
+        assert!(
+            kv.get(&ec_id).expect("should read back").is_none(),
+            "should not create a row via a prefix match"
         );
     }
 
diff --git a/crates/trusted-server-core/src/ec/kv_backend.rs b/crates/trusted-server-core/src/ec/kv_backend.rs
index 60f938291..1da9bab8f 100644
--- a/crates/trusted-server-core/src/ec/kv_backend.rs
+++ b/crates/trusted-server-core/src/ec/kv_backend.rs
@@ -107,6 +107,17 @@ pub trait EcKvStore {
         limit: u32,
     ) -> Result<u32, Report<TrustedServerError>>;
 
+    /// Whether the store holds exactly `key`.
+    ///
+    /// Must be an exact match and strongly consistent. A prefix scan is not a
+    /// substitute: another key may carry this one as a prefix, and a read that
+    /// may lag would report a freshly written key as absent.
+    ///
+    /// # Errors
+    ///
+    /// Returns [`TrustedServerError::KvStore`] on store open or list failure.
+    fn key_exists(&self, key: &str) -> Result<bool, Report<TrustedServerError>>;
+
     /// Hard-deletes a key.
     ///
     /// # Errors
@@ -205,6 +216,11 @@ pub(crate) mod test_support {
             Ok(count as u32)
         }
 
+        fn key_exists(&self, key: &str) -> Result<bool, Report<TrustedServerError>> {
+            let entries = self.entries.lock().expect("should lock in-memory store");
+            Ok(entries.contains_key(key))
+        }
+
         fn delete(&self, key: &str) -> Result<(), Report<TrustedServerError>> {
             let mut entries = self.entries.lock().expect("should lock in-memory store");
             entries.remove(key);
@@ -256,6 +272,10 @@ pub(crate) mod test_support {
             Err(self.error("list"))
         }
 
+        fn key_exists(&self, _key: &str) -> Result<bool, Report<TrustedServerError>> {
+            Err(self.error("key_exists"))
+        }
+
         fn delete(&self, _key: &str) -> Result<(), Report<TrustedServerError>> {
             Err(self.error("delete"))
         }

From 45c0689b3797e8cbd6bb0583a7f383df0ccdbed8 Mon Sep 17 00:00:00 2001
From: prk-Jr <prakashsagwara@gmail.com>
Date: Wed, 2 Sep 2026 11:01:50 +0530
Subject: [PATCH 04/11] Follow every page when checking for an exact key

Scanning one page of prefix matches assumed the exact key would be in it.
Nothing guarantees that when other keys share the prefix, and stopping early
reports a held identity as missing, discarding its withdrawal. Iterate the
pages instead.

Also correct two test comments that still described the removed grammar gate.
---
 .../src/ec_kv.rs                              | 26 +++++++++----------
 crates/trusted-server-core/src/ec/kv.rs       | 17 +++++-------
 2 files changed, 19 insertions(+), 24 deletions(-)

diff --git a/crates/trusted-server-adapter-fastly/src/ec_kv.rs b/crates/trusted-server-adapter-fastly/src/ec_kv.rs
index 697f9feba..e7305bb8f 100644
--- a/crates/trusted-server-adapter-fastly/src/ec_kv.rs
+++ b/crates/trusted-server-adapter-fastly/src/ec_kv.rs
@@ -42,12 +42,6 @@ impl FastlyEcKvStore {
     }
 }
 
-/// Keys to request when checking for one exact key.
-///
-/// Only an exact match matters, but a page of one could be filled by a longer
-/// key sharing the prefix, so a small page is requested and scanned.
-const EXACT_MATCH_LIST_LIMIT: u32 = 10;
-
 impl EcKvStore for FastlyEcKvStore {
     fn store_name(&self) -> &str {
         &self.store_name
@@ -140,17 +134,21 @@ impl EcKvStore for FastlyEcKvStore {
         // prefix — so the returned keys are compared for equality rather than
         // counted. A longer key carrying this one as a prefix is a different
         // identity and must not answer for it.
-        let page = store
-            .build_list()
-            .prefix(key)
-            .limit(EXACT_MATCH_LIST_LIMIT)
-            .execute()
-            .change_context(TrustedServerError::KvStore {
+        //
+        // Every page is followed. Nothing guarantees the exact key lands in the
+        // first one when other keys share its prefix, and stopping early would
+        // report a held identity as missing and discard its withdrawal.
+        for page in store.build_list().prefix(key).iter() {
+            let page = page.change_context(TrustedServerError::KvStore {
                 store_name: self.store_name.clone(),
-                message: format!("Failed to check key '{}'", key.get(..8).unwrap_or(key),),
+                message: format!("Failed to check key '{}'", key.get(..8).unwrap_or(key)),
             })?;
+            if page.keys().iter().any(|listed| listed == key) {
+                return Ok(true);
+            }
+        }
 
-        Ok(page.keys().iter().any(|listed| listed == key))
+        Ok(false)
     }
 
     fn delete(&self, key: &str) -> Result<(), Report<TrustedServerError>> {
diff --git a/crates/trusted-server-core/src/ec/kv.rs b/crates/trusted-server-core/src/ec/kv.rs
index a43f599ba..6cb7eee26 100644
--- a/crates/trusted-server-core/src/ec/kv.rs
+++ b/crates/trusted-server-core/src/ec/kv.rs
@@ -1427,19 +1427,15 @@ mod tests {
     }
 
     #[test]
-    fn key_exists_confirmed_refuses_a_value_that_is_not_an_ec_id() {
+    fn key_exists_confirmed_refuses_a_value_no_key_could_be() {
         let kv = KvIdentityGraph::in_memory("test_store");
         kv.create(&format!("{}.ABC123", "e".repeat(64)), &live_entry())
             .expect("should create");
 
-        // A prefix query would match every key for an empty value and unrelated
-        // keys for a truncated one, so neither may reach the store.
-        for value in [
-            "",
-            "e".repeat(64).as_str(),
-            "not-an-ec-id",
-            "E".repeat(64).as_str(),
-        ] {
+        // The check is exact, so a malformed value is simply absent. Only an
+        // empty or over-long value is refused ahead of the store, and that is
+        // to bound the work a cookie can ask for, not for correctness.
+        for value in ["", &"z".repeat(257), "not-an-ec-id", &"E".repeat(64)] {
             assert!(
                 !kv.key_exists_confirmed(value)
                     .expect("should resolve the check"),
@@ -1548,7 +1544,8 @@ mod tests {
     fn a_failing_list_is_not_a_way_to_write_for_an_identity_that_was_never_issued() {
         // The caller controls the identifier and can drive load, so a store
         // failure must not become a route to the write this gate exists to
-        // prevent. A lookup cannot report a row that does not exist.
+        // prevent. A lagging lookup may return a row already deleted, but it
+        // cannot return one for an identifier this deployment never issued.
         let kv = KvIdentityGraph::new(ListFailingEcKv::new());
         let ec_id = format!("{}.ABC123", "8".repeat(64));
 

From b5b68bba9180fe05f3068756f0937ccc93517eb2 Mon Sep 17 00:00:00 2001
From: prk-Jr <prakashsagwara@gmail.com>
Date: Wed, 2 Sep 2026 11:22:14 +0530
Subject: [PATCH 05/11] Redact an identifier by character, not by byte
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

A byte index landing inside a multi-byte character makes `get` return
`None`, and the fallback printed the whole identifier — the opposite of what
the redaction is for. Truncate by character, and reuse the helper in the
Fastly store rather than repeating the byte form there.

Prove the bounds check avoids a store call with a counting backend, and
rename the test that claimed to cover a grammar gate that no longer exists.
---
 .../src/ec_kv.rs                              | 10 +--
 crates/trusted-server-core/src/ec/kv.rs       | 89 ++++++++++++++++---
 crates/trusted-server-core/src/ec/mod.rs      | 38 +++++++-
 3 files changed, 117 insertions(+), 20 deletions(-)

diff --git a/crates/trusted-server-adapter-fastly/src/ec_kv.rs b/crates/trusted-server-adapter-fastly/src/ec_kv.rs
index e7305bb8f..1eb57612c 100644
--- a/crates/trusted-server-adapter-fastly/src/ec_kv.rs
+++ b/crates/trusted-server-adapter-fastly/src/ec_kv.rs
@@ -9,6 +9,7 @@ use fastly::kv_store::{InsertMode, KVStore};
 use trusted_server_core::ec::kv_backend::{
     EcKvLookup, EcKvStore, EcKvWrite, EcKvWriteMode, EcKvWriteOutcome,
 };
+use trusted_server_core::ec::log_id;
 use trusted_server_core::error::TrustedServerError;
 
 /// Fastly KV Store backend for the EC identity graph.
@@ -56,7 +57,7 @@ impl EcKvStore for FastlyEcKvStore {
                 return Err(
                     Report::new(err).change_context(TrustedServerError::KvStore {
                         store_name: self.store_name.clone(),
-                        message: format!("Failed to read key '{}'", key.get(..8).unwrap_or(key),),
+                        message: format!("Failed to read key '{}'", log_id(key),),
                     }),
                 );
             }
@@ -117,10 +118,7 @@ impl EcKvStore for FastlyEcKvStore {
             .execute()
             .change_context(TrustedServerError::KvStore {
                 store_name: self.store_name.clone(),
-                message: format!(
-                    "Failed to list keys with prefix '{}'",
-                    prefix.get(..8).unwrap_or(prefix),
-                ),
+                message: format!("Failed to list keys with prefix '{}'", log_id(prefix),),
             })?;
 
         #[allow(clippy::cast_possible_truncation)]
@@ -141,7 +139,7 @@ impl EcKvStore for FastlyEcKvStore {
         for page in store.build_list().prefix(key).iter() {
             let page = page.change_context(TrustedServerError::KvStore {
                 store_name: self.store_name.clone(),
-                message: format!("Failed to check key '{}'", key.get(..8).unwrap_or(key)),
+                message: format!("Failed to check key '{}'", log_id(key)),
             })?;
             if page.keys().iter().any(|listed| listed == key) {
                 return Ok(true);
diff --git a/crates/trusted-server-core/src/ec/kv.rs b/crates/trusted-server-core/src/ec/kv.rs
index 6cb7eee26..846877c74 100644
--- a/crates/trusted-server-core/src/ec/kv.rs
+++ b/crates/trusted-server-core/src/ec/kv.rs
@@ -1427,25 +1427,35 @@ mod tests {
     }
 
     #[test]
-    fn key_exists_confirmed_refuses_a_value_no_key_could_be() {
-        let kv = KvIdentityGraph::in_memory("test_store");
-        kv.create(&format!("{}.ABC123", "e".repeat(64)), &live_entry())
-            .expect("should create");
+    fn an_out_of_bounds_value_is_refused_without_reaching_the_store() {
+        let (store, checks) = CountingEcKv::new();
+        let kv = KvIdentityGraph::new(store);
+        let count = || *checks.lock().expect("should lock the check counter");
 
-        // The check is exact, so a malformed value is simply absent. Only an
-        // empty or over-long value is refused ahead of the store, and that is
-        // to bound the work a cookie can ask for, not for correctness.
-        for value in ["", &"z".repeat(257), "not-an-ec-id", &"E".repeat(64)] {
+        for value in ["", &"z".repeat(257)] {
             assert!(
                 !kv.key_exists_confirmed(value)
                     .expect("should resolve the check"),
                 "should not report `{value}` as held"
             );
         }
+        assert_eq!(
+            count(),
+            0,
+            "a value no key could be must not cost a store round trip"
+        );
+
+        // A value that is within bounds but simply absent does reach the store.
+        let absent = format!("{}.ABC123", "e".repeat(64));
+        assert!(
+            !kv.key_exists_confirmed(&absent).expect("should check"),
+            "an absent identity is not held"
+        );
+        assert_eq!(count(), 1, "an in-bounds value is checked exactly once");
     }
 
     #[test]
-    fn write_withdrawal_tombstone_refuses_a_value_that_is_not_an_ec_id() {
+    fn write_withdrawal_tombstone_refuses_an_empty_identifier() {
         let kv = KvIdentityGraph::in_memory("test_store");
         kv.create(&format!("{}.ABC123", "f".repeat(64)), &live_entry())
             .expect("should create");
@@ -1454,7 +1464,7 @@ mod tests {
             kv.write_withdrawal_tombstone("")
                 .expect("should resolve the withdrawal"),
             TombstoneOutcome::UnknownIdentity,
-            "an empty identifier must not be treated as a match-all prefix"
+            "an empty identifier names no key and must not withdraw anything"
         );
         let (held, _) = kv
             .get(&format!("{}.ABC123", "f".repeat(64)))
@@ -1466,6 +1476,65 @@ mod tests {
         );
     }
 
+    /// Store double that records how many existence checks reach it.
+    struct CountingEcKv {
+        inner: super::super::kv_backend::test_support::InMemoryEcKv,
+        existence_checks: std::sync::Arc<std::sync::Mutex<usize>>,
+    }
+
+    impl CountingEcKv {
+        /// Returns the store and a handle to its counter, which stays readable
+        /// after the store moves into the graph.
+        fn new() -> (Self, std::sync::Arc<std::sync::Mutex<usize>>) {
+            let counter = std::sync::Arc::new(std::sync::Mutex::new(0));
+            (
+                Self {
+                    inner: super::super::kv_backend::test_support::InMemoryEcKv::new("test_store"),
+                    existence_checks: std::sync::Arc::clone(&counter),
+                },
+                counter,
+            )
+        }
+    }
+
+    impl EcKvStore for CountingEcKv {
+        fn store_name(&self) -> &str {
+            self.inner.store_name()
+        }
+
+        fn lookup(&self, key: &str) -> Result<Option<EcKvLookup>, Report<TrustedServerError>> {
+            self.inner.lookup(key)
+        }
+
+        fn insert(
+            &self,
+            key: &str,
+            write: EcKvWrite<'_>,
+        ) -> Result<EcKvWriteOutcome, Report<TrustedServerError>> {
+            self.inner.insert(key, write)
+        }
+
+        fn count_keys_with_prefix(
+            &self,
+            prefix: &str,
+            limit: u32,
+        ) -> Result<u32, Report<TrustedServerError>> {
+            self.inner.count_keys_with_prefix(prefix, limit)
+        }
+
+        fn key_exists(&self, key: &str) -> Result<bool, Report<TrustedServerError>> {
+            *self
+                .existence_checks
+                .lock()
+                .expect("should lock the check counter") += 1;
+            self.inner.key_exists(key)
+        }
+
+        fn delete(&self, key: &str) -> Result<(), Report<TrustedServerError>> {
+            self.inner.delete(key)
+        }
+    }
+
     /// Store double whose list API always fails while writes still work.
     struct ListFailingEcKv {
         inner: super::super::kv_backend::test_support::InMemoryEcKv,
diff --git a/crates/trusted-server-core/src/ec/mod.rs b/crates/trusted-server-core/src/ec/mod.rs
index 840ce90d3..1e7396390 100644
--- a/crates/trusted-server-core/src/ec/mod.rs
+++ b/crates/trusted-server-core/src/ec/mod.rs
@@ -49,14 +49,20 @@ pub mod pull_sync;
 pub mod rate_limiter;
 pub mod registry;
 
+/// Characters of an identifier kept when redacting it for a log.
+const LOG_ID_PREFIX_CHARS: usize = 8;
+
 /// Truncates an EC ID for safe inclusion in log messages.
 ///
-/// Returns the first 8 characters followed by `…` to aid debugging without
-/// writing the full user identifier to logs (satisfies the `CodeQL`
-/// "cleartext logging of sensitive information" rule).
+/// Returns the first [`LOG_ID_PREFIX_CHARS`] characters followed by `…` to aid
+/// debugging without writing the full user identifier to logs (satisfies the
+/// `CodeQL` "cleartext logging of sensitive information" rule).
 #[must_use]
 pub fn log_id(ec_id: &str) -> String {
-    let prefix = ec_id.get(..8).unwrap_or(ec_id);
+    // Truncated by character, not by byte. A byte index that lands inside a
+    // multi-byte character makes `get` return `None`, and falling back to the
+    // whole value would print in full the identifier this exists to redact.
+    let prefix: String = ec_id.chars().take(LOG_ID_PREFIX_CHARS).collect();
     format!("{prefix}\u{2026}")
 }
 
@@ -493,6 +499,30 @@ pub(crate) fn current_timestamp() -> u64 {
 
 #[cfg(test)]
 mod tests {
+
+    #[test]
+    fn log_id_never_emits_more_than_the_redacted_prefix() {
+        // A byte index inside a multi-byte character used to make the
+        // truncation fall back to the whole value, printing in full the
+        // identifier this redacts.
+        let boundary_splitting = "abcdefg\u{e9}-tail-that-must-not-be-logged";
+        let redacted = log_id(boundary_splitting);
+
+        assert!(
+            !redacted.contains("must-not-be-logged"),
+            "should not disclose the rest of the identifier: {redacted}"
+        );
+        assert_eq!(
+            redacted.chars().count(),
+            9,
+            "should be eight characters plus the ellipsis: {redacted}"
+        );
+
+        // The ordinary case is unchanged.
+        assert_eq!(log_id("0123456789abcdef.ABC123"), "01234567\u{2026}");
+        // A value shorter than the prefix is emitted whole, which is all there is.
+        assert_eq!(log_id("abc"), "abc\u{2026}");
+    }
     use super::*;
     use crate::platform::test_support::noop_services;
     use crate::test_support::tests::create_test_settings;

From f6181d149a380ea839d4c34596854bacb79c6cf3 Mon Sep 17 00:00:00 2001
From: prk-Jr <prakashsagwara@gmail.com>
Date: Wed, 2 Sep 2026 13:15:26 +0530
Subject: [PATCH 06/11] Move the paged existence decision somewhere it can be
 tested

The exact-key scan lived in the Fastly store, where no test double can
exercise paging. Extract it as `contains_exact_key` and have the backend
supply pages to it, so multi-page matches, prefix-only keys, early exit and
page errors are all covered natively.
---
 .../src/ec_kv.rs                              |  29 ++--
 .../trusted-server-core/src/ec/kv_backend.rs  | 126 ++++++++++++++++++
 2 files changed, 137 insertions(+), 18 deletions(-)

diff --git a/crates/trusted-server-adapter-fastly/src/ec_kv.rs b/crates/trusted-server-adapter-fastly/src/ec_kv.rs
index 1eb57612c..ad7390318 100644
--- a/crates/trusted-server-adapter-fastly/src/ec_kv.rs
+++ b/crates/trusted-server-adapter-fastly/src/ec_kv.rs
@@ -7,7 +7,7 @@
 use error_stack::{Report, ResultExt};
 use fastly::kv_store::{InsertMode, KVStore};
 use trusted_server_core::ec::kv_backend::{
-    EcKvLookup, EcKvStore, EcKvWrite, EcKvWriteMode, EcKvWriteOutcome,
+    EcKvLookup, EcKvStore, EcKvWrite, EcKvWriteMode, EcKvWriteOutcome, contains_exact_key,
 };
 use trusted_server_core::ec::log_id;
 use trusted_server_core::error::TrustedServerError;
@@ -129,24 +129,17 @@ impl EcKvStore for FastlyEcKvStore {
     fn key_exists(&self, key: &str) -> Result<bool, Report<TrustedServerError>> {
         let store = self.open_store()?;
         // The list is strongly consistent, unlike a lookup, but it matches by
-        // prefix — so the returned keys are compared for equality rather than
-        // counted. A longer key carrying this one as a prefix is a different
-        // identity and must not answer for it.
-        //
-        // Every page is followed. Nothing guarantees the exact key lands in the
-        // first one when other keys share its prefix, and stopping early would
-        // report a held identity as missing and discard its withdrawal.
-        for page in store.build_list().prefix(key).iter() {
-            let page = page.change_context(TrustedServerError::KvStore {
-                store_name: self.store_name.clone(),
-                message: format!("Failed to check key '{}'", log_id(key)),
-            })?;
-            if page.keys().iter().any(|listed| listed == key) {
-                return Ok(true);
-            }
-        }
+        // prefix, so the decision of what counts as a match belongs to
+        // `contains_exact_key` — which is where that behaviour is tested.
+        let pages = store.build_list().prefix(key).iter().map(|page| {
+            page.map(fastly::kv_store::ListPage::into_keys)
+                .change_context(TrustedServerError::KvStore {
+                    store_name: self.store_name.clone(),
+                    message: format!("Failed to check key '{}'", log_id(key)),
+                })
+        });
 
-        Ok(false)
+        contains_exact_key(pages, key)
     }
 
     fn delete(&self, key: &str) -> Result<(), Report<TrustedServerError>> {
diff --git a/crates/trusted-server-core/src/ec/kv_backend.rs b/crates/trusted-server-core/src/ec/kv_backend.rs
index 1da9bab8f..561835b05 100644
--- a/crates/trusted-server-core/src/ec/kv_backend.rs
+++ b/crates/trusted-server-core/src/ec/kv_backend.rs
@@ -70,6 +70,32 @@ pub enum EcKvWriteOutcome {
 /// Infrastructure failures are reported as [`TrustedServerError::KvStore`];
 /// write precondition failures are part of the normal control flow and are
 /// returned as [`EcKvWriteOutcome::PreconditionFailed`] instead of errors.
+/// Whether `key` appears exactly in a paged listing.
+///
+/// Backends that can only match by prefix use this to decide existence: a
+/// longer key carrying `key` as a prefix is a different identity and must not
+/// answer for it, so the listed keys are compared for equality.
+///
+/// Every page is visited until a match is found. Nothing guarantees the exact
+/// key lands in the first page when other keys share its prefix, and stopping
+/// early would report a held identity as missing.
+///
+/// # Errors
+///
+/// Returns the first page error, so a listing that cannot be read is never
+/// mistaken for a listing that does not contain the key.
+pub fn contains_exact_key<E>(
+    pages: impl IntoIterator<Item = Result<Vec<String>, E>>,
+    key: &str,
+) -> Result<bool, E> {
+    for page in pages {
+        if page?.iter().any(|listed| listed == key) {
+            return Ok(true);
+        }
+    }
+    Ok(false)
+}
+
 pub trait EcKvStore {
     /// Returns the platform store name, used in log and error messages.
     fn store_name(&self) -> &str;
@@ -281,3 +307,103 @@ pub(crate) mod test_support {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::contains_exact_key;
+
+    /// Page iterator that records how many pages were consumed.
+    struct CountingPages {
+        pages: std::vec::IntoIter<Result<Vec<String>, &'static str>>,
+        consumed: std::rc::Rc<std::cell::Cell<usize>>,
+    }
+
+    impl Iterator for CountingPages {
+        type Item = Result<Vec<String>, &'static str>;
+
+        fn next(&mut self) -> Option<Self::Item> {
+            let page = self.pages.next();
+            if page.is_some() {
+                self.consumed.set(self.consumed.get() + 1);
+            }
+            page
+        }
+    }
+
+    fn counting(
+        pages: Vec<Result<Vec<String>, &'static str>>,
+    ) -> (CountingPages, std::rc::Rc<std::cell::Cell<usize>>) {
+        let consumed = std::rc::Rc::new(std::cell::Cell::new(0));
+        (
+            CountingPages {
+                pages: pages.into_iter(),
+                consumed: std::rc::Rc::clone(&consumed),
+            },
+            consumed,
+        )
+    }
+
+    fn page(keys: &[&str]) -> Result<Vec<String>, &'static str> {
+        Ok(keys.iter().map(|key| (*key).to_owned()).collect())
+    }
+
+    #[test]
+    fn finds_a_match_on_a_later_page() {
+        let (pages, _) = counting(vec![
+            page(&["wanted-suffix-a", "wanted-suffix-b"]),
+            page(&["wanted-suffix-c"]),
+            page(&["wanted"]),
+        ]);
+
+        assert!(
+            contains_exact_key(pages, "wanted").expect("should scan the pages"),
+            "a match on the last page must still be found"
+        );
+    }
+
+    #[test]
+    fn ignores_keys_that_only_start_with_the_one_asked_for() {
+        let (pages, _) = counting(vec![page(&["wanted-suffix", "wantedx", "wanted2"])]);
+
+        assert!(
+            !contains_exact_key(pages, "wanted").expect("should scan the pages"),
+            "a longer key is a different identity"
+        );
+    }
+
+    #[test]
+    fn stops_at_the_first_match() {
+        let (pages, consumed) = counting(vec![
+            page(&["wanted"]),
+            page(&["never-read"]),
+            page(&["never-read-either"]),
+        ]);
+
+        assert!(
+            contains_exact_key(pages, "wanted").expect("should scan the pages"),
+            "should find the match"
+        );
+        assert_eq!(consumed.get(), 1, "should not read past the match");
+    }
+
+    #[test]
+    fn propagates_a_page_error_rather_than_reporting_absent() {
+        let (pages, _) = counting(vec![page(&["wanted-suffix"]), Err("list unavailable")]);
+
+        assert_eq!(
+            contains_exact_key(pages, "wanted"),
+            Err("list unavailable"),
+            "an unreadable listing must not read as absent"
+        );
+    }
+
+    #[test]
+    fn reports_absent_for_an_empty_listing() {
+        let (pages, _) = counting(Vec::new());
+
+        assert!(
+            !contains_exact_key(pages, "wanted").expect("should scan the pages"),
+            "nothing listed means nothing held"
+        );
+    }
+}

From d80311f6016cac23c1cb559ae5cbd28f96167ab8 Mon Sep 17 00:00:00 2001
From: prk-Jr <prakashsagwara@gmail.com>
Date: Wed, 2 Sep 2026 15:10:03 +0530
Subject: [PATCH 07/11] Report an undetermined existence check as an error
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

A third `Ok` variant was discarded by any caller inspecting only the error
case, dropping a possibly-unrecorded withdrawal in silence. Returning `Err`
keeps the underlying reports intact instead of flattening them into a string.

Finish the redaction pass — insert, delete, and the deserialize paths still
embedded the raw key — treat an empty prefix listing as absent rather than a
failure, bound the pages an existence check will walk, and put the store
trait's doc comment back on the trait.
---
 .../src/ec_kv.rs                              |  41 +++++--
 crates/trusted-server-core/src/ec/finalize.rs |  25 ++--
 crates/trusted-server-core/src/ec/kv.rs       | 113 ++++++++++++------
 .../trusted-server-core/src/ec/kv_backend.rs  |  12 +-
 4 files changed, 122 insertions(+), 69 deletions(-)

diff --git a/crates/trusted-server-adapter-fastly/src/ec_kv.rs b/crates/trusted-server-adapter-fastly/src/ec_kv.rs
index ad7390318..c63d6ef49 100644
--- a/crates/trusted-server-adapter-fastly/src/ec_kv.rs
+++ b/crates/trusted-server-adapter-fastly/src/ec_kv.rs
@@ -43,6 +43,17 @@ impl FastlyEcKvStore {
     }
 }
 
+/// Keys requested per page when checking for one exact key.
+const EXACT_MATCH_PAGE_SIZE: u32 = 100;
+/// Pages walked before giving up on an exact-key check.
+///
+/// A well-formed identifier is the whole key, so at most a handful of keys can
+/// carry it as a prefix and one page is the normal case. The cap stops a short
+/// prefix from walking the keyspace on the response path; exceeding it yields
+/// no match rather than a wrong answer, and the caller treats that as
+/// unconfirmed.
+const EXACT_MATCH_MAX_PAGES: usize = 4;
+
 impl EcKvStore for FastlyEcKvStore {
     fn store_name(&self) -> &str {
         &self.store_name
@@ -99,7 +110,7 @@ impl EcKvStore for FastlyEcKvStore {
             Err(err) => Err(
                 Report::new(err).change_context(TrustedServerError::KvStore {
                     store_name: self.store_name.clone(),
-                    message: format!("Failed to write entry for key '{key}'"),
+                    message: format!("Failed to write entry for key '{}'", log_id(key)),
                 }),
             ),
         }
@@ -131,13 +142,25 @@ impl EcKvStore for FastlyEcKvStore {
         // The list is strongly consistent, unlike a lookup, but it matches by
         // prefix, so the decision of what counts as a match belongs to
         // `contains_exact_key` — which is where that behaviour is tested.
-        let pages = store.build_list().prefix(key).iter().map(|page| {
-            page.map(fastly::kv_store::ListPage::into_keys)
-                .change_context(TrustedServerError::KvStore {
-                    store_name: self.store_name.clone(),
-                    message: format!("Failed to check key '{}'", log_id(key)),
-                })
-        });
+        let pages = store
+            .build_list()
+            .prefix(key)
+            .limit(EXACT_MATCH_PAGE_SIZE)
+            .iter()
+            .take(EXACT_MATCH_MAX_PAGES)
+            .map(|page| match page {
+                // A store that reports nothing under the prefix is telling us
+                // the key is absent, which is what `lookup` already does for a
+                // missing key. Treating it as a failure would send every
+                // withdrawal naming an unissued identity down the error path.
+                Err(fastly::kv_store::KVStoreError::ItemNotFound) => Ok(Vec::new()),
+                other => other
+                    .map(fastly::kv_store::ListPage::into_keys)
+                    .change_context(TrustedServerError::KvStore {
+                        store_name: self.store_name.clone(),
+                        message: format!("Failed to check key '{}'", log_id(key)),
+                    }),
+            });
 
         contains_exact_key(pages, key)
     }
@@ -148,7 +171,7 @@ impl EcKvStore for FastlyEcKvStore {
             .delete(key)
             .change_context(TrustedServerError::KvStore {
                 store_name: self.store_name.clone(),
-                message: format!("Failed to delete key '{key}'"),
+                message: format!("Failed to delete key '{}'", log_id(key)),
             })
     }
 }
diff --git a/crates/trusted-server-core/src/ec/finalize.rs b/crates/trusted-server-core/src/ec/finalize.rs
index e6834d9ca..20433e34a 100644
--- a/crates/trusted-server-core/src/ec/finalize.rs
+++ b/crates/trusted-server-core/src/ec/finalize.rs
@@ -173,10 +173,9 @@ fn finalize_unusable_consent(
 ///
 /// An unknown identity is expected traffic rather than a fault: the identifier
 /// comes from a client-supplied cookie, so it may name something this
-/// deployment never issued. An unconfirmed identity is different: the store
-/// could not answer, so a real row may have gone unmarked and that is logged as
-/// an error. The browser cookie is expired in every case, and that is the
-/// primary enforcement.
+/// deployment never issued. An error is different: nothing was recorded, so a
+/// real row may have gone unmarked, and that is logged as a fault. The browser
+/// cookie is expired in every case, and that is the primary enforcement.
 fn log_tombstone_outcome(
     ec_id: &str,
     outcome: Result<TombstoneOutcome, Report<TrustedServerError>>,
@@ -189,20 +188,14 @@ fn log_tombstone_outcome(
                 log_id(ec_id),
             );
         }
-        Ok(TombstoneOutcome::Unconfirmed { reason }) => {
-            // Unlike an unknown identity, this is a fault worth surfacing: the
-            // store could not be read, so a real identity may have gone
-            // untombstoned for the batch-sync window. Reported once, here,
-            // with the detail the store layer handed back.
-            log::error!(
-                "Could not confirm EC ID '{}' to tombstone it, so a withdrawal may go \
-                 unrecorded; the browser cookie is still expired. {reason}",
-                log_id(ec_id),
-            );
-        }
         Err(err) => {
+            // Covers both a failed write and a check that could not determine
+            // whether the identity exists. Either way no marker was recorded,
+            // so a withdrawal may go unrecorded for the batch-sync window; the
+            // browser cookie is expired regardless.
             log::error!(
-                "Failed to write withdrawal tombstone for EC ID '{}': {err:?}",
+                "Could not record the withdrawal of EC ID '{}', so it may go unrecorded \
+                 for the batch-sync window; the browser cookie is still expired: {err:?}",
                 log_id(ec_id),
             );
         }
diff --git a/crates/trusted-server-core/src/ec/kv.rs b/crates/trusted-server-core/src/ec/kv.rs
index 846877c74..dc47c4ab2 100644
--- a/crates/trusted-server-core/src/ec/kv.rs
+++ b/crates/trusted-server-core/src/ec/kv.rs
@@ -131,19 +131,12 @@ impl fmt::Debug for KvIdentityGraph {
 const MAX_EC_ID_LEN: usize = 256;
 
 /// Result of [`KvIdentityGraph::write_withdrawal_tombstone`].
-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum TombstoneOutcome {
     /// The identity was found and is now tombstoned.
     Written,
     /// No such identity is held, so there was nothing to mark withdrawn.
     UnknownIdentity,
-    /// The store could not say whether the identity exists, so nothing was
-    /// written. The browser cookie is still expired by the caller, which is the
-    /// primary enforcement; only the batch-sync revocation marker is missing.
-    ///
-    /// Carries why, so the caller can log it once rather than each layer
-    /// reporting the same incident.
-    Unconfirmed { reason: String },
 }
 
 impl KvIdentityGraph {
@@ -236,13 +229,16 @@ impl KvIdentityGraph {
         let entry: KvEntry =
             serde_json::from_slice(body_bytes).change_context(TrustedServerError::KvStore {
                 store_name: store_name.to_owned(),
-                message: format!("Failed to deserialize entry for key '{ec_id}'"),
+                message: format!("Failed to deserialize entry for key '{}'", log_id(ec_id)),
             })?;
 
         entry.validate().map_err(|message| {
             Report::new(TrustedServerError::KvStore {
                 store_name: store_name.to_owned(),
-                message: format!("Loaded invalid entry for key '{ec_id}': {message}"),
+                message: format!(
+                    "Loaded invalid entry for key '{}': {message}",
+                    log_id(ec_id)
+                ),
             })
         })?;
 
@@ -271,7 +267,7 @@ impl KvIdentityGraph {
         let meta: KvMetadata =
             serde_json::from_slice(&meta_bytes).change_context(TrustedServerError::KvStore {
                 store_name: self.store_name().to_owned(),
-                message: format!("Failed to deserialize metadata for key '{ec_id}'"),
+                message: format!("Failed to deserialize metadata for key '{}'", log_id(ec_id)),
             })?;
 
         Ok(Some(meta))
@@ -694,17 +690,17 @@ impl KvIdentityGraph {
     /// lookup. A lookup may lag, so it can miss a very recent write, and it can
     /// return a row already deleted or expired at the primary — but it cannot
     /// report an identifier this deployment never issued, which is what the
-    /// gate is for. If that cannot answer either, nothing is written and
-    /// [`TombstoneOutcome::Unconfirmed`] is returned with the reason; the caller
-    /// still expires the browser cookie, which is the primary enforcement.
+    /// gate is for. If that cannot answer either, nothing is written and an
+    /// error is returned; the caller still expires the browser cookie, which is
+    /// the primary enforcement.
     ///
     /// # Errors
     ///
-    /// Returns [`TrustedServerError::KvStore`] when the tombstone write itself
-    /// fails. A failure to determine whether the identity exists is reported as
-    /// [`TombstoneOutcome::Unconfirmed`] rather than an error, since nothing was
-    /// written. Callers on the browser path should log at `error` level and
-    /// continue — cookie deletion is the primary enforcement mechanism.
+    /// Returns [`TrustedServerError::KvStore`] when the tombstone write fails,
+    /// and when it cannot be determined whether the identity exists — in that
+    /// case nothing is written. Callers on the browser path should log at
+    /// `error` level and continue: cookie deletion is the primary enforcement
+    /// mechanism.
     pub fn write_withdrawal_tombstone(
         &self,
         ec_id: &str,
@@ -721,22 +717,28 @@ impl KvIdentityGraph {
                 //
                 // The raw form is deliberate: a row whose body no longer
                 // deserializes is still a row, and presence is all that matters.
+                //
+                // Failing to determine existence is an error, not a third
+                // outcome. A caller that only inspects the error case still
+                // reports it, where an extra `Ok` variant would be discarded in
+                // silence.
                 match self.lookup_raw(ec_id) {
-                    Ok(Some(_)) => {}
+                    Ok(Some(_)) => {
+                        log::warn!(
+                            "Confirmed EC ID '{}' by lookup after a list failure",
+                            log_id(ec_id),
+                        );
+                    }
                     Ok(None) => {
-                        return Ok(TombstoneOutcome::Unconfirmed {
-                            reason: format!(
-                                "existence check failed and a lookup found nothing: {list_error:?}"
-                            ),
-                        });
+                        return Err(list_error.attach(
+                            "a lookup found nothing, but it may lag behind a recent write",
+                        ));
                     }
                     Err(lookup_error) => {
-                        return Ok(TombstoneOutcome::Unconfirmed {
-                            reason: format!(
-                                "neither the existence check nor a lookup could be read: \
-                                 {list_error:?}; {lookup_error:?}"
-                            ),
-                        });
+                        return Err(lookup_error.attach(format!(
+                            "the exact check also failed: {}",
+                            list_error.current_context()
+                        )));
                     }
                 }
             }
@@ -755,7 +757,7 @@ impl KvIdentityGraph {
             Ok(_) => Ok(TombstoneOutcome::Written),
             Err(report) => Err(report.change_context(TrustedServerError::KvStore {
                 store_name: self.store_name().to_owned(),
-                message: format!("Failed to write tombstone for key '{ec_id}'"),
+                message: format!("Failed to write tombstone for key '{}'", log_id(ec_id)),
             })),
         }
     }
@@ -1365,6 +1367,24 @@ mod tests {
         assert_eq!(result, UpsertResult::ConsentWithdrawn);
     }
 
+    #[test]
+    fn a_store_error_never_carries_the_whole_identifier() {
+        // Every message in this module goes through `log_id`, so a report that
+        // reaches a log cannot disclose the identifier it is about.
+        let kv = KvIdentityGraph::failing("test_store");
+        let ec_id = format!("{}.ABC123", "a".repeat(64));
+
+        let report = kv
+            .create(&ec_id, &live_entry())
+            .expect_err("the failing store should error");
+
+        let rendered = format!("{report:?}");
+        assert!(
+            !rendered.contains(&ec_id),
+            "a store error must not disclose the identifier: {rendered}"
+        );
+    }
+
     #[test]
     fn write_withdrawal_tombstone_overwrites_live_entry() {
         let kv = KvIdentityGraph::in_memory("test_store");
@@ -1619,12 +1639,9 @@ mod tests {
         let ec_id = format!("{}.ABC123", "8".repeat(64));
 
         assert!(
-            matches!(
-                kv.write_withdrawal_tombstone(&ec_id)
-                    .expect("should resolve the withdrawal"),
-                TombstoneOutcome::Unconfirmed { .. }
-            ),
-            "should refuse to write when existence cannot be established"
+            kv.write_withdrawal_tombstone(&ec_id).is_err(),
+            "an undetermined check is a fault, so a caller inspecting only the \
+             error case still reports it"
         );
         assert!(
             kv.get(&ec_id).expect("should read back").is_none(),
@@ -1632,6 +1649,26 @@ mod tests {
         );
     }
 
+    #[test]
+    fn a_caller_that_only_inspects_the_error_case_still_sees_an_undetermined_check() {
+        // The withdrawal call site is edited by more than one branch. Reporting
+        // an undetermined check through `Err` means the common
+        // `if let Err(..) = ...` shape cannot discard it, where a third `Ok`
+        // variant would be dropped without a compiler complaint.
+        let kv = KvIdentityGraph::new(ListFailingEcKv::new());
+        let ec_id = format!("{}.ABC123", "6".repeat(64));
+
+        let mut reported = false;
+        if let Err(_err) = kv.write_withdrawal_tombstone(&ec_id) {
+            reported = true;
+        }
+
+        assert!(
+            reported,
+            "an undetermined check must reach an error-only caller"
+        );
+    }
+
     #[test]
     fn a_longer_key_does_not_answer_for_the_identity_it_starts_with() {
         let kv = KvIdentityGraph::in_memory("test_store");
diff --git a/crates/trusted-server-core/src/ec/kv_backend.rs b/crates/trusted-server-core/src/ec/kv_backend.rs
index 561835b05..6710befc2 100644
--- a/crates/trusted-server-core/src/ec/kv_backend.rs
+++ b/crates/trusted-server-core/src/ec/kv_backend.rs
@@ -64,12 +64,6 @@ pub enum EcKvWriteOutcome {
     PreconditionFailed,
 }
 
-/// Raw KV store primitives backing the EC identity graph.
-///
-/// Implementations map these operations onto the platform KV API.
-/// Infrastructure failures are reported as [`TrustedServerError::KvStore`];
-/// write precondition failures are part of the normal control flow and are
-/// returned as [`EcKvWriteOutcome::PreconditionFailed`] instead of errors.
 /// Whether `key` appears exactly in a paged listing.
 ///
 /// Backends that can only match by prefix use this to decide existence: a
@@ -96,6 +90,12 @@ pub fn contains_exact_key<E>(
     Ok(false)
 }
 
+/// Raw KV store primitives backing the EC identity graph.
+///
+/// Implementations map these operations onto the platform KV API.
+/// Infrastructure failures are reported as [`TrustedServerError::KvStore`];
+/// write precondition failures are part of the normal control flow and are
+/// returned as [`EcKvWriteOutcome::PreconditionFailed`] instead of errors.
 pub trait EcKvStore {
     /// Returns the platform store name, used in log and error messages.
     fn store_name(&self) -> &str;

From 0d22e198aa57826f2a9bef167d20be94b2fa5af5 Mon Sep 17 00:00:00 2001
From: prk-Jr <prakashsagwara@gmail.com>
Date: Wed, 2 Sep 2026 20:56:42 +0530
Subject: [PATCH 08/11] Answer an exhausted prefix listing as undetermined, not
 absent
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The exact-key check followed a listing for a bounded number of pages and read
running out of budget as absence. Absence is what tells the withdrawal path
the identity was never issued, so a real identity on an unread page had its
tombstone silently dropped — while the constant's own note and the tombstone
docs both said the caller would treat that case as unconfirmed.

Report it as a third outcome and map it to an error at the adapter, which puts
it on the path that already re-checks by lookup. The page budget moves into
the checked function so the listing is passed untruncated and there is no
count to keep in agreement at the call site.
---
 .../src/ec_kv.rs                              |  28 +++--
 .../trusted-server-core/src/ec/kv_backend.rs  | 100 +++++++++++++++---
 2 files changed, 108 insertions(+), 20 deletions(-)

diff --git a/crates/trusted-server-adapter-fastly/src/ec_kv.rs b/crates/trusted-server-adapter-fastly/src/ec_kv.rs
index c63d6ef49..87ecf119c 100644
--- a/crates/trusted-server-adapter-fastly/src/ec_kv.rs
+++ b/crates/trusted-server-adapter-fastly/src/ec_kv.rs
@@ -7,7 +7,8 @@
 use error_stack::{Report, ResultExt};
 use fastly::kv_store::{InsertMode, KVStore};
 use trusted_server_core::ec::kv_backend::{
-    EcKvLookup, EcKvStore, EcKvWrite, EcKvWriteMode, EcKvWriteOutcome, contains_exact_key,
+    EcKvLookup, EcKvStore, EcKvWrite, EcKvWriteMode, EcKvWriteOutcome, ExactKeyMatch,
+    contains_exact_key,
 };
 use trusted_server_core::ec::log_id;
 use trusted_server_core::error::TrustedServerError;
@@ -49,9 +50,10 @@ const EXACT_MATCH_PAGE_SIZE: u32 = 100;
 ///
 /// A well-formed identifier is the whole key, so at most a handful of keys can
 /// carry it as a prefix and one page is the normal case. The cap stops a short
-/// prefix from walking the keyspace on the response path; exceeding it yields
-/// no match rather than a wrong answer, and the caller treats that as
-/// unconfirmed.
+/// prefix from walking the keyspace on the response path. Exceeding it is
+/// reported as an error rather than as absence, so the caller re-checks by
+/// lookup instead of discarding a withdrawal for an identity that may be held
+/// on a page this never read.
 const EXACT_MATCH_MAX_PAGES: usize = 4;
 
 impl EcKvStore for FastlyEcKvStore {
@@ -147,7 +149,9 @@ impl EcKvStore for FastlyEcKvStore {
             .prefix(key)
             .limit(EXACT_MATCH_PAGE_SIZE)
             .iter()
-            .take(EXACT_MATCH_MAX_PAGES)
+            // Unbounded here on purpose: `contains_exact_key` stops reading at
+            // its budget, so the page count lives with the code that is tested
+            // against it rather than being recomputed at the call site.
             .map(|page| match page {
                 // A store that reports nothing under the prefix is telling us
                 // the key is absent, which is what `lookup` already does for a
@@ -162,7 +166,19 @@ impl EcKvStore for FastlyEcKvStore {
                     }),
             });
 
-        contains_exact_key(pages, key)
+        match contains_exact_key(pages, key, EXACT_MATCH_MAX_PAGES)? {
+            ExactKeyMatch::Found => Ok(true),
+            ExactKeyMatch::Absent => Ok(false),
+            // Not an answer, so it must not be returned as one. The caller
+            // falls back to a lookup on an error.
+            ExactKeyMatch::Undetermined => Err(Report::new(TrustedServerError::KvStore {
+                store_name: self.store_name.clone(),
+                message: format!(
+                    "Listing for key '{}' exceeded {EXACT_MATCH_MAX_PAGES} pages, so existence is undetermined",
+                    log_id(key)
+                ),
+            })),
+        }
     }
 
     fn delete(&self, key: &str) -> Result<(), Report<TrustedServerError>> {
diff --git a/crates/trusted-server-core/src/ec/kv_backend.rs b/crates/trusted-server-core/src/ec/kv_backend.rs
index 6710befc2..881c615bf 100644
--- a/crates/trusted-server-core/src/ec/kv_backend.rs
+++ b/crates/trusted-server-core/src/ec/kv_backend.rs
@@ -74,6 +74,14 @@ pub enum EcKvWriteOutcome {
 /// key lands in the first page when other keys share its prefix, and stopping
 /// early would report a held identity as missing.
 ///
+/// `max_pages` bounds how far the listing is followed. Running out of budget
+/// is reported as [`ExactKeyMatch::Undetermined`] rather than as absence: the
+/// key may sit on a page that was never read, and answering "absent" would
+/// discard a withdrawal for an identity the store actually holds. The listing
+/// is read lazily and at most one page beyond `max_pages` — just far enough to
+/// learn that it continues — so callers pass their listing untruncated rather
+/// than pre-trimming it to a count that has to agree with this one.
+///
 /// # Errors
 ///
 /// Returns the first page error, so a listing that cannot be read is never
@@ -81,13 +89,29 @@ pub enum EcKvWriteOutcome {
 pub fn contains_exact_key<E>(
     pages: impl IntoIterator<Item = Result<Vec<String>, E>>,
     key: &str,
-) -> Result<bool, E> {
-    for page in pages {
+    max_pages: usize,
+) -> Result<ExactKeyMatch, E> {
+    for (index, page) in pages.into_iter().enumerate() {
+        if index >= max_pages {
+            return Ok(ExactKeyMatch::Undetermined);
+        }
         if page?.iter().any(|listed| listed == key) {
-            return Ok(true);
+            return Ok(ExactKeyMatch::Found);
         }
     }
-    Ok(false)
+    Ok(ExactKeyMatch::Absent)
+}
+
+/// What a bounded prefix listing established about one exact key.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum ExactKeyMatch {
+    /// The listing contained the key.
+    Found,
+    /// The listing was read to its end and did not contain the key.
+    Absent,
+    /// The listing was longer than the budget allowed, so the key's absence
+    /// was never established.
+    Undetermined,
 }
 
 /// Raw KV store primitives backing the EC identity graph.
@@ -310,7 +334,7 @@ pub(crate) mod test_support {
 
 #[cfg(test)]
 mod tests {
-    use super::contains_exact_key;
+    use super::{ExactKeyMatch, contains_exact_key};
 
     /// Page iterator that records how many pages were consumed.
     struct CountingPages {
@@ -355,8 +379,9 @@ mod tests {
             page(&["wanted"]),
         ]);
 
-        assert!(
-            contains_exact_key(pages, "wanted").expect("should scan the pages"),
+        assert_eq!(
+            contains_exact_key(pages, "wanted", 8).expect("should scan the pages"),
+            ExactKeyMatch::Found,
             "a match on the last page must still be found"
         );
     }
@@ -365,8 +390,9 @@ mod tests {
     fn ignores_keys_that_only_start_with_the_one_asked_for() {
         let (pages, _) = counting(vec![page(&["wanted-suffix", "wantedx", "wanted2"])]);
 
-        assert!(
-            !contains_exact_key(pages, "wanted").expect("should scan the pages"),
+        assert_eq!(
+            contains_exact_key(pages, "wanted", 8).expect("should scan the pages"),
+            ExactKeyMatch::Absent,
             "a longer key is a different identity"
         );
     }
@@ -379,8 +405,9 @@ mod tests {
             page(&["never-read-either"]),
         ]);
 
-        assert!(
-            contains_exact_key(pages, "wanted").expect("should scan the pages"),
+        assert_eq!(
+            contains_exact_key(pages, "wanted", 8).expect("should scan the pages"),
+            ExactKeyMatch::Found,
             "should find the match"
         );
         assert_eq!(consumed.get(), 1, "should not read past the match");
@@ -391,18 +418,63 @@ mod tests {
         let (pages, _) = counting(vec![page(&["wanted-suffix"]), Err("list unavailable")]);
 
         assert_eq!(
-            contains_exact_key(pages, "wanted"),
+            contains_exact_key(pages, "wanted", 8),
             Err("list unavailable"),
             "an unreadable listing must not read as absent"
         );
     }
 
+    #[test]
+    fn reports_undetermined_when_the_listing_outruns_the_budget() {
+        // The key sits past the budget, which is exactly the case that must
+        // not read as absent: answering "absent" discards the withdrawal.
+        let (pages, consumed) = counting(vec![
+            page(&["wanted-suffix-a"]),
+            page(&["wanted-suffix-b"]),
+            page(&["wanted"]),
+        ]);
+
+        assert_eq!(
+            contains_exact_key(pages, "wanted", 2).expect("should scan the pages"),
+            ExactKeyMatch::Undetermined,
+            "a key beyond the budget must not read as absent"
+        );
+        assert_eq!(
+            consumed.get(),
+            3,
+            "should read one page past the budget to learn the listing continues"
+        );
+    }
+
+    #[test]
+    fn reports_absent_for_a_listing_that_ends_exactly_at_the_budget() {
+        let (pages, _) = counting(vec![page(&["wanted-suffix-a"]), page(&["wanted-suffix-b"])]);
+
+        assert_eq!(
+            contains_exact_key(pages, "wanted", 2).expect("should scan the pages"),
+            ExactKeyMatch::Absent,
+            "a listing that fits the budget is answered definitively"
+        );
+    }
+
+    #[test]
+    fn finds_a_match_on_the_last_page_within_the_budget() {
+        let (pages, _) = counting(vec![page(&["wanted-suffix-a"]), page(&["wanted"])]);
+
+        assert_eq!(
+            contains_exact_key(pages, "wanted", 2).expect("should scan the pages"),
+            ExactKeyMatch::Found,
+            "the final permitted page still counts"
+        );
+    }
+
     #[test]
     fn reports_absent_for_an_empty_listing() {
         let (pages, _) = counting(Vec::new());
 
-        assert!(
-            !contains_exact_key(pages, "wanted").expect("should scan the pages"),
+        assert_eq!(
+            contains_exact_key(pages, "wanted", 8).expect("should scan the pages"),
+            ExactKeyMatch::Absent,
             "nothing listed means nothing held"
         );
     }

From 44d904d2873afc5daed5ab6cdda7549d7e2941c0 Mon Sep 17 00:00:00 2001
From: prk-Jr <prakashsagwara@gmail.com>
Date: Wed, 2 Sep 2026 20:56:56 +0530
Subject: [PATCH 09/11] Redact the identifier in errors this module builds
 itself

Nine messages interpolated the whole identifier: a duplicate create, upserts
naming a missing or withdrawn key, and the CAS-exhaustion paths. Callers log
these reports with debug formatting, so each one put a full identifier in a
log line. The existing test only drove an injected backend failure, which
never reaches them, so the module's claim that every message goes through the
truncating helper held for the wrong reason.

Route them through it too, and cover the paths a request can actually reach.
---
 crates/trusted-server-core/src/ec/kv.rs | 62 +++++++++++++++++++++----
 1 file changed, 53 insertions(+), 9 deletions(-)

diff --git a/crates/trusted-server-core/src/ec/kv.rs b/crates/trusted-server-core/src/ec/kv.rs
index dc47c4ab2..1b676fce5 100644
--- a/crates/trusted-server-core/src/ec/kv.rs
+++ b/crates/trusted-server-core/src/ec/kv.rs
@@ -287,7 +287,7 @@ impl KvIdentityGraph {
         match self.write_entry(ec_id, &body, &meta_str, ENTRY_TTL, EcKvWriteMode::Add)? {
             EcKvWriteOutcome::Written => Ok(()),
             EcKvWriteOutcome::PreconditionFailed => {
-                Err(self.kv_error(format!("Key '{ec_id}' already exists")))
+                Err(self.kv_error(format!("Key '{}' already exists", log_id(ec_id))))
             }
         }
     }
@@ -398,7 +398,8 @@ impl KvIdentityGraph {
         }
 
         Err(self.kv_error(format!(
-            "CAS conflict after {MAX_CAS_RETRIES} retries reviving tombstone for '{ec_id}'"
+            "CAS conflict after {MAX_CAS_RETRIES} retries reviving tombstone for '{}'",
+            log_id(ec_id),
         )))
     }
 
@@ -432,8 +433,9 @@ impl KvIdentityGraph {
                         updates.len(),
                     );
                     return Err(self.kv_error(format!(
-                        "Cannot upsert {} partner IDs for missing key '{ec_id}'",
+                        "Cannot upsert {} partner IDs for missing key '{}'",
                         updates.len(),
+                        log_id(ec_id),
                     )));
                 }
             };
@@ -447,8 +449,9 @@ impl KvIdentityGraph {
                     updates.len(),
                 );
                 return Err(self.kv_error(format!(
-                    "Cannot upsert {} partner IDs for withdrawn key '{ec_id}'",
+                    "Cannot upsert {} partner IDs for withdrawn key '{}'",
                     updates.len(),
+                    log_id(ec_id),
                 )));
             }
 
@@ -478,8 +481,9 @@ impl KvIdentityGraph {
         }
 
         Err(self.kv_error(format!(
-            "CAS conflict after {MAX_CAS_RETRIES} retries upserting {} partner IDs for '{ec_id}'",
+            "CAS conflict after {MAX_CAS_RETRIES} retries upserting {} partner IDs for '{}'",
             updates.len(),
+            log_id(ec_id),
         )))
     }
 
@@ -510,7 +514,8 @@ impl KvIdentityGraph {
                         log_id(ec_id)
                     );
                     return Err(self.kv_error(format!(
-                        "Cannot upsert partner '{partner_id}' for missing key '{ec_id}'"
+                        "Cannot upsert partner '{partner_id}' for missing key '{}'",
+                        log_id(ec_id),
                     )));
                 }
             };
@@ -523,7 +528,8 @@ impl KvIdentityGraph {
                     log_id(ec_id),
                 );
                 return Err(self.kv_error(format!(
-                    "Cannot upsert partner '{partner_id}' for withdrawn key '{ec_id}'"
+                    "Cannot upsert partner '{partner_id}' for withdrawn key '{}'",
+                    log_id(ec_id),
                 )));
             }
 
@@ -566,7 +572,8 @@ impl KvIdentityGraph {
         }
 
         Err(self.kv_error(format!(
-            "CAS conflict after {MAX_CAS_RETRIES} retries upserting partner '{partner_id}' for '{ec_id}'"
+            "CAS conflict after {MAX_CAS_RETRIES} retries upserting partner '{partner_id}' for '{}'",
+            log_id(ec_id),
         )))
     }
 
@@ -636,7 +643,8 @@ impl KvIdentityGraph {
         }
 
         Err(self.kv_error(format!(
-            "CAS conflict after {MAX_CAS_RETRIES} retries upserting partner '{partner_id}' for '{ec_id}'"
+            "CAS conflict after {MAX_CAS_RETRIES} retries upserting partner '{partner_id}' for '{}'",
+            log_id(ec_id),
         )))
     }
 
@@ -1385,6 +1393,42 @@ mod tests {
         );
     }
 
+    #[test]
+    fn a_locally_built_error_never_carries_the_whole_identifier() {
+        // The injected-failure case above covers errors the backend produces.
+        // These are built in this module from the identifier itself, on paths
+        // a request can reach: a duplicate create, and an upsert naming a key
+        // the store does not hold.
+        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 duplicate = kv
+            .create(&ec_id, &live_entry())
+            .expect_err("a second create should be refused");
+        let missing = kv
+            .upsert_partner_id(&format!("{}.ZZZ999", "b".repeat(64)), "partner", "uid")
+            .expect_err("an upsert on a missing key should be refused");
+        let withdrawn = {
+            kv.write_withdrawal_tombstone(&ec_id)
+                .expect("should tombstone");
+            kv.upsert_partner_id(&ec_id, "partner", "uid")
+                .expect_err("an upsert on a withdrawn key should be refused")
+        };
+
+        for (label, report) in [
+            ("duplicate create", duplicate),
+            ("missing key", missing),
+            ("withdrawn key", withdrawn),
+        ] {
+            let rendered = format!("{report:?}");
+            assert!(
+                !rendered.contains(&ec_id) && !rendered.contains(&"b".repeat(64)),
+                "the {label} error must not disclose the identifier: {rendered}"
+            );
+        }
+    }
+
     #[test]
     fn write_withdrawal_tombstone_overwrites_live_entry() {
         let kv = KvIdentityGraph::in_memory("test_store");

From 6cce0d31629a249029fb04da9f7827f203b69895 Mon Sep 17 00:00:00 2001
From: prk-Jr <prakashsagwara@gmail.com>
Date: Wed, 2 Sep 2026 21:25:45 +0530
Subject: [PATCH 10/11] Cover every self-built error in the redaction test

The test drove three of the message paths, so the other five held only by
inspection. Extend it to the batched upserts and the three CAS-exhaustion
terminal errors, which needed a conflict-injecting store that can hold a live
entry rather than only a tombstone.
---
 crates/trusted-server-core/src/ec/kv.rs | 62 +++++++++++++++++++++++--
 1 file changed, 59 insertions(+), 3 deletions(-)

diff --git a/crates/trusted-server-core/src/ec/kv.rs b/crates/trusted-server-core/src/ec/kv.rs
index 1b676fce5..f99315d1a 100644
--- a/crates/trusted-server-core/src/ec/kv.rs
+++ b/crates/trusted-server-core/src/ec/kv.rs
@@ -1042,6 +1042,23 @@ mod tests {
                 )
                 .expect("should seed tombstone");
         }
+
+        fn seed_live(&self, ec_id: &str) {
+            let (body, meta) =
+                KvIdentityGraph::serialize_entry(&live_entry(), self.inner.store_name())
+                    .expect("should serialize live entry");
+            self.inner
+                .insert(
+                    ec_id,
+                    EcKvWrite {
+                        body: &body,
+                        metadata: &meta,
+                        ttl: TOMBSTONE_TTL,
+                        mode: EcKvWriteMode::Add,
+                    },
+                )
+                .expect("should seed live entry");
+        }
     }
 
     impl EcKvStore for ConflictInjectingEcKv {
@@ -1396,9 +1413,10 @@ mod tests {
     #[test]
     fn a_locally_built_error_never_carries_the_whole_identifier() {
         // The injected-failure case above covers errors the backend produces.
-        // These are built in this module from the identifier itself, on paths
-        // a request can reach: a duplicate create, and an upsert naming a key
-        // the store does not hold.
+        // These are built in this module from the identifier itself, on every
+        // path a request can reach: a duplicate create, single and batched
+        // upserts naming a key the store does not hold or has withdrawn, and
+        // the CAS-exhaustion terminal errors.
         let kv = KvIdentityGraph::in_memory("test_store");
         let ec_id = format!("{}.ABC123", "a".repeat(64));
         kv.create(&ec_id, &live_entry()).expect("should create");
@@ -1409,17 +1427,55 @@ mod tests {
         let missing = kv
             .upsert_partner_id(&format!("{}.ZZZ999", "b".repeat(64)), "partner", "uid")
             .expect_err("an upsert on a missing key should be refused");
+        let batched_missing = kv
+            .upsert_partner_ids(
+                &format!("{}.ZZZ999", "b".repeat(64)),
+                &[PartnerIdUpdate::new("partner", "uid")],
+            )
+            .expect_err("a batched upsert on a missing key should be refused");
         let withdrawn = {
             kv.write_withdrawal_tombstone(&ec_id)
                 .expect("should tombstone");
             kv.upsert_partner_id(&ec_id, "partner", "uid")
                 .expect_err("an upsert on a withdrawn key should be refused")
         };
+        let batched_withdrawn = kv
+            .upsert_partner_ids(&ec_id, &[PartnerIdUpdate::new("partner", "uid")])
+            .expect_err("a batched upsert on a withdrawn key should be refused");
+
+        // The CAS-exhaustion paths build their message the same way, and a
+        // store that never lets a write land is the only way to reach them.
+        let cas_revive = {
+            let store = ConflictInjectingEcKv::new(MAX_CAS_RETRIES + 1, false);
+            store.seed_tombstone(&ec_id);
+            KvIdentityGraph::new(store)
+                .create_or_revive(&ec_id, &live_entry())
+                .expect_err("should exhaust CAS retries")
+        };
+        let cas_upsert = {
+            let store = ConflictInjectingEcKv::new(MAX_CAS_RETRIES + 1, false);
+            store.seed_live(&ec_id);
+            KvIdentityGraph::new(store)
+                .upsert_partner_id(&ec_id, "partner", "uid")
+                .expect_err("should exhaust CAS retries")
+        };
+        let cas_batched = {
+            let store = ConflictInjectingEcKv::new(MAX_CAS_RETRIES + 1, false);
+            store.seed_live(&ec_id);
+            KvIdentityGraph::new(store)
+                .upsert_partner_ids(&ec_id, &[PartnerIdUpdate::new("partner", "uid")])
+                .expect_err("should exhaust CAS retries")
+        };
 
         for (label, report) in [
             ("duplicate create", duplicate),
             ("missing key", missing),
+            ("batched missing key", batched_missing),
             ("withdrawn key", withdrawn),
+            ("batched withdrawn key", batched_withdrawn),
+            ("CAS exhaustion reviving", cas_revive),
+            ("CAS exhaustion upserting", cas_upsert),
+            ("CAS exhaustion batch upserting", cas_batched),
         ] {
             let rendered = format!("{report:?}");
             assert!(

From b36eb1e43a0eec9c3588106af804b3c0421f264e Mon Sep 17 00:00:00 2001
From: prk-Jr <prakashsagwara@gmail.com>
Date: Thu, 3 Sep 2026 09:59:04 +0530
Subject: [PATCH 11/11] Reach the last redacted message from the test

The conditional partner upsert's CAS-exhaustion error used the redacted template but no test executed it, so it was the one message still holding by inspection alone.
---
 crates/trusted-server-core/src/ec/kv.rs | 8 ++++++++
 1 file changed, 8 insertions(+)

diff --git a/crates/trusted-server-core/src/ec/kv.rs b/crates/trusted-server-core/src/ec/kv.rs
index f99315d1a..c4aa9647c 100644
--- a/crates/trusted-server-core/src/ec/kv.rs
+++ b/crates/trusted-server-core/src/ec/kv.rs
@@ -1466,6 +1466,13 @@ mod tests {
                 .upsert_partner_ids(&ec_id, &[PartnerIdUpdate::new("partner", "uid")])
                 .expect_err("should exhaust CAS retries")
         };
+        let cas_if_exists = {
+            let store = ConflictInjectingEcKv::new(MAX_CAS_RETRIES + 1, false);
+            store.seed_live(&ec_id);
+            KvIdentityGraph::new(store)
+                .upsert_partner_id_if_exists(&ec_id, "partner", "uid")
+                .expect_err("should exhaust CAS retries")
+        };
 
         for (label, report) in [
             ("duplicate create", duplicate),
@@ -1476,6 +1483,7 @@ mod tests {
             ("CAS exhaustion reviving", cas_revive),
             ("CAS exhaustion upserting", cas_upsert),
             ("CAS exhaustion batch upserting", cas_batched),
+            ("CAS exhaustion upserting if present", cas_if_exists),
         ] {
             let rendered = format!("{report:?}");
             assert!(