fix(cache): re-check ownership after the disk write, not only before

src/core/src/cache/core.rs

@@ -1138,6 +1138,7 @@ impl LiquidCache {
             self.budget.release_disk(len);
             return Ok(());
         }
+        let is_rewrite = matches!(identity, WriteIdentity::Rewrite(_));
         let identity = identity.value();
         self.store
             .put(entry_id_to_key(&entry_id, identity), bytes.to_vec())
@@ -1153,6 +1154,34 @@ impl LiquidCache {
             },
             CacheEntry::DiskLiquid { .. } | CacheEntry::MemoryLiquid(_) => DiskKind::Liquid,
         };
+        // The check above ran before the await, so a takeover can have landed
+        // while the bytes were being written. Ask the index again, not the
+        // record: the record may still name a previous owner this writer is
+        // legitimately superseding, whereas the index names whoever the bytes
+        // can actually be read by. A rewrite that is no longer that owner is
+        // stale and must leave the current one's record and object alone.
+        //
+        // The index read and the swap below are two steps, so this narrows the
+        // window rather than closing it — as with every other check-then-act
+        // pair in this file.
+        if is_rewrite
+            && self
+                .index
+                .get_with_identity(&entry_id)
+                .is_some_and(|(current, _)| current != identity)
+        {
+            // Stale: another identity owns the key now. Remove the object this
+            // write just made — nothing names it — and hand back its
+            // reservation. The remove holds no claim on the key, so a second
+            // caller may race it; `t4::Store::remove` reports a missing key as
+            // `Ok(false)` rather than an error, so that is harmless.
+            self.store
+                .remove(&entry_id_to_key(&entry_id, identity))
+                .await
+                .expect("disk remove failed");
+            self.budget.release_disk(len);
+            return Ok(());
+        }
         let previous = self.disk_copies.lock().unwrap().insert(
             entry_id,
             DiskCopy {
@@ -1165,8 +1194,8 @@ impl LiquidCache {
             // Same owner: the put replaced that object, so its reservation
             // goes with it and there is nothing left to delete.
             //
-            // Different owner: this is the current owner superseding one that
-            // has let the key go (a stale writer never reaches here). The key
+            // Different owner: this is a caller taking the key from one that
+            // has let it go — a stale rewrite was turned away above. The key
             // carries the identity, so the put landed somewhere else and the
             // previous object is still there — with no record naming it and
             // nothing that would ever reach it. Releasing its reservation