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 <noreply@anthropic.com>

* volume(rust): untrack a write rejected for a remounted volume before replying

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

---------

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

seaweed-volume/src/server/write_queue.rs

@@ -376,9 +376,10 @@ fn process_batch(state: Arc<VolumeServerState>, batch: Vec<WriteRequest>) {
         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<VolumeServerState>, batch: Vec<WriteRequest>) {
                 })
                 .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);