From a2b9b37ee543c15dfb275882d9563a0a6dc22a36 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Thu, 12 Nov 2026 16:11:10 -0800 Subject: [PATCH] volume(rust): untrack queued writes before replying (#12595) * volume(rust): untrack queued writes before replying The durable write worker sent each write's result and only then dropped the batch's pending-write trackers, so a caller could see its write finish while the write was still tracked as pending. Go untracks in the caller before writeNeedle2 returns. Drop the trackers before sending the results so the Rust queue path matches. test_replicate_delete_cancels_a_queued_write asserted the count was zero right after the fsync write returned and failed about 1 run in 100 under load (Rust Unit Tests on master c2625dd). Co-Authored-By: Claude Opus 5.5 * volume(rust): untrack a write rejected for a remounted volume before replying Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Claude Opus 5.5 --- seaweed-volume/src/server/write_queue.rs | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/seaweed-volume/src/server/write_queue.rs b/seaweed-volume/src/server/write_queue.rs index 6286bc81c2..2398ec75d1 100644 --- a/seaweed-volume/src/server/write_queue.rs +++ b/seaweed-volume/src/server/write_queue.rs @@ -376,9 +376,10 @@ fn process_batch(state: Arc, batch: Vec) { 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) + if let (Ok(vol), Some(tracked)) = (&vol, &pending) + && !vol.tracks_pending_write(tracked) { + drop(pending); let _ = response_tx.send(Err(VolumeError::NotFound)); continue; } @@ -396,6 +397,9 @@ fn process_batch(state: Arc, batch: Vec) { }) .collect(), }; + // Untrack before replying, as Go does: a caller that sees its write + // finish must not still find it pending. + drop(pendings); for (response_tx, result) in senders.into_iter().zip(results) { // Send result back; ignore error if receiver dropped. let _ = response_tx.send(result);