Address PR review: improve error context, guards, and testability

- Pass store_name to serialize_entry for meaningful error messages
- Add attempt number to upsert_partner_id race-retry log
- Add explicit backwards-timestamp guard in update_last_seen
  (previously absorbed silently into debounce check)
- Reuse store handle and serialized data in create_or_revive
  CAS loop instead of re-opening and re-serializing
- Add comment explaining intentional dual store handles in
  upsert_partner_id (get() opens its own; outer is for writes)
- Document tombstone created-field reset as intentional design
- Add PartialEq derive to all schema types for test ergonomics

Author: ChristianPavilonis

crates/trusted-server-core/src/ec/kv.rs

@@ -72,15 +72,18 @@ impl KvIdentityGraph {
     }
 
     /// Serializes an entry body and metadata for insertion.
-    fn serialize_entry(entry: &KvEntry) -> Result<(String, String), Report<TrustedServerError>> {
+    fn serialize_entry(
+        entry: &KvEntry,
+        store_name: &str,
+    ) -> Result<(String, String), Report<TrustedServerError>> {
         let body = serde_json::to_string(entry).change_context(TrustedServerError::KvStore {
-            store_name: String::new(),
+            store_name: store_name.to_owned(),
             message: "Failed to serialize KV entry body".to_owned(),
         })?;
         let meta = KvMetadata::from_entry(entry);
         let meta_str =
             serde_json::to_string(&meta).change_context(TrustedServerError::KvStore {
-                store_name: String::new(),
+                store_name: store_name.to_owned(),
                 message: "Failed to serialize KV entry metadata".to_owned(),
             })?;
         Ok((body, meta_str))
@@ -169,7 +172,7 @@ impl KvIdentityGraph {
     /// key already exists (`ItemPreconditionFailed`).
     pub fn create(&self, ec_hash: &str, entry: &KvEntry) -> Result<(), Report<TrustedServerError>> {
         let store = self.open_store()?;
-        let (body, meta_str) = Self::serialize_entry(entry)?;
+        let (body, meta_str) = Self::serialize_entry(entry, &self.store_name)?;
         let created = Self::try_insert_add(&store, ec_hash, &body, &meta_str, &self.store_name)?;
         if created {
             Ok(())
@@ -231,9 +234,11 @@ impl KvIdentityGraph {
         ec_hash: &str,
         entry: &KvEntry,
     ) -> Result<(), Report<TrustedServerError>> {
-        // Try create first — fast path for new entries.
+        // Serialize once and reuse across the fast path and CAS loop.
         let store = self.open_store()?;
-        let (body, meta_str) = Self::serialize_entry(entry)?;
+        let (body, meta_str) = Self::serialize_entry(entry, &self.store_name)?;
+
+        // Try create first — fast path for new entries.
         if Self::try_insert_add(&store, ec_hash, &body, &meta_str, &self.store_name)? {
             return Ok(());
         }
@@ -253,8 +258,6 @@ impl KvIdentityGraph {
 
         // Tombstone — CAS overwrite to revive.
         log::info!("create_or_revive: reviving tombstone for '{ec_hash}'");
-        let store = self.open_store()?;
-        let (body, meta_str) = Self::serialize_entry(entry)?;
 
         let mut current_gen = generation;
         for attempt in 0..MAX_CAS_RETRIES {
@@ -325,6 +328,10 @@ impl KvIdentityGraph {
         uid: &str,
         synced: u64,
     ) -> Result<(), Report<TrustedServerError>> {
+        // Open store once for write operations. Note: `self.get()` opens
+        // its own handle internally — this is intentional since `KVStore::open`
+        // is a cheap name lookup, and keeping the read/write APIs independent
+        // simplifies the method signatures.
         let store = self.open_store()?;
 
         for attempt in 0..MAX_CAS_RETRIES {
@@ -336,7 +343,7 @@ impl KvIdentityGraph {
                         "upsert_partner_id: no entry for '{ec_hash}', creating minimal entry"
                     );
                     let minimal = KvEntry::minimal(partner_id, uid, synced);
-                    let (min_body, min_meta) = Self::serialize_entry(&minimal)?;
+                    let (min_body, min_meta) = Self::serialize_entry(&minimal, &self.store_name)?;
                     if Self::try_insert_add(
                         &store,
                         ec_hash,
@@ -348,7 +355,9 @@ impl KvIdentityGraph {
                     }
                     // Key appeared between get() and create — re-read on next iteration.
                     log::debug!(
-                        "upsert_partner_id: minimal create raced for '{ec_hash}', retrying"
+                        "upsert_partner_id: minimal create raced for '{ec_hash}', retrying (attempt {}/{})",
+                        attempt + 1,
+                        MAX_CAS_RETRIES,
                     );
                     continue;
                 }
@@ -377,7 +386,7 @@ impl KvIdentityGraph {
                 },
             );
 
-            let (body, meta_str) = Self::serialize_entry(&entry)?;
+            let (body, meta_str) = Self::serialize_entry(&entry, &self.store_name)?;
 
             match store
                 .build_insert()
@@ -448,18 +457,27 @@ impl KvIdentityGraph {
             return Ok(());
         }
 
+        // Guard against stale/out-of-order timestamps.
+        if timestamp <= entry.last_seen {
+            log::trace!(
+                "update_last_seen: stale timestamp for '{ec_hash}' (stored={}, incoming={timestamp})",
+                entry.last_seen,
+            );
+            return Ok(());
+        }
+
         // Debounce: skip if the stored value is recent enough.
-        if timestamp.saturating_sub(entry.last_seen) < LAST_SEEN_DEBOUNCE_SECS {
+        if timestamp - entry.last_seen < LAST_SEEN_DEBOUNCE_SECS {
             log::trace!(
                 "update_last_seen: debounced for '{ec_hash}' (delta={}s)",
-                timestamp.saturating_sub(entry.last_seen),
+                timestamp - entry.last_seen,
             );
             return Ok(());
         }
 
         entry.last_seen = timestamp;
         let store = self.open_store()?;
-        let (body, meta_str) = Self::serialize_entry(&entry)?;
+        let (body, meta_str) = Self::serialize_entry(&entry, &self.store_name)?;
 
         store
             .build_insert()
@@ -493,7 +511,7 @@ impl KvIdentityGraph {
     ) -> Result<(), Report<TrustedServerError>> {
         let store = self.open_store()?;
         let entry = KvEntry::tombstone(current_timestamp());
-        let (body, meta_str) = Self::serialize_entry(&entry)?;
+        let (body, meta_str) = Self::serialize_entry(&entry, &self.store_name)?;
 
         store
             .build_insert()
@@ -557,7 +575,7 @@ mod tests {
     fn serialize_entry_produces_valid_json() {
         let entry = KvEntry::tombstone(1000);
         let (body, meta) =
-            KvIdentityGraph::serialize_entry(&entry).expect("should serialize entry");
+            KvIdentityGraph::serialize_entry(&entry, "test-store").expect("should serialize entry");
 
         // Verify body is valid JSON.
         let _: KvEntry =

crates/trusted-server-core/src/ec/kv_types.rs

@@ -20,7 +20,7 @@ pub const SCHEMA_VERSION: u8 = 1;
 ///
 /// **KV key:** 64-character hex hash (the stable prefix from the EC ID).
 /// **KV value:** JSON-serialized `KvEntry` (max ~5KB).
-#[derive(Debug, Clone, Serialize, Deserialize)]
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 pub struct KvEntry {
     /// Schema version — always [`SCHEMA_VERSION`].
     pub v: u8,
@@ -41,7 +41,7 @@ pub struct KvEntry {
 }
 
 /// Consent state within a KV entry.
-#[derive(Debug, Clone, Serialize, Deserialize)]
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 pub struct KvConsent {
     /// Raw TCF v2 consent string.
     #[serde(default, skip_serializing_if = "Option::is_none")]
@@ -56,7 +56,7 @@ pub struct KvConsent {
 }
 
 /// Geo location within a KV entry.
-#[derive(Debug, Clone, Serialize, Deserialize)]
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 pub struct KvGeo {
     /// ISO 3166-1 alpha-2 country code (e.g. `"US"`).
     pub country: String,
@@ -66,7 +66,7 @@ pub struct KvGeo {
 }
 
 /// A synced partner user ID within a KV entry.
-#[derive(Debug, Clone, Serialize, Deserialize)]
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 pub struct KvPartnerId {
     /// The partner's user identifier.
     pub uid: String,
@@ -78,7 +78,7 @@ pub struct KvPartnerId {
 ///
 /// Fastly KV metadata is limited to 2048 bytes and can be read without
 /// streaming the full body. Used by batch sync for fast consent checks.
-#[derive(Debug, Clone, Serialize, Deserialize)]
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 pub struct KvMetadata {
     /// Mirrors [`KvConsent::ok`] — `false` means tombstone.
     pub ok: bool,
@@ -144,6 +144,10 @@ impl KvEntry {
     ///
     /// Sets `consent.ok = false`, clears all partner IDs, and uses a
     /// placeholder geo. The caller should apply a 24-hour TTL when writing.
+    ///
+    /// **Note:** The original `created` timestamp is intentionally not
+    /// preserved — reading the existing entry first would add latency on
+    /// the consent-withdrawal hot path, and the tombstone expires in 24h.
     #[must_use]
     pub fn tombstone(now: u64) -> Self {
         Self {