volume: a delete cancels writes of its needle still in flight (#12445)

* volume: a delete cancels the writes of its needle still in flight

A write queued behind the volume locks, for example behind another write's
fsync on a busy disk, is not in the needle map yet. A delete of that needle
in the meantime finds nothing, and the write lands afterwards, so the needle
comes back on this copy only while the other replicas have it deleted.

Each volume now tracks its writes from entry until they return. A delete
cancels the ones for its needle, and a cancelled write that landed is
tombstoned before it returns, unless the needle was deleted or rewritten
since.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* volume: a replicate delete that finds nothing cancels the pending write

Rolling back a failed chunk upload sends DELETE ?type=replicate to every
holder, including the slow one whose write timed out on the client. That
holder has not indexed the write yet, so the delete answered 404 and the
write landed live afterwards. Cancel the needle's pending writes, then look
again in case one finished in between. A plain delete that finds nothing
still answers 404 without cancelling: it is not sent on to the replicas.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* seaweed-volume: a delete cancels the writes of its needle still in flight

Mirrors the Go volume. Volume::track_pending_write registers a write until
its guard drops, delete_needle cancels the needle's pending writes, and
settle_pending_write tombstones a cancelled write that landed unless the
needle was deleted or rewritten since.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* seaweed-volume: track uploads and cancel them on a replicate delete

An upload is tracked from the handler, and a durable one from submission
to the write queue, where the guard travels with the request: a handler
dropped while the write waits in the queue does not untrack it, and the
worker settles it after the batch. A replicate delete that finds nothing
cancels the needle's pending writes and looks again, like the Go handler.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* volume: take a cancelled write back in the lock span that publishes it

The write tombstoned itself only after it returned, with the volume locks
released. In that gap a compaction could move the record so the offset
check skipped it, a close could drop it, and an identical write sent after
the delete could match the live record and report success before the
tombstone removed it. The check now runs right after the publish, under
the same dataFileSyncLock and dataFileAccessLock hold, in the plain, fsync
and batch write paths.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* volume: a delete cancels only the writes carrying its cookie

A delete with the right needle id but the wrong cookie could cancel an
in-flight write and get it tombstoned. Each pending write now records its
cookie, and a cancel only marks the matching ones.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* seaweed-volume: take a cancelled write back in the lock span that publishes it

Mirror the Go change: settle_pending_write ran after write_needles_grouped
returned, so a compaction, a close or an identical write later in the same
batch could slip in before the tombstone. drop_cancelled_write now runs
right after the publish, under the same sync and access lock hold, in the
synced run, the unsynced run and write_needle_unsynced_or_refused.
write_pending_needles carries each write's PendingWrite down to that
point.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* seaweed-volume: a delete cancels only the writes carrying its cookie

Mirror the Go change: a delete with the right needle id and the wrong
cookie no longer cancels an in-flight write. Each tracked write keeps its
cookie next to its cancel flag.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* seaweed-volume: refuse a queued write whose volume was remounted

The write queue tracks a write on the volume found at submit, but the
worker looks the volume up again by id. After a remount, a delete reaches
the new volume and cannot cancel the write registered on the old one, so
the worker now refuses such a write instead of landing it on the new
volume.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* seaweed-volume: decode the delete's type parameter like Go's FormValue

The delete handler matched the raw query text for type=replicate, so a
percent-encoded name or value was not treated as a replica delete and did
not cancel the write it undoes. Parse the query the way the upload handler
does.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* volume: test a replica delete landing in a write's fsync window

Drive DeleteHandler while a durable write is parked in its .dat fsync,
appended but not indexed: the replica delete answers 404 and the write is
taken back when it publishes, while a plain delete leaves it live.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* volume: flush a take-back tombstone before the index row that points at it

A durable write taken back during its fsync appends the tombstone after the
.dat sync, and the .idx sync that follows made its row durable while the
record itself could still be lost, which reloads the volume read only. The
fsync and batch paths now sync the .dat again first when a take-back
appended a tombstone; a failure stops the volume taking writes like a failed
index sync.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* seaweed-volume: flush a take-back tombstone before the index row that points at it

A durable write taken back in write_synced_run appends its tombstone after
the .dat sync, so the .idx flush that follows could make the row durable
while the tombstone record itself was still in the page cache. The run now
syncs the .dat again first when a take-back appended a tombstone; a failure
stops the volume taking writes like a failed index flush.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* volume: a skip-cookie batch delete cancels the writes of the file id it names

BatchDelete with SkipCookieCheck left the needle's cookie at zero, so the
delete's cancel matched none of the in-flight writes and a retry queued
behind it could land afterwards. The non-EC delete now carries the file
id's cookie, as the Rust volume server already does; the EC path keeps
cookie 0, which DeleteEcShardNeedle reads as "skip the check".

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* volume: release a held test write before the store closes

A failed wait ended the test before release ran, so the writer stayed parked
in its fsync and the store cleanup blocked on its lock until the package
timed out.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

seaweed-volume/src/server/handlers.rs

@@ -3419,10 +3419,14 @@ async fn write_upload_needle(
         Ok(vol) => vol,
         Err(refused) => return Ok(Err(refused)),
     };
+    let pending = vol.track_pending_write(&n);
     if fsync {
-        vol.write_needle(&mut n, true, fsync).map(Ok)
+        vol.write_pending_needles(&mut [(n, true)], &[Some(pending)])
+            .pop()
+            .unwrap_or(Err(crate::storage::volume::VolumeError::NotFound))
+            .map(Ok)
     } else {
-        vol.write_needle_unsynced_or_refused(&mut n)
+        vol.write_pending_needle_unsynced(&mut n, Some(&pending))
     }
 }
 
