diff --git a/crates/didbot-pds/src/blobs.rs b/crates/didbot-pds/src/blobs.rs
index 63e59ed7..535e653c 100644
--- a/crates/didbot-pds/src/blobs.rs
+++ b/crates/didbot-pds/src/blobs.rs
@@ -543,10 +543,9 @@ pub trait BlobStore: Send + Sync {
     /// The write-time half of reference resolution: a caller that has just
     /// written a record walks it for blob references — see
     /// [`crate::records::blob_refs`] — and marks each one here. A CID that
-    /// names nothing this account holds is ignored rather than refused: a
-    /// record naming a blob that was never uploaded, or was uploaded to a
-    /// different account, is not this call's problem to raise, and a
-    /// reference count is the only thing kept here.
+    /// names nothing this account holds is ignored: this runs after the
+    /// record is in the repository and has no way to refuse, which is what
+    /// [`BlobStore::missing_blobs`] is asked before the write for.
     ///
     /// Calling this twice for the same CID from the same record would double
     /// its count; callers are expected to call it once per record that holds
@@ -554,6 +553,16 @@ pub trait BlobStore: Send + Sync {
     /// dedupes within one record.
     fn mark_referenced(&self, did: &str, cids: &[String]);
 
+    /// Which of `cids` this account does not hold, in the order given.
+    ///
+    /// Asked before a write lands, because [`BlobStore::mark_referenced`]
+    /// cannot refuse: it is called after the record is in the repository, so
+    /// by the time it meets a CID nothing holds, the client has already been
+    /// told its write succeeded. A record naming a blob the account does not
+    /// hold is a record whose image is a permanent `404`, and the reference
+    /// implementation refuses such a write rather than storing it.
+    fn missing_blobs(&self, did: &str, cids: &[String]) -> Vec<String>;
+
     /// The inverse of [`BlobStore::mark_referenced`]: one fewer live record
     /// holds each of `cids`.
     ///
@@ -829,7 +838,17 @@ impl BlobIndex {
                 cid: reference.cid,
             });
         }
-        if let Some(held) = account.blobs.get(&reference.cid) {
+        if let Some(held) = account.blobs.get_mut(&reference.cid) {
+            // The clock starts again. A client re-uploading a blob it is
+            // about to reference is asking for exactly the grace a first
+            // upload gets, and answering `200` while leaving the old age in
+            // place means the next sweep collects it anyway — the client is
+            // told its blob is fine and loses it regardless. Only while
+            // nothing references it: a referenced blob is not collectable,
+            // and moving its age would hide how long it has been held.
+            if held.refs == 0 {
+                held.uploaded_at = uploaded_at;
+            }
             return Ok(Some(held.reference.clone()));
         }
         if account.bytes.saturating_add(reference.size) > limits.account_quota_bytes {
@@ -901,6 +920,17 @@ impl BlobIndex {
         }
     }
 
+    /// Which of `cids` this account does not hold. See
+    /// [`BlobStore::missing_blobs`].
+    pub(crate) fn missing_blobs(&self, did: &str, cids: &[String]) -> Vec<String> {
+        let accounts = self.lock();
+        let held = accounts.get(did);
+        cids.iter()
+            .filter(|cid| !held.is_some_and(|account| account.blobs.contains_key(cid.as_str())))
+            .cloned()
+            .collect()
+    }
+
     /// Marks each of `cids`, where held by `did`, as referenced once more.
     /// See [`BlobStore::mark_referenced`].
     pub(crate) fn mark_referenced(&self, did: &str, cids: &[String]) {
@@ -999,6 +1029,48 @@ impl BlobIndex {
         candidates
     }
 
+    /// Collects one candidate, if it is still collectable, running `take`
+    /// under this index's own lock.
+    ///
+    /// **The lock is the exclusion.** `collect_candidates` reads, and an
+    /// upload of the same CID can land between that read and this call:
+    /// [`BlobIndex::insert_checked`] holds this lock across its own
+    /// `persist`, so a collector that removed the entry and unlinked the
+    /// bytes outside it could be told "already held" by an upload whose
+    /// bytes it was in the middle of deleting, or delete the bytes that
+    /// upload had just written. Doing both here, under one hold, makes the
+    /// two orders impossible.
+    ///
+    /// Re-checked as well as re-locked: a re-upload refreshes the entry's
+    /// `uploaded_at` ([`BlobIndex::insert_checked`]), so a candidate chosen
+    /// a moment ago may no longer be past `cutoff`, and a record written
+    /// since may have referenced it. Either way this leaves it alone and
+    /// answers `None`, and `take` never runs.
+    ///
+    /// `take` is where a durable store logs the collection and unlinks the
+    /// file. Returning `Err` from it leaves the entry in place for the next
+    /// sweep.
+    pub(crate) fn collect_one<E>(
+        &self,
+        did: &str,
+        cid: &str,
+        cutoff: OffsetDateTime,
+        take: impl FnOnce() -> Result<(), E>,
+    ) -> Option<Result<BlobRef, E>> {
+        let mut accounts = self.lock();
+        let account = accounts.get_mut(did)?;
+        let held = account.blobs.get(cid)?;
+        if held.refs != 0 || held.uploaded_at > cutoff {
+            return None;
+        }
+        if let Err(error) = take() {
+            return Some(Err(error));
+        }
+        let held = account.blobs.remove(cid)?;
+        account.bytes = account.bytes.saturating_sub(held.reference.size);
+        Some(Ok(held.reference))
+    }
+
     /// Removes one blob from the index, answering what it was.
     ///
     /// The second half of a collection, called once the caller has already
@@ -1253,6 +1325,10 @@ impl BlobStore for MemoryBlobStore {
         self.limits
     }
 
+    fn missing_blobs(&self, did: &str, cids: &[String]) -> Vec<String> {
+        self.inner.index.missing_blobs(did, cids)
+    }
+
     fn mark_referenced(&self, did: &str, cids: &[String]) {
         self.inner.index.mark_referenced(did, cids);
     }
@@ -1266,8 +1342,13 @@ impl BlobStore for MemoryBlobStore {
         let candidates = self.inner.index.collect_candidates(cutoff);
         let mut collected = Vec::with_capacity(candidates.len());
         for (did, cid, bytes) in candidates {
-            if self.inner.index.remove_one(&did, &cid).is_some() {
+            // Both halves under the index's own lock, so an upload of the
+            // same CID cannot land between them; see `BlobIndex::collect_one`.
+            let taken = self.inner.index.collect_one(&did, &cid, cutoff, || {
                 self.inner.bytes().remove(&(did.clone(), cid.clone()));
+                Ok::<(), std::convert::Infallible>(())
+            });
+            if matches!(taken, Some(Ok(_))) {
                 collected.push(CollectedBlob { did, cid, bytes });
             }
         }
@@ -1543,6 +1624,57 @@ mod tests {
         upload.commit().expect("the upload should commit")
     }
 
+    /// **The window a re-upload used to fall into.** `collect_candidates`
+    /// reads, and an upload of the same CID can land before the collector
+    /// gets to that candidate — which is exactly what a client re-uploading
+    /// "to be safe" does. `collect_one` re-checks under the lock the upload
+    /// took, so the refreshed entry is left alone and its bytes stay.
+    #[test]
+    fn a_candidate_re_uploaded_since_the_sweep_began_is_not_collected() {
+        let index = BlobIndex::default();
+        let reference = BlobRef {
+            cid: "bafkreitestcid".to_owned(),
+            mime_type: "image/png".to_owned(),
+            size: 4,
+        };
+        let long_ago = OffsetDateTime::now_utc() - time::Duration::hours(12);
+        index
+            .insert_checked(
+                "did:web:agent.example",
+                reference.clone(),
+                BlobLimits::default(),
+                long_ago,
+                0,
+                |_| Ok(()),
+            )
+            .expect("the first upload lands");
+
+        let cutoff = OffsetDateTime::now_utc() - time::Duration::hours(6);
+        let candidates = index.collect_candidates(cutoff);
+        assert_eq!(candidates.len(), 1, "past its grace: {candidates:?}");
+
+        // The re-upload, between the read above and the take below.
+        index
+            .insert_checked(
+                "did:web:agent.example",
+                reference.clone(),
+                BlobLimits::default(),
+                OffsetDateTime::now_utc(),
+                0,
+                |_| panic!("a held blob is not persisted twice"),
+            )
+            .expect("the re-upload is answered from the index");
+
+        let mut unlinked = false;
+        let taken = index.collect_one("did:web:agent.example", &reference.cid, cutoff, || {
+            unlinked = true;
+            Ok::<(), std::convert::Infallible>(())
+        });
+        assert!(taken.is_none(), "the re-uploaded blob was collected anyway");
+        assert!(!unlinked, "its bytes were removed anyway");
+        assert!(index.get("did:web:agent.example", &reference.cid).is_some());
+    }
+
     #[test]
     fn a_blob_is_named_by_its_bytes_however_they_arrive() {
         let store = MemoryBlobStore::new();
diff --git a/crates/didbot-pds/src/durable.rs b/crates/didbot-pds/src/durable.rs
index e8e6ae5b..8cc3a171 100644
--- a/crates/didbot-pds/src/durable.rs
+++ b/crates/didbot-pds/src/durable.rs
@@ -1956,6 +1956,10 @@ impl BlobStore for FileBlobStore {
         self.limits
     }
 
+    fn missing_blobs(&self, did: &str, cids: &[String]) -> Vec<String> {
+        self.inner.index.missing_blobs(did, cids)
+    }
+
     fn mark_referenced(&self, did: &str, cids: &[String]) {
         self.inner.index.mark_referenced(did, cids);
     }
@@ -1978,35 +1982,45 @@ impl BlobStore for FileBlobStore {
         let candidates = self.inner.index.collect_candidates(cutoff);
         let mut collected = Vec::with_capacity(candidates.len());
         for (did, cid, bytes) in candidates {
-            if let Err(error) = self.inner.wal.append(
-                &Entry::BlobCollected {
-                    did: did.clone(),
-                    cid: cid.clone(),
-                },
-                Durability::Sync,
-            ) {
-                tracing::error!(
+            // Logged, unlinked and dropped from the index under one hold of
+            // the index's own lock, which is what `insert_checked` holds
+            // across its own `persist`: without that, an upload of the same
+            // CID could be told "already held" while these bytes were being
+            // removed, or write bytes this loop then deleted. The candidate
+            // is re-checked under that lock too, so a re-upload that
+            // restarted the grace is left alone. See
+            // `BlobIndex::collect_one`.
+            let taken = self.inner.index.collect_one(&did, &cid, cutoff, || {
+                self.inner.wal.append(
+                    &Entry::BlobCollected {
+                        did: did.clone(),
+                        cid: cid.clone(),
+                    },
+                    Durability::Sync,
+                )?;
+                let path = self.inner.account_dir(&did).join(&cid);
+                if let Err(error) = std::fs::remove_file(&path) {
+                    if error.kind() != std::io::ErrorKind::NotFound {
+                        tracing::warn!(
+                            did,
+                            cid,
+                            path = %path.display(),
+                            %error,
+                            "could not remove a collected blob's bytes; the next sweep will find it"
+                        );
+                    }
+                }
+                Ok(())
+            });
+            match taken {
+                Some(Ok(_)) => collected.push(CollectedBlob { did, cid, bytes }),
+                Some(Err(error)) => tracing::error!(
                     did,
                     cid,
                     error = %backend(error),
                     "could not log a blob collection; leaving the blob in place"
-                );
-                continue;
-            }
-            let path = self.inner.account_dir(&did).join(&cid);
-            if let Err(error) = std::fs::remove_file(&path) {
-                if error.kind() != std::io::ErrorKind::NotFound {
-                    tracing::warn!(
-                        did,
-                        cid,
-                        path = %path.display(),
-                        %error,
-                        "could not remove a collected blob's bytes; the next sweep will find it"
-                    );
-                }
-            }
-            if self.inner.index.remove_one(&did, &cid).is_some() {
-                collected.push(CollectedBlob { did, cid, bytes });
+                ),
+                None => {}
             }
         }
         if !collected.is_empty() {
diff --git a/crates/didbot-pds/src/object_blobs.rs b/crates/didbot-pds/src/object_blobs.rs
index 60d0efac..fc2538b5 100644
--- a/crates/didbot-pds/src/object_blobs.rs
+++ b/crates/didbot-pds/src/object_blobs.rs
@@ -364,6 +364,10 @@ impl BlobStore for ObjectBlobStore {
         self.limits
     }
 
+    fn missing_blobs(&self, did: &str, cids: &[String]) -> Vec<String> {
+        self.inner.index.missing_blobs(did, cids)
+    }
+
     fn mark_referenced(&self, did: &str, cids: &[String]) {
         self.inner.index.mark_referenced(did, cids);
     }
@@ -393,32 +397,38 @@ impl BlobStore for ObjectBlobStore {
         let candidates = self.inner.index.collect_candidates(cutoff);
         let mut collected = Vec::with_capacity(candidates.len());
         for (did, cid, bytes) in candidates {
-            if let Err(error) = self.inner.wal.append(
-                &Entry::BlobCollected {
-                    did: did.clone(),
-                    cid: cid.clone(),
-                },
-                Durability::Sync,
-            ) {
-                tracing::error!(
+            // Under one hold of the index's own lock, and re-checked there,
+            // for the reason `BlobIndex::collect_one` gives: an upload of
+            // the same CID holds that lock across its own persist, so
+            // anything done outside it can race a client being told `200`.
+            let taken = self.inner.index.collect_one(&did, &cid, cutoff, || {
+                self.inner.wal.append(
+                    &Entry::BlobCollected {
+                        did: did.clone(),
+                        cid: cid.clone(),
+                    },
+                    Durability::Sync,
+                )?;
+                let key = ObjectState::key(&did, &cid);
+                if let Err(error) = self.inner.backend.delete(&key) {
+                    tracing::warn!(
+                        did,
+                        cid,
+                        %error,
+                        "could not delete a collected blob from the object store; it is logged as gone and now leaks there"
+                    );
+                }
+                Ok(())
+            });
+            match taken {
+                Some(Ok(_)) => collected.push(CollectedBlob { did, cid, bytes }),
+                Some(Err(error)) => tracing::error!(
                     did,
                     cid,
                     error = %wal_message(error),
                     "could not log an object blob collection; leaving the blob in place"
-                );
-                continue;
-            }
-            let key = ObjectState::key(&did, &cid);
-            if let Err(error) = self.inner.backend.delete(&key) {
-                tracing::warn!(
-                    did,
-                    cid,
-                    %error,
-                    "could not delete a collected blob from the object store; it is logged as gone and now leaks there"
-                );
-            }
-            if self.inner.index.remove_one(&did, &cid).is_some() {
-                collected.push(CollectedBlob { did, cid, bytes });
+                ),
+                None => {}
             }
         }
         if !collected.is_empty() {
diff --git a/crates/didbot-pds/src/provision.rs b/crates/didbot-pds/src/provision.rs
index 75f0e413..c1f74696 100644
--- a/crates/didbot-pds/src/provision.rs
+++ b/crates/didbot-pds/src/provision.rs
@@ -4337,6 +4337,12 @@ where
                 .map(crate::records::blob_refs)
                 .unwrap_or_default();
             let new_refs = crate::records::blob_refs(&record);
+            // Before the write lands, because `mark_referenced` runs after it
+            // and cannot refuse — see `BlobStore::missing_blobs`. A record
+            // whose image this account does not hold is a permanent `404`
+            // nothing later repairs, so it is refused while the client still
+            // has the bytes.
+            self.require_blobs_held(account.did.as_str(), &new_refs)?;
             // Named under the same lock and for the same reason: after the
             // write nothing downstream of the store can tell a create from
             // an update, and the firehose has to say which it was — and, for
@@ -4463,6 +4469,28 @@ where
     /// different facts: freezing an account must not depend on revoking
     /// every session it holds, some of which — see `crate::session` — are
     /// not durable across a restart.
+    /// Refuses a write naming a blob `did` does not hold.
+    ///
+    /// The reference implementation's "Could not find blob", and for its
+    /// reason: `BlobStore::mark_referenced` runs after the record is in the
+    /// repository and skips a CID it does not hold, so without this a client
+    /// that uploaded a blob, composed past the grace period, and then wrote
+    /// the record is told its write succeeded and serves a `404` for the
+    /// image forever.
+    fn require_blobs_held(&self, did: &str, cids: &[String]) -> Result<(), ProvisionError> {
+        if cids.is_empty() {
+            return Ok(());
+        }
+        let missing = self.blobs.missing_blobs(did, cids);
+        let Some(cid) = missing.into_iter().next() else {
+            return Ok(());
+        };
+        tracing::info!(did, cid, "write refused: the account holds no such blob");
+        Err(ProvisionError::Record(
+            crate::records::RecordError::MissingBlob { cid },
+        ))
+    }
+
     fn require_writable(&self, account: &AgentAccount) -> Result<(), ProvisionError> {
         if account.policy().accepts_external_writes {
             return Ok(());
@@ -6734,6 +6762,20 @@ where
                 .iter()
                 .map(|op| self.existing_record_for_op(did, op))
                 .collect();
+            // Every blob the batch names, checked before any of it applies —
+            // all-or-nothing, the same rule the record store applies. See
+            // `require_blobs_held`.
+            let named: Vec<String> = writes
+                .iter()
+                .filter_map(|op| match op {
+                    BatchOp::Create { record, .. } | BatchOp::Update { record, .. } => {
+                        Some(crate::records::blob_refs(record))
+                    }
+                    BatchOp::Delete { .. } => None,
+                })
+                .flatten()
+                .collect();
+            self.require_blobs_held(did, &named)?;
             // Named while the old content is still readable. A firehose op's
             // `prev` is what lets a consumer run an update or a delete backwards,
             // and after `apply_batch` there is nothing left to name.
diff --git a/crates/didbot-pds/src/records.rs b/crates/didbot-pds/src/records.rs
index 5c635c7e..16b59d08 100644
--- a/crates/didbot-pds/src/records.rs
+++ b/crates/didbot-pds/src/records.rs
@@ -178,6 +178,19 @@ pub enum RecordError {
     /// The record's `$type` is present but is not a string.
     #[error("record `$type` must be a string")]
     TypeNotAString,
+    /// The record names a blob this account does not hold.
+    ///
+    /// A blob has a grace period before it is collected, and a client that
+    /// uploads one and composes for longer than that reaches the write with
+    /// a reference to bytes that are gone. Storing the record anyway means
+    /// answering `200` and serving a permanent `404` for the image, which
+    /// nothing later can repair: the write is refused instead, so the client
+    /// still holding the bytes can upload them again.
+    #[error("record names blob `{cid}`, which this account does not hold")]
+    MissingBlob {
+        /// The CID the record named.
+        cid: String,
+    },
     /// The record does not satisfy the lexicon that defines its collection.
     ///
     /// The message names the field and what was wrong with it, because the
diff --git a/crates/didbot-pds/tests/blob_references.rs b/crates/didbot-pds/tests/blob_references.rs
index c615f514..93567c8e 100644
--- a/crates/didbot-pds/tests/blob_references.rs
+++ b/crates/didbot-pds/tests/blob_references.rs
@@ -248,18 +248,19 @@ fn a_blob_is_collectable_when_its_last_record_goes_and_not_when_its_first_does()
     let _ = std::fs::remove_dir_all(&dir);
 }
 
-/// A record written before the blob it names holds nothing on it, and a
-/// restart says the same.
+/// A record naming bytes the account does not hold is refused, and the
+/// upload that follows is what makes the same write land — with the count a
+/// restart agrees on.
 ///
 /// A reference is resolved against the blobs the account holds at the moment
-/// the record is written, so a record naming bytes that arrive afterwards
-/// names something the account did not have: the live deployment counts zero,
-/// and the upload that follows starts at zero like any other. A restart that
-/// counted the finished state instead — every record against every blob —
-/// would answer one here, and the blob would sit in the account's quota for
-/// as long as the record did, on a rule no running process ever applied.
+/// the record is written, so a record naming bytes that have not arrived is a
+/// record whose image is a permanent `404`: it is refused rather than stored
+/// at a count of zero. Once the bytes are there the write lands and counts
+/// one, and a restart that counted the finished state some other way — every
+/// record against every blob, or the log's own order ignored — would answer
+/// something else here.
 #[test]
-fn a_record_written_before_the_blob_it_names_holds_no_reference_on_it() {
+fn a_record_naming_bytes_the_account_does_not_hold_is_refused_until_they_arrive() {
     let dir = scratch("out-of-order");
 
     let (durable, pds) = boot(&dir);
@@ -270,44 +271,46 @@ fn a_record_written_before_the_blob_it_names_holds_no_reference_on_it() {
     pds.delete(elsewhere.as_str()).expect("the delete lands");
 
     let did = account(&pds, "kestrel");
-    pds.put_record(
-        &did,
-        COLLECTION,
-        None,
-        json!({"image": reference.to_json()}),
-        &Swap::default(),
-    )
-    .expect("a record may name bytes this account does not hold");
+    let record = json!({"image": reference.to_json()});
+    let err = pds
+        .put_record(&did, COLLECTION, None, record.clone(), &Swap::default())
+        .expect_err("a record may not name bytes this account does not hold");
+    assert!(
+        err.to_string().contains(&reference.cid),
+        "the refusal names the blob the client has to upload: {err}"
+    );
+
     let landed = upload(&pds, &did, b"shearwater");
     assert_eq!(landed.cid, reference.cid, "content addressing, by hand");
+    pds.put_record(&did, COLLECTION, None, record, &Swap::default())
+        .expect("the write lands once the bytes are here");
     churn(&pds);
 
-    let unreferenced = |durable: &Durable, at: &str| {
+    let referenced = |durable: &Durable, at: &str| {
         let stats = durable.blobs().stats();
         assert_eq!(stats.held.count, 2, "{at}: an avatar and this upload");
         assert_eq!(
-            stats.referenced.count, 1,
-            "{at}: the profile's avatar, and nothing else: {stats:?}"
+            stats.referenced.count, 2,
+            "{at}: the profile's avatar and the record's own image: {stats:?}"
         );
         assert_eq!(
-            stats.collectable.count, 1,
-            "{at}: the upload the record did not reach: {stats:?}"
+            stats.collectable.count, 0,
+            "{at}: a record names every blob here: {stats:?}"
         );
     };
-    unreferenced(&durable, "before any restart");
+    referenced(&durable, "before any restart");
     drop(pds);
     drop(durable);
 
     let (durable, pds) = boot(&dir);
-    unreferenced(&durable, "over the log as it was written");
+    referenced(&durable, "over the log as it was written");
     drop(pds);
     drop(durable);
 
     let (durable, pds) = boot(&dir);
-    unreferenced(&durable, "over the checkpoint");
+    referenced(&durable, "over the checkpoint");
     let collected = pds.collect_blobs();
-    assert_eq!(collected.len(), 1, "{collected:?}");
-    assert_eq!(collected[0].cid, reference.cid);
+    assert!(collected.is_empty(), "{collected:?}");
     drop(pds);
     drop(durable);
     let _ = std::fs::remove_dir_all(&dir);
diff --git a/crates/didbot-pds/tests/object_blobs.rs b/crates/didbot-pds/tests/object_blobs.rs
index b2358714..1ebf2d22 100644
--- a/crates/didbot-pds/tests/object_blobs.rs
+++ b/crates/didbot-pds/tests/object_blobs.rs
@@ -140,13 +140,11 @@ fn an_unreferenced_upload_is_collected_and_its_bytes_leave_the_backend() {
     );
 }
 
+/// A record naming a blob nobody uploaded is refused, over the object store
+/// the same way it is over every other one: the check reads the index, which
+/// this backend shares.
 #[test]
-fn a_record_referencing_a_blob_never_uploaded_is_written_anyway() {
-    // `records::blob_refs` cannot tell a blob reference from a coincidence,
-    // and this crate's stated position (see `BlobStore::mark_referenced`'s
-    // docs) is that a record naming a CID nothing holds is not this call's
-    // problem to raise. The record lands; only a later `getBlob` for that
-    // CID fails, the same as it would for any CID nobody ever uploaded.
+fn a_record_referencing_a_blob_never_uploaded_is_refused() {
     let pds = provisioner("dangling-reference");
     let did = account(&pds, "ptarmigan");
     let dangling = BlobRef {
@@ -155,14 +153,16 @@ fn a_record_referencing_a_blob_never_uploaded_is_written_anyway() {
         size: 4,
     };
 
-    pds.put_record(
-        &did,
-        COLLECTION,
-        None,
-        json!({"image": dangling.to_json()}),
-        &Swap::default(),
-    )
-    .expect("a write naming an unknown blob is not this layer's problem to refuse");
+    let err = pds
+        .put_record(
+            &did,
+            COLLECTION,
+            None,
+            json!({"image": dangling.to_json()}),
+            &Swap::default(),
+        )
+        .expect_err("a write naming a blob this account does not hold is refused");
+    assert!(err.to_string().contains(&dangling.cid), "{err}");
 
     assert!(
         pds.fetch_blob(&did, &dangling.cid).is_err(),
diff --git a/crates/didbot-serve/tests/blob_collect.rs b/crates/didbot-serve/tests/blob_collect.rs
index 071daa70..b55ffd39 100644
--- a/crates/didbot-serve/tests/blob_collect.rs
+++ b/crates/didbot-serve/tests/blob_collect.rs
@@ -97,3 +97,84 @@ async fn the_scheduled_collector_unlinks_an_unreferenced_blob_and_spares_a_refer
         "a blob a record references must survive the collector"
     );
 }
+
+/// **The blob that vanishes after a 200.** A client uploads a blob, composes
+/// past the collection grace, and then writes the record naming it. The blob
+/// is gone by then, and before this the write was accepted anyway: the client
+/// was told it succeeded and the image `404`s forever, with nothing left that
+/// could repair it. The write is refused instead, while the client still has
+/// the bytes.
+#[tokio::test]
+async fn a_write_naming_a_collected_blob_is_refused_rather_than_stored() {
+    let pds = provisioner();
+    let did = pds
+        .provision(ProvisionRequest::new("junco", None))
+        .expect("provisioning should succeed")
+        .account
+        .did
+        .as_str()
+        .to_owned();
+    let cid = upload(pds.as_ref(), &did, b"an avatar's worth of bytes");
+
+    // The compose, longer than the grace. The collector takes it.
+    tokio::time::sleep(TEST_GRACE * 3).await;
+    assert!(!pds.collect_blobs().is_empty(), "the blob was collected");
+    assert!(pds.fetch_blob(&did, &cid).is_err());
+
+    let record = json!({"image": {"$type": "blob", "ref": {"$link": cid},
+                                  "mimeType": "image/png", "size": 26}});
+    let err = pds
+        .put_record(
+            &did,
+            "com.example.thing",
+            None,
+            record.clone(),
+            &Swap::default(),
+        )
+        .expect_err("a write naming a blob this account does not hold is refused");
+    assert!(
+        err.to_string().contains(&cid),
+        "the refusal names the blob the client has to re-upload: {err}"
+    );
+    assert!(
+        pds.list_records(&did, "com.example.thing", &didbot_pds::ListParams::new(10))
+            .expect("the collection reads")
+            .is_empty(),
+        "a refused write must leave nothing behind"
+    );
+
+    // Re-uploading the bytes is what makes the write work, which is the
+    // whole point of refusing rather than storing.
+    let again = upload(pds.as_ref(), &did, b"an avatar's worth of bytes");
+    assert_eq!(again, cid);
+    pds.put_record(&did, "com.example.thing", None, record, &Swap::default())
+        .expect("the write lands once the blob is back");
+}
+
+/// **The re-upload that changes nothing.** A client that re-uploads a blob it
+/// already holds is told `200`, and before this the blob's age still counted
+/// from the first upload — so the next sweep collected it anyway. The clock
+/// starts again, which is what a `200` on an upload has to mean.
+#[tokio::test]
+async fn re_uploading_a_held_blob_restarts_its_grace() {
+    let pds = provisioner();
+    let did = pds
+        .provision(ProvisionRequest::new("junco", None))
+        .expect("provisioning should succeed")
+        .account
+        .did
+        .as_str()
+        .to_owned();
+    let cid = upload(pds.as_ref(), &did, b"an avatar's worth of bytes");
+    tokio::time::sleep(TEST_GRACE * 3).await;
+
+    // "To be safe." The store answers that it already holds it.
+    let again = upload(pds.as_ref(), &did, b"an avatar's worth of bytes");
+    assert_eq!(again, cid);
+
+    assert!(
+        pds.collect_blobs().is_empty(),
+        "a blob re-uploaded a moment ago is not past its grace"
+    );
+    assert!(pds.fetch_blob(&did, &cid).is_ok());
+}