@@ -3615,15 +3619,29 @@ pub async fn delete_handler(
 
     // Read needle first to validate cookie (matching Go behavior)
     let original_cookie = cookie;
+    let is_replicate = serde_urlencoded::from_str::<Vec<(String, String)>>(&del_query)
+        .unwrap_or_default()
+        .into_iter()
+        .find(|(name, _)| name == "type")
+        .is_some_and(|(_, value)| value == "replicate");
     {
         let vol = {
             let store = state.store.read().unwrap();
             store.find_volume(vid).map(|(_, vol)| vol)
         };
-        match vol
+        let mut found = vol
+            .as_ref()
             .ok_or(crate::storage::volume::VolumeError::NotFound)
-            .and_then(|vol| vol.read_needle_with_option(&mut n, &mut delete_read_option()))
+            .and_then(|vol| vol.read_needle_with_option(&mut n, &mut delete_read_option()));
+        if found.is_err()
+            && is_replicate
+            && let Some(vol) = &vol
         {
+            // the write this delete undoes can still be queued here
+            vol.cancel_pending_writes(needle_id, original_cookie);
+            found = vol.read_needle_with_option(&mut n, &mut delete_read_option());
+        }
+        match found {
             Ok(_) => {}
             Err(_) => {
                 let result = DeleteResult { size: 0 };
@@ -3711,7 +3729,6 @@ pub async fn delete_handler(
         }
     };
 
-    let is_replicate = del_query.split('&').any(|p| p == "type=replicate");
     if !is_replicate && delete_result.is_ok() {
         let needs_replication = {
             let store = state.store.read().unwrap();
@@ -9050,6 +9067,98 @@ mod tests {
         }
     }
 
+    /// A write queued behind another write's fsync is not indexed yet, so a
+    /// replicate delete of it finds nothing. The delete cancels the write, so
+    /// it is taken back when it lands instead of reviving the needle on this
+    /// copy alone, even when its handler was dropped while it waited in the
+    /// durable queue. A plain delete that finds nothing is not sent on to the
+    /// replicas, so it leaves the write alone.
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    #[allow(clippy::await_holding_lock)]
+    async fn test_replicate_delete_cancels_a_queued_write() {
+        use crate::server::write_queue::{SyncBatchOption, WriteQueue};
+        use crate::storage::volume::VolumeError;
+        use std::time::{Duration, Instant};
+
+        for (id, fsync, query, abandon) in [
+            (0x6e7a_0c01u64, false, "?type=replicate", false),
+            (0x6e7a_0c02, true, "?type=replicate", false),
+            (0x6e7a_0c03, true, "?type=replicate", true),
+            (0x6e7a_0c04, false, "", false),
+            (0x6e7a_0c05, false, "?%74ype=replicat%65", false),
+        ] {
+            let replicate = !query.is_empty();
+            let tmp = tempfile::TempDir::new().unwrap();
+            let state = volume_test_state(&tmp);
+            assert!(
+                state
+                    .write_queue
+                    .set(WriteQueue::new(
+                        state.clone(),
+                        128,
+                        SyncBatchOption::default()
+                    ))
+                    .is_ok()
+            );
+            let vol = state
+                .store
+                .read()
+                .unwrap()
+                .find_volume(VolumeId(1))
+                .unwrap()
+                .1;
+            let held = vol.sync_lock_for_test();
+
+            let mut n = Needle {
+                id: NeedleId(id),
+                cookie: Cookie(TEST_COOKIE),
+                data: b"late write".to_vec(),
+                data_size: 10,
+                ..Needle::default()
+            };
+            n.checksum = CRC::new(&n.data);
+            let write = tokio::spawn({
+                let state = state.clone();
+                async move { write_upload_needle(&state, VolumeId(1), n, fsync).await }
+            });
+            let deadline = Instant::now() + Duration::from_secs(10);
+            while vol.pending_writes_for_test(NeedleId(id)) == 0 {
+                assert!(Instant::now() < deadline, "the write never registered");
+                tokio::time::sleep(Duration::from_millis(1)).await;
+            }
+
+            let path = format!("/1,{:x}{:08x}{}", id, TEST_COOKIE, query);
+            assert_eq!(send_delete(&state, &path).await, StatusCode::NOT_FOUND);
+            if abandon {
+                write.abort();
+                drop(held);
+                while vol.pending_writes_for_test(NeedleId(id)) > 0 {
+                    assert!(Instant::now() < deadline, "the queued write never settled");
+                    tokio::time::sleep(Duration::from_millis(1)).await;
+                }
+            } else {
+                drop(held);
+                let written = write.await.unwrap();
+                if replicate {
+                    assert!(matches!(written, Err(VolumeError::Deleted)), "{id:x}");
+                } else {
+                    assert!(matches!(written, Ok(Ok(_))), "{id:x}");
+                }
+                assert_eq!(vol.pending_writes_for_test(NeedleId(id)), 0);
+            }
+
+            let read = vol.read_needle(&mut Needle {
+                id: NeedleId(id),
+                ..Needle::default()
+            });
+            if replicate {
+                assert!(read.is_err(), "{id:x}: the cancelled write is live");
+            } else {
+                assert!(read.is_ok(), "{id:x}: {read:?}");
+            }
+        }
+    }
+
     /// Issue #801: an fsync upload takes the per-volume worker so its sync is
     /// shared across the batch (Go's `writeNeedle2` split); a non-fsync write,
     /// or an fsync write landing while the server is stopping, takes the

seaweed-volume/src/server/write_queue.rs

@@ -18,7 +18,7 @@ use tracing::debug;
 
 use crate::storage::needle::needle::Needle;
 use crate::storage::types::{Size, VolumeId};
-use crate::storage::volume::VolumeError;
+use crate::storage::volume::{PendingWrite, VolumeError};
 
 use super::volume_server::VolumeServerState;
 
@@ -26,7 +26,12 @@ use super::volume_server::VolumeServerState;
 pub type WriteResult = Result<(u64, Size, bool), VolumeError>;
 
 /// The needles queued for one volume, each with the durability it asked for.
-type VolumeBatch = Vec<(Needle, bool, oneshot::Sender<WriteResult>)>;
+type VolumeBatch = Vec<(
+    Needle,
+    bool,
+    oneshot::Sender<WriteResult>,
+    Option<PendingWrite>,
+)>;
 
 /// A request to write a needle, submitted to the write queue.
 pub struct WriteRequest {
@@ -38,6 +43,8 @@ pub struct WriteRequest {
     /// that cancels `submit` after the send must not free the slot while the
     /// worker still holds the needle.
     pub queue_permit: Option<OwnedSemaphorePermit>,
+    /// Tracked from submission, so a delete cancels the write while it waits.
+    pub pending: Option<PendingWrite>,
 }
 
 /// Maximum number of write requests to batch together.
@@ -167,12 +174,13 @@ impl WriteQueue {
         #[cfg(test)]
         self.submits
             .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
-        {
+        let pending = {
             let store = self.state.store.read().unwrap();
-            if store.find_volume(volume_id).is_none() {
-                return Err(VolumeError::NotFound);
+            match store.find_volume(volume_id) {
+                Some((_, vol)) => vol.track_pending_write(&needle),
+                None => return Err(VolumeError::NotFound),
             }
-        }
+        };
         // The permit travels inside the request, so the global queued bound
         // covers the request until the worker drops it — including after a
         // canceled submit.
@@ -189,6 +197,7 @@ impl WriteQueue {
             fsync,
             response_tx,
             queue_permit: Some(queue_permit),
+            pending: Some(pending),
         };
 
         // A send fails only against a retiring worker's closed channel; the
@@ -345,10 +354,11 @@ fn process_batch(state: Arc<VolumeServerState>, batch: Vec<WriteRequest>) {
 
     for req in batch {
         let vid = req.volume_id;
+        let entry = (req.needle, req.fsync, req.response_tx, req.pending);
         if let Some(group) = groups.iter_mut().find(|(v, _)| *v == vid) {
-            group.1.push((req.needle, req.fsync, req.response_tx));
+            group.1.push(entry);
         } else {
-            groups.push((vid, vec![(req.needle, req.fsync, req.response_tx)]));
+            groups.push((vid, vec![entry]));
         }
     }
 
@@ -356,16 +366,28 @@ fn process_batch(state: Arc<VolumeServerState>, batch: Vec<WriteRequest>) {
     // the group's I/O with the guard released — one volume's write must not
     // block admission on another volume's.
     for (vid, entries) in groups {
-        let (mut writes, senders): (Vec<_>, Vec<_>) = entries
-            .into_iter()
-            .map(|(needle, fsync, response_tx)| ((needle, fsync), response_tx))
-            .unzip();
         let vol = {
             let store = state.store.read().unwrap();
             store.writable_volume(vid)
         };
-        let results = match vol {
-            Ok(vol) => vol.write_needles_grouped(&mut writes),
+        let mut writes = Vec::with_capacity(entries.len());
+        let mut senders = Vec::with_capacity(entries.len());
+        let mut pendings = Vec::with_capacity(entries.len());
+        for (needle, fsync, response_tx, pending) in entries {
+            // A delete reaching a remounted volume cannot cancel a write
+            // queued for the old one.
+            if let (Ok(vol), Some(pending)) = (&vol, &pending)
+                && !vol.tracks_pending_write(pending)
+            {
+                let _ = response_tx.send(Err(VolumeError::NotFound));
+                continue;
+            }
+            writes.push((needle, fsync));
+            senders.push(response_tx);
+            pendings.push(pending);
+        }
+        let results: Vec<WriteResult> = match vol {
+            Ok(vol) => vol.write_pending_needles(&mut writes, &pendings),
             Err(e) => writes
                 .iter()
                 .map(|_| match e {
@@ -577,6 +599,7 @@ mod tests {
                     fsync: true,
                     response_tx,
                     queue_permit: None,
+                    pending: None,
                 }
             })
             .collect();
@@ -669,6 +692,7 @@ mod tests {
             fsync: false,
             response_tx,
             queue_permit: None,
+            pending: None,
         }];
         let writer = {
             let state = state.clone();

seaweed-volume/src/storage/volume.rs

@@ -1404,6 +1404,8 @@ pub struct Volume {
     /// `data_file_sync_lock` -> `data_file_access_control` -> `io`.
     data_file_sync_lock: std::sync::Mutex<()>,
 
+    pending_writes: Arc<PendingWrites>,
+
     /// Go's superBlockAccessLock.
     super_block: RwLock<SuperBlock>,
 
@@ -1453,6 +1455,36 @@ pub struct Volume {
 const NO_WRITE_OR_DELETE: u8 = 1;
 const NO_WRITE_CAN_DELETE: u8 = 2;
 
+#[derive(Default)]
+struct PendingWrites(Mutex<HashMap<NeedleId, Vec<Arc<CancelFlag>>>>);
+
+struct CancelFlag {
+    cookie: Cookie,
+    cancelled: AtomicBool,
+}
+
+/// A write that has entered the volume but not returned yet. A delete of its
+/// needle in that window cancels it: the write may still be queued behind the
+/// volume locks, and landing after the delete would revive the needle on this
+/// copy alone. Dropping it untracks the write.
+pub struct PendingWrite {
+    writes: Arc<PendingWrites>,
+    id: NeedleId,
+    flag: Arc<CancelFlag>,
+}
+
+impl Drop for PendingWrite {
+    fn drop(&mut self) {
+        let mut by_id = self.writes.0.lock().unwrap();
+        if let Some(writes) = by_id.get_mut(&self.id) {
+            writes.retain(|w| !Arc::ptr_eq(w, &self.flag));
+            if writes.is_empty() {
+                by_id.remove(&self.id);
+            }
+        }
+    }
+}
+
 /// What a volume is created with beyond its id, directories and index kind:
 /// the tail of Go's `NewVolume` argument list. The default is an empty
 /// collection with no replication, no TTL and no preallocation at the
@@ -1527,6 +1559,7 @@ impl Volume {
             needle_map_kind,
             data_file_access_control: Arc::new(DataFileAccessControl::default()),
             data_file_sync_lock: std::sync::Mutex::new(()),
+            pending_writes: Arc::default(),
             super_block: RwLock::new(SuperBlock {
                 replica_placement: replica_placement.unwrap_or_default(),
                 ttl: ttl.unwrap_or(crate::storage::needle::ttl::TTL::EMPTY),
@@ -1584,6 +1617,7 @@ impl Volume {
             needle_map_kind: NeedleMapKind::InMemory,
             data_file_access_control: Arc::new(DataFileAccessControl::default()),
             data_file_sync_lock: std::sync::Mutex::new(()),
+            pending_writes: Arc::default(),
             super_block: RwLock::new(SuperBlock::default()),
             no_write: AtomicU8::new(0),
             io_unavailable: Arc::default(),
@@ -2776,6 +2810,15 @@ impl Volume {
     pub fn write_needle_unsynced_or_refused(
         &self,
         n: &mut Needle,
+    ) -> Result<Result<(u64, Size, bool), VolumeError>, VolumeError> {
+        self.write_pending_needle_unsynced(n, None)
+    }
+
+    /// write_needle_unsynced_or_refused for a write a delete can cancel.
+    pub fn write_pending_needle_unsynced(
+        &self,
+        n: &mut Needle,
+        pending: Option<&PendingWrite>,
     ) -> Result<Result<(u64, Size, bool), VolumeError>, VolumeError> {
         let _sync_guard = self.data_file_sync_lock.lock().unwrap();
         let _guard = self.data_file_access_control.write_lock();
@@ -2783,7 +2826,9 @@ impl Volume {
         if let Err(refused) = self.check_writable() {
             return Ok(Err(refused));
         }
-        self.do_write_request(&mut io, n, true, false).map(Ok)
+        let written = self.do_write_request(&mut io, n, true, false)?;
+        self.drop_cancelled_write(&mut io, pending, n, &written)?;
+        Ok(Ok(written))
     }
 
     /// Write a batch of needles the way Go's processBatch does: one .dat
@@ -2797,6 +2842,16 @@ impl Volume {
     pub fn write_needles_grouped(
         &self,
         writes: &mut [(Needle, bool)],
+    ) -> Vec<Result<(u64, Size, bool), VolumeError>> {
+        self.write_pending_needles(writes, &[])
+    }
+
+    /// write_needles_grouped for writes a delete can cancel, `pending[i]`
+    /// tracking `writes[i]`.
+    pub fn write_pending_needles(
+        &self,
+        writes: &mut [(Needle, bool)],
+        pending: &[Option<PendingWrite>],
     ) -> Vec<Result<(u64, Size, bool), VolumeError>> {
         // The sync lock serializes a .dat mutation with the flush that makes
         // it durable: a failed sync's rollback can never clobber an append
@@ -2811,16 +2866,21 @@ impl Volume {
             let mut ids = HashSet::new();
             let len = rest.iter().take_while(|(n, _)| ids.insert(n.id)).count();
             let (run, tail) = std::mem::take(&mut rest).split_at_mut(len);
+            let tracked = pending.get(results.len()..).unwrap_or_default();
             if run.iter().any(|(_, fsync)| *fsync) {
-                results.extend(self.write_synced_run(run));
+                results.extend(self.write_synced_run(run, tracked));
             } else {
                 let _guard = self.data_file_access_control.write_lock();
                 let mut io = self.io.write().unwrap();
-                for (n, _) in run.iter_mut() {
-                    results.push(
-                        self.check_writable()
-                            .and_then(|()| self.do_write_request(&mut io, n, true, false)),
-                    );
+                for (i, (n, _)) in run.iter_mut().enumerate() {
+                    let written = self
+                        .check_writable()
+                        .and_then(|()| self.do_write_request(&mut io, n, true, false));
+                    results.push(written.and_then(|written| {
+                        let pending = tracked.get(i).and_then(Option::as_ref);
+                        self.drop_cancelled_write(&mut io, pending, n, &written)
+                            .map(|()| written)
+                    }));
                 }
             }
             rest = tail;
@@ -2831,6 +2891,7 @@ impl Volume {
     fn write_synced_run(
         &self,
         run: &mut [(Needle, bool)],
+        pending: &[Option<PendingWrite>],
     ) -> Vec<Result<(u64, Size, bool), VolumeError>> {
         // Per entry: Some(offset) once appended, None when it dedups.
         let mut staged = Vec::with_capacity(run.len());
@@ -2874,6 +2935,7 @@ impl Volume {
             }
         }
 
+        let mut took_back = false;
         {
             let _guard = self.data_file_access_control.write_lock();
             let mut io = self.io.write().unwrap();
@@ -2890,24 +2952,37 @@ impl Volume {
             // writes, so the entries after it are refused the way a lone
             // write would be.
             let mut refused = false;
-            for ((n, fsync), r) in run.iter().zip(staged.iter_mut()) {
+            for (i, ((n, fsync), r)) in run.iter().zip(staged.iter_mut()).enumerate() {
                 if refused {
                     *r = Err(self
                         .check_writable()
                         .err()
                         .unwrap_or(VolumeError::ReadOnly(self.id)));
-                } else if let Ok(Some(offset)) = *r
-                    && let Err(e) = self.publish_write(&mut io, n, offset, *fsync)
-                {
-                    *r = Err(e);
-                    refused = *fsync;
+                } else if let Ok(Some(offset)) = *r {
+                    if let Err(e) = self.publish_write(&mut io, n, offset, *fsync) {
+                        *r = Err(e);
+                        refused = *fsync;
+                    } else if let Err(e) = self.drop_cancelled_write(
+                        &mut io,
+                        pending.get(i).and_then(Option::as_ref),
+                        n,
+                        &(offset, n.size, false),
+                    ) {
+                        took_back |= matches!(e, VolumeError::Deleted);
+                        *r = Err(e);
+                    }
                 }
             }
         }
 
         if sync {
             let io = self.io.read().unwrap();
-            if let Err(e) = self.flush_idx(&io) {
+            let flushed = if took_back {
+                self.flush_taken_back(&io)
+            } else {
+                Ok(())
+            };
+            if let Err(e) = flushed.and_then(|()| self.flush_idx(&io)) {
                 return staged
                     .into_iter()
                     .map(|r| r.and_then(|_| Err(run_error(&e))))
@@ -3028,6 +3103,17 @@ impl Volume {
         Ok(())
     }
 
+    /// Flush the tombstones drop_cancelled_write appended after the .dat sync,
+    /// so their .idx rows never outlive them. Like flush_idx, a failure cannot
+    /// be undone and stops the volume taking writes.
+    fn flush_taken_back(&self, io: &VolumeIo) -> Result<(), VolumeError> {
+        self.flush_dat(io).map_err(|e| {
+            self.check_read_write_error(Some(&e));
+            self.set_no_write_flag(NO_WRITE_OR_DELETE, true);
+            VolumeError::Io(e)
+        })
+    }
+
     /// Take an unflushed append back off the .dat after its sync failed.
     pub(crate) fn truncate_dat(&self, io: &VolumeIo, len: u64) -> io::Result<()> {
         #[cfg(test)]
@@ -3410,10 +3496,79 @@ impl Volume {
         Ok((offset, n.size, actual_size))
     }
 
+    pub fn track_pending_write(&self, n: &Needle) -> PendingWrite {
+        let flag = Arc::new(CancelFlag {
+            cookie: n.cookie,
+            cancelled: AtomicBool::new(false),
+        });
+        self.pending_writes
+            .0
+            .lock()
+            .unwrap()
+            .entry(n.id)
+            .or_default()
+            .push(flag.clone());
+        PendingWrite {
+            writes: self.pending_writes.clone(),
+            id: n.id,
+            flag,
+        }
+    }
+
+    /// Whether the write was tracked on this volume rather than on one that
+    /// has since been unmounted under the same id.
+    pub fn tracks_pending_write(&self, pending: &PendingWrite) -> bool {
+        Arc::ptr_eq(&pending.writes, &self.pending_writes)
+    }
+
+    /// Cancels the writes of the needle that are in flight. A caller that found
+    /// the needle missing must look it up again afterwards: a write that
+    /// finished in between is no longer pending.
+    pub fn cancel_pending_writes(&self, id: NeedleId, cookie: Cookie) {
+        if let Some(writes) = self.pending_writes.0.lock().unwrap().get(&id) {
+            for w in writes.iter().filter(|w| w.cookie == cookie) {
+                w.cancelled.store(true, Ordering::Release);
+            }
+        }
+    }
+
+    /// Tombstones a cancelled write right after it is published, under the
+    /// same lock hold, so no reader, later write or compaction sees it live
+    /// in between.
+    fn drop_cancelled_write(
+        &self,
+        io: &mut VolumeIo,
+        pending: Option<&PendingWrite>,
+        n: &Needle,
+        written: &(u64, Size, bool),
+    ) -> Result<(), VolumeError> {
+        let (offset, _, unchanged) = *written;
+        if unchanged || !pending.is_some_and(|p| p.flag.cancelled.load(Ordering::Acquire)) {
+            return Ok(());
+        }
+        let live = match io.nm.as_ref() {
+            Some(nm) => nm.get(n.id)?.is_some_and(|nv| {
+                !nv.size.is_deleted() && nv.offset == Offset::from_actual_offset(offset as i64)
+            }),
+            None => false,
+        };
+        if !live {
+            return Ok(());
+        }
+        let mut tombstone = Needle {
+            id: n.id,
+            cookie: n.cookie,
+            ..Needle::default()
+        };
+        self.do_delete_request(io, &mut tombstone)?;
+        Err(VolumeError::Deleted)
+    }
+
     // ---- Delete ----
 
     /// Delete a needle from the volume.
     pub fn delete_needle(&self, n: &mut Needle) -> Result<Size, VolumeError> {
+        self.cancel_pending_writes(n.id, n.cookie);
         let _sync_guard = self.data_file_sync_lock.lock().unwrap();
         let _guard = self.data_file_access_control.write_lock();
         let mut io = self.io.write().unwrap();
@@ -6158,6 +6313,22 @@ impl Volume {
         self.io.read().unwrap()
     }
 
+    /// Holds the lock a write waits on behind another write's fsync.
+    #[cfg(test)]
+    pub(crate) fn sync_lock_for_test(&self) -> std::sync::MutexGuard<'_, ()> {
+        self.data_file_sync_lock.lock().unwrap()
+    }
+
+    #[cfg(test)]
+    pub(crate) fn pending_writes_for_test(&self, id: NeedleId) -> usize {
+        self.pending_writes
+            .0
+            .lock()
+            .unwrap()
+            .get(&id)
+            .map_or(0, Vec::len)
+    }
+
     #[cfg(test)]
     pub(crate) fn fail_next_fsync_for_test(&self, fail: bool) {
         self.io.write().unwrap().fail_fsync_for_test = fail;
@@ -8484,6 +8655,171 @@ mod tests {
         assert!(matches!(err, VolumeError::Deleted));
     }
 
+    fn pending_needle(id: u64, data: &[u8]) -> Needle {
+        Needle {
+            id: NeedleId(id),
+            cookie: Cookie(0xcc),
+            data: data.to_vec(),
+            data_size: data.len() as u32,
+            ..Needle::default()
+        }
+    }
+
+    fn write_pending(
+        v: &Volume,
+        n: Needle,
+        fsync: bool,
+        pending: PendingWrite,
+    ) -> Result<(u64, Size, bool), VolumeError> {
+        v.write_pending_needles(&mut [(n, fsync)], &[Some(pending)])
+            .pop()
+            .unwrap()
+    }
+
+    /// A delete that finds nothing can still have the write it undoes in
+    /// flight. That write must not land live afterwards.
+    #[test]
+    fn test_delete_of_missing_needle_cancels_pending_write() {
+        for fsync in [false, true] {
+            let tmp = TempDir::new().unwrap();
+            let dir = tmp.path().to_str().unwrap();
+            let v = make_test_volume(dir);
+
+            let n = pending_needle(1, b"late write");
+            let pending = v.track_pending_write(&n);
+            v.delete_needle(&mut pending_needle(1, b"")).unwrap();
+            let err = write_pending(&v, n, fsync, pending).unwrap_err();
+            assert!(matches!(err, VolumeError::Deleted));
+
+            let err = v.read_needle(&mut pending_needle(1, b"")).unwrap_err();
+            assert!(matches!(err, VolumeError::Deleted));
+            assert!(v.pending_writes.0.lock().unwrap().is_empty());
+
+            v.close();
+            let v = Volume::new(
+                dir,
+                dir,
+                VolumeId(1),
+                NeedleMapKind::InMemory,
+                &VolumeSpec::default(),
+            )
+            .unwrap();
+            assert!(v.read_needle(&mut pending_needle(1, b"")).is_err());
+        }
+    }
+
+    /// The unsynced write path takes a cancelled write back the same way.
+    #[test]
+    fn test_unsynced_write_cancelled_by_delete() {
+        let tmp = TempDir::new().unwrap();
+        let v = make_test_volume(tmp.path().to_str().unwrap());
+
+        let mut n = pending_needle(1, b"late write");
+        let pending = v.track_pending_write(&n);
+        v.cancel_pending_writes(n.id, n.cookie);
+        let err = v
+            .write_pending_needle_unsynced(&mut n, Some(&pending))
+            .unwrap_err();
+        assert!(matches!(err, VolumeError::Deleted));
+        let err = v.read_needle(&mut pending_needle(1, b"")).unwrap_err();
+        assert!(matches!(err, VolumeError::Deleted));
+    }
+
+    /// A delete that finds a copy also cancels a retry of the same write that
+    /// is still in flight, so the retry does not bring the needle back.
+    #[test]
+    fn test_delete_cancels_pending_rewrite() {
+        let tmp = TempDir::new().unwrap();
+        let dir = tmp.path().to_str().unwrap();
+        let v = make_test_volume(dir);
+
+        v.write_needle(&mut pending_needle(1, b"first"), true, false)
+            .unwrap();
+        let rewrite = pending_needle(1, b"retry");
+        let pending = v.track_pending_write(&rewrite);
+        v.delete_needle(&mut pending_needle(1, b"")).unwrap();
+        let err = write_pending(&v, rewrite, false, pending).unwrap_err();
+        assert!(matches!(err, VolumeError::Deleted));
+        let err = v.read_needle(&mut pending_needle(1, b"")).unwrap_err();
+        assert!(matches!(err, VolumeError::Deleted));
+    }
+
+    /// Only writes already in flight and carrying the delete's cookie are
+    /// cancelled.
+    #[test]
+    fn test_pending_write_cancel_leaves_later_writes() {
+        let tmp = TempDir::new().unwrap();
+        let dir = tmp.path().to_str().unwrap();
+        let v = make_test_volume(dir);
+
+        v.delete_needle(&mut pending_needle(1, b"")).unwrap();
+        let n = pending_needle(1, b"after the delete");
+        let pending = v.track_pending_write(&n);
+        write_pending(&v, n, false, pending).unwrap();
+        let mut got = pending_needle(1, b"");
+        v.read_needle(&mut got).unwrap();
+        assert_eq!(got.data, b"after the delete");
+
+        let n = pending_needle(2, b"other cookie");
+        let pending = v.track_pending_write(&n);
+        v.cancel_pending_writes(n.id, Cookie(n.cookie.0 + 1));
+        write_pending(&v, n, false, pending).unwrap();
+        let mut got = pending_needle(2, b"");
+        v.read_needle(&mut got).unwrap();
+        assert_eq!(got.data, b"other cookie");
+    }
+
+    /// The cancelled write is taken back before the next write of the needle
+    /// runs, so an identical write sent after the delete appends its own copy
+    /// instead of matching the record about to be deleted.
+    #[test]
+    fn test_write_after_cancelled_write_stays_live() {
+        for fsync in [false, true] {
+            let tmp = TempDir::new().unwrap();
+            let v = make_test_volume(tmp.path().to_str().unwrap());
+
+            let first = pending_needle(1, b"same bytes");
+            let first_pending = v.track_pending_write(&first);
+            v.cancel_pending_writes(first.id, first.cookie);
+            let retry = pending_needle(1, b"same bytes");
+            let retry_pending = v.track_pending_write(&retry);
+            let mut results = v.write_pending_needles(
+                &mut [(first, fsync), (retry, fsync)],
+                &[Some(first_pending), Some(retry_pending)],
+            );
+            let retried = results.pop().unwrap().unwrap();
+            assert!(
+                !retried.2,
+                "fsync={fsync}: the retry matched the cancelled write"
+            );
+            assert!(matches!(results.pop().unwrap(), Err(VolumeError::Deleted)));
+
+            let mut got = pending_needle(1, b"");
+            v.read_needle(&mut got).unwrap();
+            assert_eq!(got.data, b"same bytes");
+        }
+    }
+
+    /// A durable write taken back appends its tombstone after the .dat sync;
+    /// the tombstone is flushed before the .idx row that points at it.
+    #[test]
+    fn test_taken_back_tombstone_is_synced() {
+        let tmp = TempDir::new().unwrap();
+        let v = make_test_volume(tmp.path().to_str().unwrap());
+
+        let n = pending_needle(1, b"late write");
+        let pending = v.track_pending_write(&n);
+        v.cancel_pending_writes(n.id, n.cookie);
+        let err = write_pending(&v, n, true, pending).unwrap_err();
+        assert!(matches!(err, VolumeError::Deleted));
+        assert_eq!(v.sync_counts_for_test(), (2, 1));
+
+        let n = pending_needle(2, b"kept");
+        let pending = v.track_pending_write(&n);
+        write_pending(&v, n, true, pending).unwrap();
+        assert_eq!(v.sync_counts_for_test(), (3, 2));
+    }
+
     // Guard the Rust integrity-check tombstone path against the Go regression
     // where verifyDeletedNeedleIntegrity forwarded TombstoneFileSize into the
     // needle-size check, mismatched against the on-disk Size=0 header, and

weed/server/volume_grpc_batch_delete.go

@@ -10,6 +10,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
 	"github.com/seaweedfs/seaweedfs/weed/storage"
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	"github.com/seaweedfs/seaweedfs/weed/storage/types"
 )
 
 func (vs *VolumeServer) BatchDelete(ctx context.Context, req *volume_server_pb.BatchDeleteRequest) (*volume_server_pb.BatchDeleteResponse, error) {
@@ -37,8 +38,9 @@ func (vs *VolumeServer) BatchDelete(ctx context.Context, req *volume_server_pb.B
 
 		n := new(needle.Needle)
 		volumeId, _ := needle.NewVolumeId(vid)
+		var fidCookie types.Cookie
 		if req.SkipCookieCheck {
-			n.Id, _, err = needle.ParseNeedleIdCookie(id_cookie)
+			n.Id, fidCookie, err = needle.ParseNeedleIdCookie(id_cookie)
 			if err != nil {
 				resp.Results = append(resp.Results, &volume_server_pb.DeleteResult{
 					FileId: fid,
@@ -99,6 +101,10 @@ func (vs *VolumeServer) BatchDelete(ctx context.Context, req *volume_server_pb.B
 
 		n.LastModified = now
 		if !isEcVolume {
+			if req.SkipCookieCheck {
+				// the delete cancels in-flight writes carrying this cookie
+				n.Cookie = fidCookie
+			}
 			if size, err := vs.store.DeleteVolumeNeedle(volumeId, n); err != nil {
 				resp.Results = append(resp.Results, &volume_server_pb.DeleteResult{
 					FileId: fid,

weed/server/volume_server_handlers_write.go

@@ -123,6 +123,13 @@ func (vs *VolumeServer) DeleteHandler(w http.ResponseWriter, r *http.Request) {
 	}
 
 	_, ok := vs.store.ReadVolumeNeedle(volumeId, n, deleteReadOption(), nil)
+	if ok != nil && r.FormValue("type") == "replicate" {
+		// the write this delete undoes can still be queued here
+		if v := vs.store.GetVolume(volumeId); v != nil {
+			v.CancelPendingWrites(n.Id, cookie)
+			_, ok = vs.store.ReadVolumeNeedle(volumeId, n, deleteReadOption(), nil)
+		}
+	}
 	if ok != nil {
 		m := make(map[string]uint32)
 		m["size"] = 0

weed/server/volume_server_handlers_write_test.go

@@ -0,0 +1,141 @@
+package weed_server
+
+import (
+	"bytes"
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"runtime"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
+	"github.com/seaweedfs/seaweedfs/weed/security"
+	"github.com/seaweedfs/seaweedfs/weed/stats"
+	"github.com/seaweedfs/seaweedfs/weed/storage"
+	"github.com/seaweedfs/seaweedfs/weed/storage/backend"
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	"github.com/seaweedfs/seaweedfs/weed/storage/types"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+	"github.com/stretchr/testify/require"
+)
+
+// heldSyncBackend parks the first .dat fsync until released: the window where
+// a durable write is appended but not indexed yet.
+type heldSyncBackend struct {
+	backend.BackendStorageFile
+	once    sync.Once
+	entered chan struct{}
+	release chan struct{}
+}
+
+func (b *heldSyncBackend) Sync() error {
+	b.once.Do(func() {
+		close(b.entered)
+		<-b.release
+	})
+	return b.BackendStorageFile.Sync()
+}
+
+// startHeldWrite starts a durable write of needle 1 on a fresh volume and
+// returns once it is parked in its .dat fsync, appended but not indexed.
+func startHeldWrite(t *testing.T) (store *storage.Store, fid string, release func(), written <-chan error) {
+	store = storage.NewStore(nil, "localhost", 8080, 18080, "http://localhost:8080", "store-id",
+		[]string{t.TempDir()}, []int32{10}, []util.MinFreeSpace{{}}, "", storage.NeedleMapInMemory,
+		[]types.DiskType{types.HardDriveType}, nil, 0, stats.DefaultDiskIOProbeConfig())
+	t.Cleanup(store.Close)
+	require.NoError(t, store.AddVolume(1, "", storage.NeedleMapInMemory, "000", "", 0, needle.Version3, 0, types.HardDriveType, 0))
+	v := store.GetVolume(1)
+	held := &heldSyncBackend{BackendStorageFile: v.DataBackend, entered: make(chan struct{}), release: make(chan struct{})}
+	v.DataBackend = held
+
+	n := &needle.Needle{Id: 1, Cookie: 0x1234, Data: []byte("late write")}
+	n.Checksum = needle.NewCRC(n.Data)
+	done := make(chan error, 1)
+	go func() {
+		_, err := store.WriteVolumeNeedle(1, n, true, true)
+		done <- err
+	}()
+	<-held.entered
+	var releaseOnce sync.Once
+	release = func() { releaseOnce.Do(func() { close(held.release) }) }
+	t.Cleanup(release)
+	return store, needle.NewFileIdFromNeedle(1, n).String(), release, done
+}
+
+func requireTakenBack(t *testing.T, store *storage.Store, written <-chan error, takenBack bool) {
+	t.Helper()
+	err := <-written
+	_, readErr := store.ReadVolumeNeedle(1, &needle.Needle{Id: 1}, nil, nil)
+	if takenBack {
+		require.ErrorIs(t, err, storage.ErrorDeleted)
+		require.ErrorIs(t, readErr, storage.ErrorDeleted)
+	} else {
+		require.NoError(t, err)
+		require.NoError(t, readErr)
+	}
+}
+
+// A replica delete that arrives while the write it undoes is in its fsync
+// finds nothing, yet the write must not land live afterwards. A plain delete
+// that finds nothing is not sent on to the replicas, so it leaves the write.
+func TestDeleteHandlerCancelsAWriteInItsFsyncWindow(t *testing.T) {
+	for _, tc := range []struct {
+		query string
+		live  bool
+	}{
+		{"?type=replicate", false},
+		{"", true},
+	} {
+		t.Run(tc.query, func(t *testing.T) {
+			store, fid, release, written := startHeldWrite(t)
+			resp := httptest.NewRecorder()
+			vs := &VolumeServer{store: store, guard: security.NewGuard(nil, "", 0, "", 0)}
+			vs.DeleteHandler(resp, httptest.NewRequest(http.MethodDelete, "/"+fid+tc.query, nil))
+			require.Equal(t, http.StatusNotFound, resp.Code)
+			release()
+			requireTakenBack(t, store, written, !tc.live)
+		})
+	}
+}
+
+// A batch delete that skips the cookie check still cancels the in-flight
+// writes of the file id it names, and only those.
+func TestBatchDeleteCancelsAWriteInItsFsyncWindow(t *testing.T) {
+	for _, tc := range []struct {
+		cookie    string
+		takenBack bool
+	}{
+		{"", true},
+		{"00004321", false},
+	} {
+		t.Run(tc.cookie, func(t *testing.T) {
+			store, fid, release, written := startHeldWrite(t)
+			if tc.cookie != "" {
+				fid = fid[:len(fid)-8] + tc.cookie
+			}
+			deleted := make(chan error, 1)
+			go func() {
+				_, err := (&VolumeServer{store: store}).BatchDelete(context.Background(), &volume_server_pb.BatchDeleteRequest{
+					FileIds:         []string{fid},
+					SkipCookieCheck: true,
+				})
+				deleted <- err
+			}()
+			// the delete waits behind the held write's sync lock
+			require.Eventually(t, func() bool {
+				buf := make([]byte, 1<<20)
+				return bytes.Contains(buf[:runtime.Stack(buf, true)], []byte("(*Volume).syncDelete"))
+			}, 10*time.Second, time.Millisecond)
+			release()
+			require.NoError(t, <-deleted)
+			err := <-written
+			if tc.takenBack {
+				require.ErrorIs(t, err, storage.ErrorDeleted)
+			} else {
+				require.NoError(t, err)
+			}
+		})
+	}
+}

weed/storage/volume.go

@@ -59,6 +59,8 @@ type Volume struct {
 	asyncRequestsChan chan *needle.AsyncRequest
 	asyncWorkerClosed bool
 
+	pendingWrites pendingWrites
+
 	lastModifiedTsSeconds    uint64         // unix time in seconds
 	lastAppendAtNs           uint64         // unix time in nanoseconds
 	lastWriteAppendAtNs      uint64         // AppendAtNs of the newest write; tombstones don't move it

weed/storage/volume_pending_write.go

@@ -0,0 +1,95 @@
+package storage
+
+import (
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	. "github.com/seaweedfs/seaweedfs/weed/storage/types"
+)
+
+// pendingWrite is a write that has entered the volume but not returned yet.
+// A delete of its needle in that window cancels it: the write may still be
+// queued behind the volume locks, and landing after the delete would revive
+// the needle on this copy alone.
+type pendingWrite struct {
+	n         *needle.Needle
+	cookie    Cookie
+	cancelled bool
+}
+
+type pendingWrites struct {
+	sync.Mutex
+	byId map[NeedleId][]*pendingWrite
+}
+
+func (v *Volume) trackPendingWrite(n *needle.Needle) *pendingWrite {
+	w := &pendingWrite{n: n, cookie: n.Cookie}
+	v.pendingWrites.Lock()
+	defer v.pendingWrites.Unlock()
+	if v.pendingWrites.byId == nil {
+		v.pendingWrites.byId = make(map[NeedleId][]*pendingWrite)
+	}
+	v.pendingWrites.byId[n.Id] = append(v.pendingWrites.byId[n.Id], w)
+	return w
+}
+
+func (v *Volume) untrackPendingWrite(w *pendingWrite) {
+	v.pendingWrites.Lock()
+	defer v.pendingWrites.Unlock()
+	id := w.n.Id
+	writes := v.pendingWrites.byId[id]
+	for i, x := range writes {
+		if x == w {
+			writes = append(writes[:i], writes[i+1:]...)
+			break
+		}
+	}
+	if len(writes) == 0 {
+		delete(v.pendingWrites.byId, id)
+	} else {
+		v.pendingWrites.byId[id] = writes
+	}
+}
+
+// CancelPendingWrites cancels the writes of the needle that are in flight. A
+// caller that found the needle missing must look it up again afterwards: a
+// write that finished in between is no longer pending.
+func (v *Volume) CancelPendingWrites(id NeedleId, cookie Cookie) {
+	v.pendingWrites.Lock()
+	defer v.pendingWrites.Unlock()
+	for _, w := range v.pendingWrites.byId[id] {
+		if w.cookie == cookie {
+			w.cancelled = true
+		}
+	}
+}
+
+func (v *Volume) isCancelledWrite(n *needle.Needle) bool {
+	v.pendingWrites.Lock()
+	defer v.pendingWrites.Unlock()
+	for _, w := range v.pendingWrites.byId[n.Id] {
+		if w.n == n {
+			return w.cancelled
+		}
+	}
+	return false
+}
+
+// dropCancelledWrite tombstones a cancelled write right after it is published,
+// under the same dataFileSyncLock and dataFileAccessLock hold, so no reader,
+// later write or compaction sees it live in between.
+func (v *Volume) dropCancelledWrite(n *needle.Needle, offset uint64) error {
+	if !v.isCancelledWrite(n) {
+		return nil
+	}
+	if nv, ok := v.nm.Get(n.Id); !ok || nv.Size.IsDeleted() || nv.Offset != ToOffset(int64(offset)) {
+		return nil
+	}
+	tombstone := &needle.Needle{Id: n.Id, Cookie: n.Cookie, LastModified: uint64(time.Now().Unix())}
+	if _, err := v.doDeleteRequest(tombstone); err != nil {
+		return err
+	}
+	return fmt.Errorf("needle %s: %w", needle.NewFileIdFromNeedle(v.Id, n), ErrorDeleted)
+}

weed/storage/volume_pending_write_test.go

@@ -0,0 +1,229 @@
+package storage
+
+import (
+	"fmt"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/storage/backend"
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
+	. "github.com/seaweedfs/seaweedfs/weed/storage/types"
+	"github.com/stretchr/testify/require"
+)
+
+func waitForPendingWrites(t *testing.T, v *Volume, id NeedleId, count int) {
+	t.Helper()
+	deadline := time.Now().Add(10 * time.Second)
+	for {
+		v.pendingWrites.Lock()
+		got := len(v.pendingWrites.byId[id])
+		v.pendingWrites.Unlock()
+		if got >= count {
+			return
+		}
+		if time.Now().After(deadline) {
+			t.Fatalf("%d of %d writes of needle %d registered", got, count, id)
+		}
+		time.Sleep(time.Millisecond)
+	}
+}
+
+func requireNeedleDeleted(t *testing.T, v *Volume, id NeedleId) {
+	t.Helper()
+	_, err := v.readNeedle(&needle.Needle{Id: id}, nil, nil)
+	require.ErrorIs(t, err, ErrorDeleted)
+}
+
+// A replicate delete that finds nothing can still have the write it undoes
+// queued behind the volume locks. That write must not land live afterwards.
+func TestDeleteOfMissingNeedleCancelsQueuedWrite(t *testing.T) {
+	for _, fsync := range []bool{false, true} {
+		t.Run(fmt.Sprintf("fsync=%v", fsync), func(t *testing.T) {
+			dir := t.TempDir()
+			v, err := NewVolume(dir, dir, "", 1, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0)
+			require.NoError(t, err)
+
+			n := newRandomNeedle(1)
+			v.dataFileSyncLock.Lock()
+			written := make(chan error, 1)
+			go func() {
+				_, _, _, err := v.writeNeedle2(n, true, fsync, false)
+				written <- err
+			}()
+			waitForPendingWrites(t, v, n.Id, 1)
+
+			_, err = v.readNeedle(&needle.Needle{Id: n.Id}, nil, nil)
+			require.ErrorIs(t, err, ErrorNotFound)
+			v.CancelPendingWrites(n.Id, n.Cookie)
+			v.dataFileSyncLock.Unlock()
+
+			require.ErrorIs(t, <-written, ErrorDeleted)
+			requireNeedleDeleted(t, v, n.Id)
+			require.Empty(t, v.pendingWrites.byId)
+
+			v.Close()
+			v, err = NewVolume(dir, dir, "", 1, NeedleMapInMemory, nil, nil, 0, needle.GetCurrentVersion(), 0, 0)
+			require.NoError(t, err)
+			defer v.Close()
+			requireNeedleDeleted(t, v, n.Id)
+		})
+	}
+}
+
+// A delete that finds a copy also cancels a retry of the same write that is
+// still queued, so the retry does not bring the needle back.
+func TestDeleteCancelsQueuedRewrite(t *testing.T) {
+	v := newWorkerTestVolume(t)
+	n := newRandomNeedle(1)
+	_, _, _, err := v.writeNeedle2(n, true, false, false)
+	require.NoError(t, err)
+
+	rewrite := newRandomNeedle(1)
+	rewrite.Cookie = n.Cookie
+	// holds the durable rewrite after it registers, before it reaches the worker
+	v.asyncWorkerLock.Lock()
+	written := make(chan error, 1)
+	go func() {
+		_, _, _, err := v.writeNeedle2(rewrite, true, true, false)
+		written <- err
+	}()
+	waitForPendingWrites(t, v, n.Id, 1)
+
+	_, err = v.deleteNeedle2(&needle.Needle{Id: n.Id, Cookie: n.Cookie})
+	require.NoError(t, err)
+	requireNeedleDeleted(t, v, n.Id)
+	v.asyncWorkerLock.Unlock()
+
+	require.ErrorIs(t, <-written, ErrorDeleted)
+	requireNeedleDeleted(t, v, n.Id)
+	require.Empty(t, v.pendingWrites.byId)
+}
+
+// Only writes already in flight are cancelled: writing the needle again after
+// the delete works as before.
+func TestWriteAfterDeleteIsNotCancelled(t *testing.T) {
+	v := newWorkerTestVolume(t)
+	n := newRandomNeedle(1)
+	v.CancelPendingWrites(n.Id, n.Cookie)
+	_, err := v.deleteNeedle2(&needle.Needle{Id: n.Id, Cookie: n.Cookie})
+	require.NoError(t, err)
+
+	_, _, _, err = v.writeNeedle2(n, true, false, false)
+	require.NoError(t, err)
+	got := &needle.Needle{Id: n.Id}
+	_, err = v.readNeedle(got, nil, nil)
+	require.NoError(t, err)
+	require.Equal(t, n.Data, got.Data)
+}
+
+// A delete only cancels writes that carry its cookie.
+func TestCancelWithOtherCookieKeepsWrite(t *testing.T) {
+	v := newWorkerTestVolume(t)
+	n := newRandomNeedle(1)
+	v.dataFileSyncLock.Lock()
+	written := make(chan error, 1)
+	go func() {
+		_, _, _, err := v.writeNeedle2(n, true, false, false)
+		written <- err
+	}()
+	waitForPendingWrites(t, v, n.Id, 1)
+	v.CancelPendingWrites(n.Id, n.Cookie+1)
+	v.dataFileSyncLock.Unlock()
+
+	require.NoError(t, <-written)
+	got := &needle.Needle{Id: n.Id}
+	_, err := v.readNeedle(got, nil, nil)
+	require.NoError(t, err)
+	require.Equal(t, n.Data, got.Data)
+}
+
+// The cancelled write is taken back before the next write of the needle runs,
+// so an identical write sent after the delete appends its own copy instead of
+// matching the record about to be deleted, even in the same batch.
+func TestWriteAfterCancelledWriteStaysLive(t *testing.T) {
+	v := newWorkerTestVolume(t)
+	v.location = &DiskLocation{syncBatch: SyncBatchOption{
+		MinBytes:    1 << 30,
+		MinRequests: syncBatchMaxRequests,
+		Wait:        time.Minute,
+	}}
+	armed := make(chan chan time.Time, 1)
+	prevAfter := syncBatchAfter
+	syncBatchAfter = func(time.Duration) <-chan time.Time {
+		window := make(chan time.Time, 1)
+		armed <- window
+		return window
+	}
+	defer func() { syncBatchAfter = prevAfter }()
+
+	queue := func(n *needle.Needle) *needle.AsyncRequest {
+		w := v.trackPendingWrite(n)
+		t.Cleanup(func() { v.untrackPendingWrite(w) })
+		r := needle.NewAsyncRequest(n, true)
+		r.ActualSize = needle.GetActualSize(Size(len(n.Data)), v.Version())
+		require.True(t, v.asyncRequestAppend(r))
+		return r
+	}
+
+	n := newRandomNeedle(1)
+	first := queue(n)
+	window := <-armed
+	v.CancelPendingWrites(n.Id, n.Cookie)
+	retry := queue(&needle.Needle{Id: n.Id, Cookie: n.Cookie, Data: n.Data, Checksum: n.Checksum})
+	require.Eventually(t, func() bool { return len(v.asyncRequestsChan) == 0 }, 10*time.Second, time.Millisecond)
+	window <- time.Now()
+
+	_, _, _, err := first.WaitComplete()
+	require.ErrorIs(t, err, ErrorDeleted)
+	_, _, isUnchanged, err := retry.WaitComplete()
+	require.NoError(t, err)
+	require.False(t, isUnchanged, "the retry matched the cancelled write")
+	got := &needle.Needle{Id: n.Id}
+	_, err = v.readNeedle(got, nil, nil)
+	require.NoError(t, err)
+	require.Equal(t, n.Data, got.Data)
+}
+
+type syncedSizeBackend struct {
+	backend.BackendStorageFile
+	synced atomic.Int64
+}
+
+func (b *syncedSizeBackend) Sync() error {
+	size, _, err := b.GetStat()
+	if err != nil {
+		return err
+	}
+	b.synced.Store(size)
+	return b.BackendStorageFile.Sync()
+}
+
+// A durable write taken back during its fsync appends the tombstone after
+// that fsync; it must be flushed before the .idx row that points at it.
+func TestTakenBackTombstoneIsSynced(t *testing.T) {
+	for _, isStopping := range []bool{false, true} {
+		t.Run(fmt.Sprintf("inline=%v", isStopping), func(t *testing.T) {
+			v := newWorkerTestVolume(t)
+			synced := &syncedSizeBackend{BackendStorageFile: v.DataBackend}
+			v.DataBackend = synced
+
+			n := newRandomNeedle(1)
+			v.dataFileSyncLock.Lock()
+			written := make(chan error, 1)
+			go func() {
+				_, _, _, err := v.writeNeedle2(n, true, true, isStopping)
+				written <- err
+			}()
+			waitForPendingWrites(t, v, n.Id, 1)
+			v.CancelPendingWrites(n.Id, n.Cookie)
+			v.dataFileSyncLock.Unlock()
+
+			require.ErrorIs(t, <-written, ErrorDeleted)
+			size, _, err := v.DataBackend.GetStat()
+			require.NoError(t, err)
+			require.Equal(t, size, synced.synced.Load(), "the tombstone was not flushed")
+		})
+	}
+}

weed/storage/volume_write.go

@@ -236,19 +236,42 @@ func (v *Volume) syncWrite(n *needle.Needle, checkCookie bool, fsync bool) (offs
 		return 0, 0, false, syncErr
 	}
 	if !isUnchanged {
-		err = v.publishWriteRequest(n, offset, fsync)
+		if err = v.publishWriteRequest(n, offset, fsync); err == nil {
+			err = v.dropCancelledWrite(n, offset)
+		}
 	}
 	// The .idx row goes down before the write is acknowledged, same as the
 	// .dat record above; the access lock is dropped for the fsync the same
 	// way, since the row is already published.
 	v.dataFileAccessLock.Unlock()
-	if idxErr := v.syncIndex(); err == nil {
+	var idxErr error
+	if errors.Is(err, ErrorDeleted) {
+		idxErr = v.syncTakenBack(backend)
+	}
+	if idxErr == nil {
+		idxErr = v.syncIndex()
+	}
+	if err == nil {
 		err = idxErr
 	}
 	v.dataFileAccessLock.Lock()
 	return
 }
 
+// syncTakenBack flushes the tombstones dropCancelledWrite appended after the
+// .dat sync, so their .idx rows never outlive them. Like syncIndex, a failure
+// cannot be undone and stops the volume taking writes.
+func (v *Volume) syncTakenBack(backend backend.BackendStorageFile) error {
+	if err := backend.Sync(); err != nil {
+		v.checkReadWriteError(err)
+		v.noWriteLock.Lock()
+		v.noWriteOrDelete = true
+		v.noWriteLock.Unlock()
+		return fmt.Errorf("sync volume %d tombstone: %w", v.Id, err)
+	}
+	return nil
+}
+
 // syncIndex flushes the .idx, the second half of a durable write: the row is
 // already published and cannot be taken back, so a failed sync stops the
 // volume taking writes rather than acknowledge an index entry that may not
@@ -278,6 +301,8 @@ func (v *Volume) writeNeedle2(n *needle.Needle, checkCookie bool, fsync bool, is
 		return 0, 0, false, err
 	}
 
+	defer v.untrackPendingWrite(v.trackPendingWrite(n))
+
 	if n.Ttl == needle.EMPTY_TTL && v.Ttl != needle.EMPTY_TTL {
 		n.SetHasTtl()
 		n.Ttl = v.Ttl
@@ -304,7 +329,9 @@ func (v *Volume) doWriteRequest(n *needle.Needle, checkCookie bool, fsync bool)
 	if err != nil || isUnchanged {
 		return
 	}
-	err = v.publishWriteRequest(n, offset, fsync)
+	if err = v.publishWriteRequest(n, offset, fsync); err == nil {
+		err = v.dropCancelledWrite(n, offset)
+	}
 	return
 }
 
@@ -433,6 +460,8 @@ func (v *Volume) syncDelete(n *needle.Needle) (Size, error) {
 }
 
 func (v *Volume) deleteNeedle2(n *needle.Needle) (Size, error) {
+	v.CancelPendingWrites(n.Id, n.Cookie)
+
 	// todo: delete info is always appended no fsync, it may need fsync in future
 	fsync := false
 
@@ -629,6 +658,7 @@ func (v *Volume) processBatchRun(run []*needle.AsyncRequest) {
 	}
 
 	refused := false
+	tookBack := false
 	for i := 0; i < len(run); i++ {
 		if refused {
 			run[i].UpdateResult(0, 0, false, fmt.Errorf("volume %d is read only", v.Id))
@@ -639,6 +669,9 @@ func (v *Volume) processBatchRun(run []*needle.AsyncRequest) {
 				if perr := v.publishWriteRequest(run[i].N, stagedWriteOffsets[i], true); perr != nil {
 					run[i].UpdateResult(0, 0, false, perr)
 					refused = true
+				} else if derr := v.dropCancelledWrite(run[i].N, stagedWriteOffsets[i]); derr != nil {
+					run[i].UpdateResult(0, 0, false, derr)
+					tookBack = tookBack || errors.Is(derr, ErrorDeleted)
 				}
 			}
 		} else if stagedDeleteOk[i] {
@@ -652,7 +685,14 @@ func (v *Volume) processBatchRun(run []*needle.AsyncRequest) {
 	// rows cannot be rolled back, so a failed sync fails every request in the
 	// run instead of acknowledging an index entry that may not survive.
 	if needIndexSync {
-		if idxErr := v.syncIndex(); idxErr != nil {
+		var idxErr error
+		if tookBack {
+			idxErr = v.syncTakenBack(backend)
+		}
+		if idxErr == nil {
+			idxErr = v.syncIndex()
+		}
+		if idxErr != nil {
 			for i := 0; i < len(run); i++ {
 				run[i].UpdateResult(0, 0, false, idxErr)
 			}