rust volume: release the store write lock during needle I/O (#11823)

* volume: give Volume interior locks mirroring Go's

Volume's mutable state moves behind interior locks matching Go's
per-volume lock domains, so the store can hand out shared handles
instead of requiring exclusive &mut access:

- io: RwLock<VolumeIo> owns the .dat/remote handles, needle map,
  append cursor, compaction bookkeeping and write watermarks —
  Go's dataFileAccessLock-adjacent mutable set; helpers take the
  guard as &mut VolumeIo so nothing re-locks.
- super_block and volume_info each get a RwLock
  (superBlockAccessLock, volumeInfoRWLock); dir_idx gets its own
  so path building never contends with I/O.
- no_write packs Go's noWriteLock pair into AtomicU8;
  last_modified_ts_seconds and compaction_byte_per_second become
  atomics.

The data-file write lease (dataFileAccessLock) now covers every
mutating entry that previously relied on borrow exclusivity:
writes, deletes, blob writes, set_writable's flag/index swap,
remote-file updates, local-backend swaps, index relocation,
replica-placement rewrite, put_needle_index, close and destroy.
set_read_only_persist leases only its can_delete idx-attach branch
because mark_io_unavailable reaches it under a held lease.
commit_compact takes try_write_lock — callers hold the store lock,
so blocking would stall every volume behind one slow reader;
WouldBlock surfaces as COMMIT_WAITS_FOR_READS for the caller to
retry.

Lock order is fixed: lease -> io -> volume_info/super_block/dir_idx.

* storage: mount volumes as shared Arc<Volume> handles

DiskLocation's map owns Arc<Volume> so the store can hand a caller a
handle that stays valid across store-guard releases, instead of
borrowing &mut Volume under the store write lock.

- volumes: HashMap<VolumeId, Arc<Volume>>; find_volume returns a
  cloned Arc.
- find_volume_mut and volumes_mut are gone — every Volume method is
  &self now, so shared handles cover what exclusive borrows did.
- writable_volume_mut -> writable_volume(&self) -> Result<Arc<Volume>>
  keeps the same admission checks (exists, disk-space-low).
- write_volume_needle/write_volume_needles/delete_volume_needle are
  &self; lifecycle methods that mutate the map or mounted state stay
  &mut self.

* server: run needle I/O with the store guard released

The store write lock previously wrapped needle writes and deletes
end-to-end: process_batch held it across every grouped fsync, and the
HTTP/gRPC fallbacks, tail receiver and fetch-and-write paths did the
same. One volume's slow I/O then stalled admission and metadata on
every other volume.

- write_queue resolves each group's handle via writable_volume under
  a brief store read, then runs write_needles_grouped after the guard
  drops.
- HTTP write/DELETE fallbacks, the gRPC batch-delete EC recheck +
  delete, tail receiver and fetch_and_write local writes, blob writes
  and the quarantine/sync/set_writable/remote-file paths all run
  under store.read() now that every volume method they reach is
  &self.
- Lifecycle calls that mutate the volume map or mounted state
  (mount/unmount/delete, EC shard ops, compaction commits,
  load_new_volumes, heartbeat volume actions) keep store.write().

* test: cross-volume admission proceeds during another volume's write

Regression coverage for the store-lock fix:

- write_queue's test parks a volume-1 write inside its .dat append via
  a new needle_write_hook seam (mirroring needle_read_hook), then
  proves the store still hands out a lifecycle writer and a volume-2
  handle — while volume 1's data-file lease still excludes same-volume
  writers.
- tests/store_lock_repro.rs exercises the public API: concurrent
  read-guarded writes on two volumes, and a stale Arc<Volume> handle
  failing closed after unmount instead of writing into torn-down
  files.

* test: drop store write guards that shared handles made read-only

Test setup sites held store.write() purely to reach find_volume_mut;
with Arc<Volume> and &self methods the guard is read-only (clippy
readonly_write_lock). The volume lease still serializes the writes
inside.

* volume: drop a windows-only read_exact_at that shadows the io helper

cargo's windows build failed E0255: the free fn copied with the
interior-lock conversion collided with the unconditional
crate::storage::io::read_exact_at import, which already carries the
platform branches internally.

* server: release the store read guard before needle I/O

The read guard still ran through write_needle_blob_and_index, the tail
receiver write, the fetch-and-write local write, and the HTTP write and
DELETE fallbacks, so a slow append blocked mount/unmount writers (and,
on writer-preferring locks, readers on other volumes). Resolve the
Arc<Volume> under a short guard and write through the handle after it
drops — the same scoping process_batch already used.

* volume: serialize the read-only mark and its .vif save under the lease

set_read_only_persist only leased the idx-attach branch, so concurrent
mark-readonly and mark-writable calls could interleave flag updates and
.vif writes — including two save_vif calls truncating and writing the
file at once. The lease now spans the whole flag change and save, with
a _held variant for mark_io_unavailable, which reaches this from write
paths already holding the lease (Go: markVolumeReadonly runs under
dataFileAccessLock).

* volume,server: span the tier-down transition under one write lease

update_remote_files cleared the remote-only write restriction and
released its lease before open_local_dat_backend attached the local
file, so a write could slip into the gap — pass check_writable, then
fail appending to a missing dat_file. The handler now holds the lease
across the reference trim, .vif save, and backend swap (and the rollback
re-insert), via _held variants; the store read guard is scoped to the
handle lookup. The swap still lands after the .vif trim — the ordering
that keeps a crash from pointing the volume at a deleted remote object.

* server: run a batch's volume groups on separate scoped threads

A single batch can hold writes for several volumes; running the groups
sequentially made every later volume wait behind a stalled append on an
earlier one — the same head-of-line blocking the store guard release
fixed, one level up. Each group now gets a scoped thread, preserving
order within a group while independent volumes progress in parallel.

* storage: close a volume handle a mount slot replaces

With shared handles, inserting a new Volume over an existing map entry
left the old handle alive in callers — two Volume instances could then
write the same files with separate in-memory indexes. set_volume and
create_volume now close() the replaced handle so its writers fail
closed, matching what unload_volume already does.

* test: key the write hook off a reserved needle id

The hook registry is process-global and keyed by needle id, and other
parallel tests write NeedleId(1) — an unrelated write could enter the
barriers and leave the test's real writer parked forever.

* server: wait out readers for vacuum commit without the store guard

commit_compact try-locked the write lease because callers held the store
write lock, returning COMMIT_WAITS_FOR_READS while a download was in
flight — and the master reads that as a failed vacuum, discarding
completed compaction work. The commit RPC now resolves the handle under
a brief store read and blocks on the volume's lease, matching Go's
CommitCompact under dataFileAccessLock: a slow reader stalls only this
volume's vacuum. The try-lock path stays for callers that still hold the
store lock.

* server: scope the remaining store read guards to volume lookup

Same class as the write-path fix: the batch-delete cookie check and
tombstone, the query read, volume_needle_status, and the local chunk
read all kept the read guard through the needle operation, so slow I/O
there could still stall store writers. Each now resolves the Arc<Volume>
under a short guard; the EC rechecks keep their guards.

* seaweed-volume: fail closed on detached volume handles

A handle resolved under a brief store read can outlive its mount: an
unmount+remount between lookup and lease leaves the Arc pointing at a
closed Volume while a new instance owns the files. Mark the handle in
close() and reject it under the write lease in commit_compact_held,
delete_needle, check_writable, set_read_only_persist and
update_remote_files_held; the tier-down handler also checks the flag
before deleting a remote object its replacement still reads.

* seaweed-volume: drop the store guard before mark-readonly waits on the lease

mark_volume_readonly held a store read across set_read_only_persist, which waits out the volume's leased readers; a slow download would stall every lifecycle operation behind the guard. Resolve the handle under a short read, then mark off the guard; a detached handle now maps to not_found.

* seaweed-volume: retry batch deletes that raced an EC mount

The store guard now drops before delete_needle, so ec.encode can mount shards between the pre-delete recheck and the tombstone; the encode then removes the .dat and the needle resurrects from pre-tombstone shards. Re-check has_ec_volume after each successful delete and return the retriable 503 so the filer requeues onto the EC path.

* seaweed-volume: one write-queue worker per volume

A stalled batch kept later batches for other volumes queued behind it: the single worker awaited process_batch before receiving again. Go's volume_write.go keeps one batching worker per volume, so give each volume its own FIFO worker spawned lazily, bounded by a global concurrent grouped-I/O semaphore and a global queued-request bound; workers exit only when their channel closes, and submit refuses unknown volume ids before spawning.

* seaweed-volume: move the queued-write permit inside the request

submit held the global queued-slot permit across the send and the response wait, so a canceled submit freed the slot while the worker still held the needle; repeated cancels could overflow the bound. The request now owns the permit until the worker drops it.

* seaweed-volume: retire write workers whose volume left the store

worker_for kept a sender per volume forever, so servers that mount and retire volumes accumulated idle workers. A worker now polls while idle and retires once its volume leaves the store: close the channel so racing senders fail fast, drain accepted writes, then release the map slot — a replacement worker can never start while this one still holds queued writes, so same-volume FIFO survives retirement.

seaweed-volume/src/server/grpc_server.rs

@@ -302,12 +302,6 @@ impl VolumeGrpcService {
             e.into_inner()
         })
     }
-    fn store_write(&self) -> std::sync::RwLockWriteGuard<'_, crate::storage::store::Store> {
-        self.state.store.write().unwrap_or_else(|e| {
-            tracing::warn!("recovering poisoned store write lock");
-            e.into_inner()
-        })
-    }
     /// Verifies the gRPC caller is allowed to invoke a destructive admin
     /// operation. Mirrors the Go side's checkGrpcAdminAuth: an empty
     /// whitelist accepts everyone (insecure-by-default for tests and
@@ -419,8 +413,8 @@ impl VolumeGrpcService {
             MasterVolumeInfo {
                 volume_id: vid,
                 collection: vol.collection.clone(),
-                replica_placement: vol.super_block.replica_placement.to_byte(),
-                ttl: vol.super_block.ttl.to_u32(),
+                replica_placement: vol.super_block().replica_placement.to_byte(),
+                ttl: vol.super_block().ttl.to_u32(),
                 disk_type: store.locations[loc_idx].disk_type.to_string(),
                 ip: store.ip.clone(),
                 port: store.port,
@@ -432,15 +426,23 @@ impl VolumeGrpcService {
 
         // Step 2: mark local volume readonly; Go's MarkVolumeReadonly errors
         // when the volume left the store during the step-1 master round trip.
-        {
-            let mut store = self.state.store.write().unwrap();
-            let (_, vol) = store
-                .find_volume_mut(vid)
-                .ok_or_else(|| Status::not_found(format!("volume {} not found", vid)))?;
-            vol.set_read_only_persist(can_delete, persist)
-                .map_err(|e| Status::internal(e.to_string()))?;
-            self.state.volume_state_notify.notify_one();
-        }
+        let vol = {
+            let store = self.state.store.read().unwrap();
+            store
+                .find_volume(vid)
+                .map(|(_, v)| v)
+                .ok_or_else(|| Status::not_found(format!("volume {} not found", vid)))?
+        };
+        // set_read_only_persist waits out the volume's leased readers; that
+        // wait must not hold a store guard. A detached handle fails closed.
+        vol.set_read_only_persist(can_delete, persist)
+            .map_err(|e| match e {
+                crate::storage::volume::VolumeError::NotFound => {
+                    Status::not_found(format!("volume {} not found", vid))
+                }
+                _ => Status::internal(e.to_string()),
+            })?;
+        self.state.volume_state_notify.notify_one();
 
         // Step 3: notify master again to cover heartbeat race
         self.notify_master_volume_readonly(&info, true).await?;
@@ -1008,13 +1010,15 @@ impl VolumeServer for VolumeGrpcService {
             if !req.skip_cookie_check || is_ec_volume {
                 let original_cookie = n.cookie;
                 if !is_ec_volume {
-                    let store = self.state.store.read().unwrap();
                     let mut read_option = crate::server::handlers::delete_read_option();
-                    match store.read_volume_needle_with_option(
-                        file_id.volume_id,
-                        &mut n,
-                        &mut read_option,
-                    ) {
+                    let vol = {
+                        let store = self.state.store.read().unwrap();
+                        store.find_volume(file_id.volume_id).map(|(_, v)| v)
+                    };
+                    match vol
+                        .ok_or(crate::storage::volume::VolumeError::NotFound)
+                        .and_then(|vol| vol.read_needle_with_option(&mut n, &mut read_option))
+                    {
                         Ok(_) => {}
                         Err(e) => {
                             results.push(volume_server_pb::DeleteResult {
@@ -1101,30 +1105,67 @@ impl VolumeServer for VolumeGrpcService {
             n.last_modified = now;
 
             if !is_ec_volume {
-                let mut store = self.state.store.write().unwrap();
-                // Recheck EC state under the write lock before mutating the .dat. The
-                // is_ec_volume snapshot was taken earlier under a separate read lock;
-                // ec.encode mounts EC shards (copied from the .dat) BEFORE deleting the
-                // originals, so the vid can be EC now while the .dat still exists. A
-                // delete_volume_needle here would tombstone that .dat, which the encode
-                // then removes — the delete is lost and the needle resurrected from the
-                // pre-tombstone shards. If the vid is now EC, return a retriable 503 so
-                // the filer requeues the delete onto the EC path.
-                if store.has_ec_volume(file_id.volume_id) {
-                    results.push(volume_server_pb::DeleteResult {
-                        file_id: fid_str.clone(),
-                        status: 503,
-                        error: format!(
-                            "volume {} became ec during delete, try again",
-                            file_id.volume_id
-                        ),
-                        size: 0,
-                        version: 0,
-                    });
-                    continue;
-                }
-                match store.delete_volume_needle(file_id.volume_id, &mut n) {
+                let vol = {
+                    let store = self.state.store.read().unwrap();
+                    // Recheck EC state under the store lock before mutating the .dat. The
+                    // is_ec_volume snapshot was taken earlier under a separate read lock;
+                    // ec.encode mounts EC shards (copied from the .dat) BEFORE deleting the
+                    // originals, so the vid can be EC now while the .dat still exists. A
+                    // delete_volume_needle here would tombstone that .dat, which the encode
+                    // then removes — the delete is lost and the needle resurrected from the
+                    // pre-tombstone shards. If the vid is now EC, return a retriable 503 so
+                    // the filer requeues the delete onto the EC path.
+                    if store.has_ec_volume(file_id.volume_id) {
+                        results.push(volume_server_pb::DeleteResult {
+                            file_id: fid_str.clone(),
+                            status: 503,
+                            error: format!(
+                                "volume {} became ec during delete, try again",
+                                file_id.volume_id
+                            ),
+                            size: 0,
+                            version: 0,
+                        });
+                        continue;
+                    }
+                    store.find_volume(file_id.volume_id).map(|(_, v)| v)
+                };
+                // Matches Store::delete_volume_needle: noWriteOrDelete rejects
+                // first, and the tombstone I/O runs after the store guard drops.
+                let delete_result = match vol {
+                    None => Err(crate::storage::volume::VolumeError::NotFound),
+                    Some(vol) if vol.is_no_write_or_delete() => {
+                        Err(crate::storage::volume::VolumeError::ReadOnly(file_id.volume_id))
+                    }
+                    Some(vol) => vol.delete_needle(&mut n),
+                };
+                match delete_result {
                     Ok(size) => {
+                        // The guard dropped before delete_needle, so ec.encode
+                        // could have mounted EC shards between the recheck and
+                        // the tombstone: the encode would then remove the .dat
+                        // this delete touched and resurrect the needle from
+                        // pre-tombstone shards. If the vid is EC now, return
+                        // the retriable 503 so the filer requeues onto the EC
+                        // path — the same check a delete under the old store
+                        // lock could not race.
+                        let now_ec = {
+                            let store = self.state.store.read().unwrap();
+                            store.has_ec_volume(file_id.volume_id)
+                        };
+                        if now_ec {
+                            results.push(volume_server_pb::DeleteResult {
+                                file_id: fid_str.clone(),
+                                status: 503,
+                                error: format!(
+                                    "volume {} became ec during delete, try again",
+                                    file_id.volume_id
+                                ),
+                                size: 0,
+                                version: 0,
+                            });
+                            continue;
+                        }
                         if size.0 == 0 {
                             results.push(volume_server_pb::DeleteResult {
                                 file_id: fid_str.clone(),
@@ -1151,9 +1192,10 @@ impl VolumeServer for VolumeGrpcService {
                         // "try again" token so the filer requeues it and the retry hits
                         // the EC path. A bare exact "not found" would be dropped
                         // permanently by the filer chunk-GC.
-                        if matches!(e, crate::storage::volume::VolumeError::NotFound)
-                            && store.has_ec_volume(file_id.volume_id)
-                        {
+                        if matches!(e, crate::storage::volume::VolumeError::NotFound) && {
+                            let store = self.state.store.read().unwrap();
+                            store.has_ec_volume(file_id.volume_id)
+                        } {
                             results.push(volume_server_pb::DeleteResult {
                                 file_id: fid_str.clone(),
                                 status: 503,
@@ -1300,8 +1342,24 @@ impl VolumeServer for VolumeGrpcService {
         }
 
         let commit_start = std::time::Instant::now();
-        let mut store = self.state.store.write().unwrap();
-        let result = store.commit_compact_volume(vid);
+        let result = {
+            let vol = {
+                let store = self.state.store.read().unwrap();
+                store.find_volume(vid).map(|(_, vol)| vol)
+            };
+            match vol {
+                None => Err(crate::storage::volume::VolumeError::VolumeNotFound(vid)),
+                Some(vol) => {
+                    // Go's CommitCompact waits out leased readers under
+                    // dataFileAccessLock — do it without holding the store
+                    // guard, so a slow download stalls only this vacuum.
+                    let _lease = vol.instance().write_lock();
+                    let is_read_only = vol.is_read_only();
+                    vol.commit_compact_held()
+                        .map(|_| (is_read_only, vol.dat_file_size().unwrap_or(0)))
+                }
+            }
+        };
         crate::metrics::VACUUMING_HISTOGRAM
             .with_label_values(&["commit"])
             .observe(commit_start.elapsed().as_secs_f64());
@@ -1419,10 +1477,10 @@ impl VolumeServer for VolumeGrpcService {
         Ok(Response::new(volume_server_pb::VolumeSyncStatusResponse {
             volume_id: vid.0,
             collection: vol.collection.clone(),
-            replication: vol.super_block.replica_placement.to_string(),
-            ttl: vol.super_block.ttl.to_string(),
+            replication: vol.super_block().replica_placement.to_string(),
+            ttl: vol.super_block().ttl.to_string(),
             tail_offset: vol.dat_file_size().unwrap_or(0),
-            compact_revision: vol.super_block.compaction_revision as u32,
+            compact_revision: vol.super_block().compaction_revision as u32,
             idx_file_size: vol.idx_file_size(),
             version: vol.version().0 as u32,
         }))
@@ -1438,8 +1496,8 @@ impl VolumeServer for VolumeGrpcService {
 
         // A quarantined volume is rejected before the sync does disk I/O for it.
         {
-            let mut store = self.state.store.write().unwrap();
-            let Some((_, v)) = store.find_volume_mut(vid) else {
+            let store = self.state.store.read().unwrap();
+            let Some((_, v)) = store.find_volume(vid) else {
                 return Err(Status::not_found(format!("not found volume id {}", vid)));
             };
             if let Some(e) = v.unavailable_error() {
@@ -1457,7 +1515,7 @@ impl VolumeServer for VolumeGrpcService {
         }
 
         let dat_size = v.dat_file_size().unwrap_or(0);
-        let super_block_size = v.super_block.block_size() as u64;
+        let super_block_size = v.super_block().block_size() as u64;
 
         // If since_ns is very large (after all data), return empty
         if req.since_ns == u64::MAX || dat_size <= super_block_size {
@@ -1757,8 +1815,8 @@ impl VolumeServer for VolumeGrpcService {
             MasterVolumeInfo {
                 volume_id: vid,
                 collection: vol.collection.clone(),
-                replica_placement: vol.super_block.replica_placement.to_byte(),
-                ttl: vol.super_block.ttl.to_u32(),
+                replica_placement: vol.super_block().replica_placement.to_byte(),
+                ttl: vol.super_block().ttl.to_u32(),
                 disk_type: store.locations[loc_idx].disk_type.to_string(),
                 ip: store.ip.clone(),
                 port: store.port,
@@ -1769,9 +1827,9 @@ impl VolumeServer for VolumeGrpcService {
 
         // Step 1: mark local volume as writable (save result; Go continues on error)
         let mark_result = {
-            let mut store = self.state.store.write().unwrap();
+            let store = self.state.store.read().unwrap();
             let res = store
-                .find_volume_mut(vid)
+                .find_volume(vid)
                 .ok_or_else(|| Status::not_found(format!("volume {} not found", vid)))
                 .and_then(|(_, vol)| {
                     vol.set_writable()
@@ -2027,7 +2085,7 @@ impl VolumeServer for VolumeGrpcService {
                 dat_file_timestamp_seconds: mod_time,
                 dat_file_size: vol.dat_file_size().unwrap_or(0),
                 file_count: vol.file_count() as u64,
-                compaction_revision: vol.super_block.compaction_revision as u32,
+                compaction_revision: vol.super_block().compaction_revision as u32,
                 collection: vol.collection.clone(),
                 disk_type: store.locations[loc_idx].disk_type.to_string(),
                 volume_info: Some(vol.volume_info().clone()),
@@ -2052,8 +2110,8 @@ impl VolumeServer for VolumeGrpcService {
         if !req.is_ec_volume {
             // Sync volume to disk before copying (matching Go's v.SyncToDisk())
             {
-                let mut store = self.state.store.write().unwrap();
-                if let Some((_, v)) = store.find_volume_mut(vid) {
+                let store = self.state.store.read().unwrap();
+                if let Some((_, v)) = store.find_volume(vid) {
                     let _ = v.sync_to_disk();
                 }
             }
@@ -2072,7 +2130,7 @@ impl VolumeServer for VolumeGrpcService {
             // must not be used here, or this check spuriously fails on every volume
             // that has ever been compacted even once.
             if req.compaction_revision != u32::MAX
-                && u32::from(v.super_block.compaction_revision) != req.compaction_revision
+                && u32::from(v.super_block().compaction_revision) != req.compaction_revision
             {
                 return Err(Status::failed_precondition(format!(
                     "volume {} is compacted",
@@ -2558,10 +2616,13 @@ impl VolumeServer for VolumeGrpcService {
             return Err(Status::invalid_argument(msg));
         }
 
-        let mut store = self.store_write();
-        let (_, vol) = store
-            .find_volume_mut(vid)
-            .ok_or_else(|| Status::not_found(format!("not found volume id {}", vid)))?;
+        let vol = {
+            let store = self.store_read();
+            let (_, vol) = store
+                .find_volume(vid)
+                .ok_or_else(|| Status::not_found(format!("not found volume id {}", vid)))?;
+            vol
+        };
 
         vol.write_needle_blob_and_index(needle_id, &req.needle_blob, size)
             .map_err(|e| {
@@ -2615,7 +2676,10 @@ impl VolumeServer for VolumeGrpcService {
             let (_, vol) = store
                 .find_volume(vid)
                 .ok_or_else(|| Status::not_found(format!("not found volume id {}", vid)))?;
-            (vol.version().0 as u32, vol.super_block.block_size() as u64)
+            (
+                vol.version().0 as u32,
+                vol.super_block().block_size() as u64,
+            )
         };
 
         let state = self.state.clone();
@@ -2832,9 +2896,11 @@ impl VolumeServer for VolumeGrpcService {
             }
 
             // Write needle to local volume
-            let mut store = state.store.write().unwrap();
-            store
-                .write_volume_needle(vid, &mut n, false)
+            let vol = {
+                let store = state.store.read().unwrap();
+                store.writable_volume(vid)
+            };
+            vol.and_then(|vol| vol.write_needle(&mut n, true, false))
                 .map_err(|e| Status::internal(format!("write needle: {}", e)))?;
         }
 
@@ -4245,7 +4311,7 @@ impl VolumeServer for VolumeGrpcService {
             (
                 dat_path,
                 vol.instance(),
-                vol.super_block.compaction_revision,
+                vol.super_block().compaction_revision,
             )
         };
 
@@ -4348,12 +4414,12 @@ impl VolumeServer for VolumeGrpcService {
                 // volume was compacted, replaced or removed mid-upload: the
                 // object would not match what now holds this id.
                 let abort = {
-                    let mut store = state.store.write().unwrap();
+                    let store = state.store.read().unwrap();
                     let abort = match store.find_volume(vid) {
                         None => Some(Status::not_found(format!("volume {} not found", vid))),
                         Some((_, v))
                             if !v.is_instance(&instance)
-                                || v.super_block.compaction_revision != compaction_revision =>
+                                || v.super_block().compaction_revision != compaction_revision =>
                         {
                             Some(Status::failed_precondition(format!(
                                 "volume {} was compacted or replaced during tier move to remote",
@@ -4363,7 +4429,7 @@ impl VolumeServer for VolumeGrpcService {
                         Some(_) => None,
                     };
                     if abort.is_none()
-                        && let Some((_, vol)) = store.find_volume_mut(vid)
+                        && let Some((_, vol)) = store.find_volume(vid)
                     {
                         vol.update_remote_files(|files| {
                             files.push(volume_server_pb::RemoteFile {
@@ -4631,10 +4697,16 @@ impl VolumeServer for VolumeGrpcService {
                 // before the delete only leaks the remote object; the .vif must never
                 // reference an object that has already been deleted.
                 {
-                    let mut store = state.store.write().unwrap();
-                    let (_, vol) = store.find_volume_mut(vid).ok_or_else(|| {
-                        Status::not_found(format!("volume {} disappeared during tier-down", vid))
-                    })?;
+                    let vol = {
+                        let store = state.store.read().unwrap();
+                        let (_, vol) = store.find_volume(vid).ok_or_else(|| {
+                            Status::not_found(format!(
+                                "volume {} disappeared during tier-down",
+                                vid
+                            ))
+                        })?;
+                        vol
+                    };
 
                     // The volume could have been deleted/recreated or re-tiered while the
                     // download ran. Abort rather than trim the .vif or delete a remote
@@ -4647,6 +4719,24 @@ impl VolumeServer for VolumeGrpcService {
                         )));
                     }
 
+                    // One write lease across the reference trim, .vif save and
+                    // backend swap: between clearing the remote-only write
+                    // restriction and attaching the local .dat a write would
+                    // pass check_writable and then fail on a missing dat_file.
+                    let _lease = vol.instance().write_lock();
+
+                    // The store guard dropped before the lease was taken, so an
+                    // unmount could have detached this handle in the gap. A
+                    // closed handle still passes the remote-reference check,
+                    // and proceeding would delete an object its remounted
+                    // replacement still reads.
+                    if vol.is_closed() {
+                        return Err(Status::not_found(format!(
+                            "volume {} was unmounted during tier-down",
+                            vid
+                        )));
+                    }
+
                     // Snapshot the remote reference before dropping it: the
                     // refresh it triggers can fail, and a half-applied transition
                     // leaves the volume claiming local while the remote backend
@@ -4659,7 +4749,7 @@ impl VolumeServer for VolumeGrpcService {
                     // it the first write would append to the local .dat and
                     // then fail to index.
                     let mut removed_remote = None;
-                    if let Err(e) = vol.update_remote_files(|files| {
+                    if let Err(e) = vol.update_remote_files_held(|files| {
                         if !files.is_empty() {
                             removed_remote = Some(files.remove(0));
                         }
@@ -4668,7 +4758,7 @@ impl VolumeServer for VolumeGrpcService {
                         // map back where they belong. Best effort: if even this
                         // fails the volume stays pinned read-only, which is the
                         // safe end of the transition.
-                        if let Err(restore_err) = vol.update_remote_files(|files| {
+                        if let Err(restore_err) = vol.update_remote_files_held(|files| {
                             if let Some(remote) = removed_remote {
                                 files.insert(0, remote);
                             }
@@ -4692,7 +4782,7 @@ impl VolumeServer for VolumeGrpcService {
                         )));
                     }
 
-                    if let Err(e) = vol.open_local_dat_backend() {
+                    if let Err(e) = vol.open_local_dat_backend_held() {
                         return Err(Status::internal(format!(
                             "volume {} failed to open local dat file {}: {}",
                             vid, dat_path, e
@@ -4914,9 +5004,11 @@ impl VolumeServer for VolumeGrpcService {
         let needle_id = req.needle_id;
         let size = req.size;
         let local_handle = tokio::task::spawn_blocking(move || {
-            let mut store = state_clone.store.write().unwrap();
-            store
-                .write_volume_needle(vid, &mut n_clone, false)
+            let vol = {
+                let store = state_clone.store.read().unwrap();
+                store.writable_volume(vid)
+            };
+            vol.and_then(|vol| vol.write_needle(&mut n_clone, true, false))
                 .map(|_| ())
                 .map_err(|e| format!("local write needle {} size {}: {}", needle_id, size, e))
         });
@@ -5079,11 +5171,13 @@ impl VolumeServer for VolumeGrpcService {
             };
             let original_cookie = n.cookie;
 
-            let store = self.state.store.read().unwrap();
-            store
-                .read_volume_needle(file_id.volume_id, &mut n)
+            let vol = {
+                let store = self.state.store.read().unwrap();
+                store.find_volume(file_id.volume_id).map(|(_, v)| v)
+            };
+            vol.ok_or(crate::storage::volume::VolumeError::NotFound)
+                .and_then(|vol| vol.read_needle(&mut n))
                 .map_err(|e| Status::internal(e.to_string()))?;
-            drop(store);
 
             // Cookie mismatch: log and return empty stream (matching Go behavior where err is nil)
             if n.cookie != original_cookie {
@@ -5224,15 +5318,17 @@ impl VolumeServer for VolumeGrpcService {
         let needle_id = NeedleId(req.needle_id);
 
         {
-            let store = self.state.store.read().unwrap();
-
             // Try normal volume first
-            if store.find_volume(vid).is_some() {
+            let vol = {
+                let store = self.state.store.read().unwrap();
+                store.find_volume(vid).map(|(_, v)| v)
+            };
+            if let Some(vol) = vol {
                 let mut n = Needle {
                     id: needle_id,
                     ..Needle::default()
                 };
-                match store.read_volume_needle(vid, &mut n) {
+                match vol.read_needle(&mut n) {
                     Ok(_) => {
                         let ttl_str = n.ttl.as_ref().map_or(String::new(), |t| t.to_string());
                         return Ok(Response::new(
@@ -6636,7 +6732,7 @@ fn read_all_needles_of_volume(
     let mut plan = {
         let store = state.store.read().unwrap();
         let (_, v) = store.find_volume(vid).ok_or_else(not_found)?;
-        v.dat_scan_plan(v.super_block.block_size() as u64)
+        v.dat_scan_plan(v.super_block().block_size() as u64)
             .map_err(internal)?
     };
     loop {
@@ -7271,7 +7367,7 @@ mod tests {
         let dir = tmp.path().to_str().unwrap();
 
         let (dat_bytes, super_block_size) = {
-            let mut volume = crate::storage::volume::Volume::new(
+            let volume = crate::storage::volume::Volume::new(
                 dir,
                 dir,
                 VolumeId(1),
@@ -7290,7 +7386,7 @@ mod tests {
             volume.sync_to_disk().unwrap();
             (
                 std::fs::read(volume.file_name(".dat")).unwrap(),
-                volume.super_block.block_size() as u64,
+                volume.super_block().block_size() as u64,
             )
         };
 
@@ -7455,7 +7551,7 @@ mod tests {
             )
             .unwrap();
         {
-            let (_, volume) = store.find_volume_mut(VolumeId(1)).unwrap();
+            let (_, volume) = store.find_volume(VolumeId(1)).unwrap();
             let mut needle = Needle {
                 id: NeedleId(11),
                 cookie: Cookie(0x3344),
@@ -7813,7 +7909,7 @@ mod tests {
         let revision_before = {
             let store = service.state.store.read().unwrap();
             let (_, v) = store.find_volume(VolumeId(1)).unwrap();
-            v.super_block.compaction_revision
+            v.super_block().compaction_revision
         };
         let state = service.state.clone();
         let held = state
@@ -7852,7 +7948,7 @@ mod tests {
         let revision_after = {
             let store = state.store.read().unwrap();
             let (_, v) = store.find_volume(VolumeId(1)).unwrap();
-            v.super_block.compaction_revision
+            v.super_block().compaction_revision
         };
         assert_eq!(
             revision_before, revision_after,
@@ -8275,7 +8371,7 @@ mod tests {
 
     /// Writes needles 1 and 2 and deletes 1, so a compaction moves needle 2.
     fn seed_compactable_volume(service: &VolumeGrpcService) {
-        let mut store = service.state.store.write().unwrap();
+        let store = service.state.store.read().unwrap();
         for id in [1u64, 2] {
             let data = format!("needle-{id}").into_bytes();
             let mut n = Needle {
@@ -8509,7 +8605,7 @@ mod tests {
                 .add_volume(VolumeId(1), DiskType::HardDrive, &VolumeSpec::default())
                 .unwrap();
             let (_, vol) = store.find_volume(VolumeId(1)).unwrap();
-            assert_eq!(vol.super_block.compaction_revision, 0);
+            assert_eq!(vol.super_block().compaction_revision, 0);
         }
         s3.release.add_permits(1);
 
@@ -8597,8 +8693,8 @@ mod tests {
         let (endpoint, shutdown_tx, _deletes) = spawn_fake_s3_server(dat_bytes.clone());
         register_tier_backend("s3.tier_up_then_commit", endpoint);
         {
-            let mut store = service.state.store.write().unwrap();
-            let (_, vol) = store.find_volume_mut(VolumeId(1)).unwrap();
+            let store = service.state.store.read().unwrap();
+            let (_, vol) = store.find_volume(VolumeId(1)).unwrap();
             vol.update_remote_files(|files| {
                 files.push(volume_server_pb::RemoteFile {
                     backend_type: "s3".to_string(),
@@ -8646,8 +8742,8 @@ mod tests {
         let (endpoint, shutdown_tx, _deletes) = spawn_fake_s3_server(dat_bytes.clone());
         register_tier_backend("s3.tiered_then_compact", endpoint);
         {
-            let mut store = service.state.store.write().unwrap();
-            let (_, vol) = store.find_volume_mut(VolumeId(1)).unwrap();
+            let store = service.state.store.read().unwrap();
+            let (_, vol) = store.find_volume(VolumeId(1)).unwrap();
             vol.update_remote_files(|files| {
                 files.push(volume_server_pb::RemoteFile {
                     backend_type: "s3".to_string(),
@@ -8704,8 +8800,8 @@ mod tests {
     fn make_local_service_with_large_volume() -> (VolumeGrpcService, TempDir, Vec<u8>) {
         let (service, tmp) = make_local_service_with_volume("", None);
         {
-            let mut store = service.state.store.write().unwrap();
-            let (_, v) = store.find_volume_mut(VolumeId(1)).unwrap();
+            let store = service.state.store.read().unwrap();
+            let (_, v) = store.find_volume(VolumeId(1)).unwrap();
             // Non-uniform payload so reassembly catches duplicated or reordered
             // chunks that a repeated byte would hide.
             let payload: Vec<u8> = (0..5 * 1024 * 1024usize)
@@ -9483,7 +9579,7 @@ mod tests {
         let sb_size = {
             let store = service.state.store.read().unwrap();
             let (_, v) = store.find_volume(VolumeId(1)).unwrap();
-            v.super_block.block_size()
+            v.super_block().block_size()
         };
 
         let response = service
@@ -9542,8 +9638,8 @@ mod tests {
     }
 
     fn write_volume_needle(service: &VolumeGrpcService, id: u64, data: &[u8]) {
-        let mut store = service.state.store.write().unwrap();
-        let (_, v) = store.find_volume_mut(VolumeId(1)).unwrap();
+        let store = service.state.store.read().unwrap();
+        let (_, v) = store.find_volume(VolumeId(1)).unwrap();
         let mut n = Needle {
             id: NeedleId(id),
             cookie: Cookie(0x1234),
@@ -9688,7 +9784,7 @@ mod tests {
     async fn park_read_all_then_write(
         service: &VolumeGrpcService,
         ids: &[u64],
-        f: impl FnOnce(&mut crate::storage::volume::Volume),
+        f: impl FnOnce(&crate::storage::volume::Volume),
     ) -> (
         volume_server_pb::ReadAllNeedlesResponse,
         BoxStream<volume_server_pb::ReadAllNeedlesResponse>,
@@ -9701,9 +9797,9 @@ mod tests {
         let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
         loop {
             // The scan takes a read guard briefly per record, so retry.
-            if let Ok(mut store) = service.state.store.try_write() {
-                let (_, v) = store.find_volume_mut(VolumeId(1)).unwrap();
-                f(v);
+            if let Ok(store) = service.state.store.try_write() {
+                let (_, v) = store.find_volume(VolumeId(1)).unwrap();
+                f(&v);
                 return (first, stream);
             }
             if std::time::Instant::now() > deadline {
@@ -9847,7 +9943,7 @@ mod tests {
         let super_block_size = {
             let store = service.state.store.read().unwrap();
             let (_, v) = store.find_volume(VolumeId(1)).unwrap();
-            v.super_block.block_size() as u64
+            v.super_block().block_size() as u64
         };
 
         let response = service
@@ -10205,7 +10301,7 @@ mod tests {
         let (service, _tmp) = make_local_service_with_volume("", None);
         let vid = VolumeId(1);
         let cpd = {
-            let mut store = service.state.store.write().unwrap();
+            let store = service.state.store.read().unwrap();
             for i in 100..110u64 {
                 let data = format!("data-{i}");
                 let mut n = Needle {
@@ -10261,7 +10357,7 @@ mod tests {
         {
             let store = service.state.store.read().unwrap();
             let (_, v) = store.find_volume(vid).expect("still mounted");
-            assert_eq!(v.super_block.replica_placement.to_string(), "000");
+            assert_eq!(v.super_block().replica_placement.to_string(), "000");
         }
 
         job.run(|_| true).unwrap();
@@ -10276,7 +10372,7 @@ mod tests {
         assert_eq!(resp.error, "");
         let store = service.state.store.read().unwrap();
         let (_, v) = store.find_volume(vid).unwrap();
-        assert_eq!(v.super_block.replica_placement.to_string(), "001");
+        assert_eq!(v.super_block().replica_placement.to_string(), "001");
     }
 
     // Regression test for comparing the wrong compaction-revision field.
@@ -10295,16 +10391,16 @@ mod tests {
         // actually diverge the way they do in production, rather than poking
         // the struct fields directly.
         let (revision_before, revision_after, last_compact_revision_after) = {
-            let mut store = service.state.store.write().unwrap();
-            let (_, v) = store.find_volume_mut(VolumeId(1)).unwrap();
-            let before = v.super_block.compaction_revision;
+            let store = service.state.store.read().unwrap();
+            let (_, v) = store.find_volume(VolumeId(1)).unwrap();
+            let before = v.super_block().compaction_revision;
 
             v.compact_by_index(0, 0, |_| true).unwrap();
             v.commit_compact().unwrap();
 
             (
                 before,
-                v.super_block.compaction_revision,
+                v.super_block().compaction_revision,
                 v.last_compact_revision(),
             )
         };
@@ -10429,7 +10525,7 @@ mod tests {
             .store
             .write()
             .unwrap()
-            .find_volume_mut(VolumeId(1))
+            .find_volume(VolumeId(1))
             .unwrap()
             .1
             .set_last_modified_ts_for_test(last_write);
@@ -10438,7 +10534,7 @@ mod tests {
             .store
             .write()
             .unwrap()
-            .find_volume_mut(VolumeId(1))
+            .find_volume(VolumeId(1))
             .unwrap()
             .1
             .set_read_only_persist(false, true)
@@ -11261,7 +11357,7 @@ mod tests {
 
         generate_and_mount_ec_1(&service, (0..14).collect()).await;
         {
-            let mut store = service.state.store.write().unwrap();
+            let store = service.state.store.read().unwrap();
             let mut n = Needle {
                 id: NeedleId(11),
                 cookie: Cookie(0x3344),
@@ -11306,8 +11402,8 @@ mod tests {
     async fn ec_1_with_a_manifest_needle() -> (VolumeGrpcService, TempDir) {
         let (service, tmp) = make_local_service_with_volume("", None);
         {
-            let mut store = service.state.store.write().unwrap();
-            let (_, volume) = store.find_volume_mut(VolumeId(1)).unwrap();
+            let store = service.state.store.read().unwrap();
+            let (_, volume) = store.find_volume(VolumeId(1)).unwrap();
             let mut manifest = Needle {
                 id: NeedleId(12),
                 cookie: Cookie(0x5566),
@@ -12169,7 +12265,7 @@ mod tests {
         // A real volume, really encoded: a parity check needs genuine
         // Reed-Solomon parity to agree with, or "no corruption" means nothing.
         {
-            let mut v = crate::storage::volume::Volume::new(
+            let v = crate::storage::volume::Volume::new(
                 src_s,
                 src_s,
                 vid,
@@ -12386,8 +12482,8 @@ mod tests {
     async fn volume_delete_only_garbage_deletes_a_fully_deleted_volume() {
         let (service, _tmp) = make_local_service_with_volume("", None);
         {
-            let mut store = service.state.store.write().unwrap();
-            let (_, vol) = store.find_volume_mut(VolumeId(1)).unwrap();
+            let store = service.state.store.read().unwrap();
+            let (_, vol) = store.find_volume(VolumeId(1)).unwrap();
             vol.delete_needle(&mut Needle {
                 id: NeedleId(11),
                 cookie: Cookie(0x3344),
@@ -12449,8 +12545,8 @@ mod tests {
         // though the empty check would not.
         let (service, _tmp) = make_local_service_with_volume("", None);
         {
-            let mut store = service.state.store.write().unwrap();
-            let (_, vol) = store.find_volume_mut(VolumeId(1)).unwrap();
+            let store = service.state.store.read().unwrap();
+            let (_, vol) = store.find_volume(VolumeId(1)).unwrap();
             vol.delete_needle(&mut Needle {
                 id: NeedleId(11),
                 cookie: Cookie(0x3344),
@@ -12636,7 +12732,7 @@ mod tests {
             .unwrap();
         let mut size_before_needle_3 = 0;
         {
-            let (_, volume) = store.find_volume_mut(VolumeId(1)).unwrap();
+            let (_, volume) = store.find_volume(VolumeId(1)).unwrap();
             for id in 1..=3u64 {
                 if id == 3 {
                     size_before_needle_3 = volume.dat_file_size().unwrap();
@@ -13243,7 +13339,7 @@ mod tests {
             .store
             .write()
             .unwrap()
-            .find_volume_mut(VolumeId(1))
+            .find_volume(VolumeId(1))
             .unwrap()
             .1
             .set_no_write_or_delete(true);
@@ -13263,7 +13359,7 @@ mod tests {
             .store
             .write()
             .unwrap()
-            .find_volume_mut(VolumeId(1))
+            .find_volume(VolumeId(1))
             .unwrap()
             .1
             .set_read_only_persist(true, false)
@@ -13291,5 +13387,4 @@ mod tests {
             assert_eq!(resp.results[0].error, "not found");
         }
     }
-
 }

seaweed-volume/src/server/handlers.rs

@@ -683,9 +683,9 @@ async fn do_replicated_request(
 
     let copy_count = {
         let store = state.store.read().unwrap();
-        store
-            .find_volume(VolumeId(vid))
-            .map_or(1, |(_, v)| v.super_block.replica_placement.get_copy_count())
+        store.find_volume(VolumeId(vid)).map_or(1, |(_, v)| {
+            v.super_block().replica_placement.get_copy_count()
+        })
     };
     let allow_delete = method == axum::http::Method::DELETE;
     let eligible_locations: Vec<_> = locations
@@ -1286,7 +1286,7 @@ async fn check_download_limit(
                 let store = state.store.read().unwrap();
                 store
                     .find_volume(vid)
-                    .is_some_and(|(_, vol)| vol.super_block.replica_placement.has_replication())
+                    .is_some_and(|(_, vol)| vol.super_block().replica_placement.has_replication())
             };
         if should_try_replica
             && let Some(info) = build_proxy_request_info(path, request_headers, &query_string)
@@ -2956,8 +2956,11 @@ pub async fn post_handler(
     let write_result = if let Some(wq) = state.write_queue.get() {
         wq.submit(vid, n, fsync).await
     } else {
-        let mut store = state.store.write().unwrap();
-        store.write_volume_needle(vid, &mut n, fsync)
+        let vol = {
+            let store = state.store.read().unwrap();
+            store.writable_volume(vid)
+        };
+        vol.and_then(|vol| vol.write_needle(&mut n, true, fsync))
     };
 
     // Replicate to remote volume servers if this volume has replicas.
@@ -2967,7 +2970,7 @@ pub async fn post_handler(
             let store = state.store.read().unwrap();
             store
                 .find_volume(vid)
-                .is_some_and(|(_, v)| v.super_block.replica_placement.get_copy_count() > 1)
+                .is_some_and(|(_, v)| v.super_block().replica_placement.get_copy_count() > 1)
         };
         if needs_replication {
             let state_clone = state.clone();
@@ -3242,8 +3245,14 @@ pub async fn delete_handler(
     // Read needle first to validate cookie (matching Go behavior)
     let original_cookie = cookie;
     {
-        let store = state.store.read().unwrap();
-        match store.read_volume_needle_with_option(vid, &mut n, &mut delete_read_option()) {
+        let vol = {
+            let store = state.store.read().unwrap();
+            store.find_volume(vid).map(|(_, vol)| vol)
+        };
+        match vol
+            .ok_or(crate::storage::volume::VolumeError::NotFound)
+            .and_then(|vol| vol.read_needle_with_option(&mut n, &mut delete_read_option()))
+        {
             Ok(_) => {}
             Err(_) => {
                 let result = DeleteResult { size: 0 };
@@ -3314,8 +3323,21 @@ pub async fn delete_handler(
     }
 
     let delete_result = {
-        let mut store = state.store.write().unwrap();
-        store.delete_volume_needle(vid, &mut n)
+        let vol = {
+            let store = state.store.read().unwrap();
+            store
+                .find_volume(vid)
+                .map(|(_, vol)| vol)
+                .ok_or(crate::storage::volume::VolumeError::NotFound)
+        };
+        match vol {
+            Err(e) => Err(e),
+            // Matches Store::delete_volume_needle: noWriteOrDelete rejects first.
+            Ok(vol) if vol.is_no_write_or_delete() => {
+                Err(crate::storage::volume::VolumeError::ReadOnly(vid))
+            }
+            Ok(vol) => vol.delete_needle(&mut n),
+        }
     };
 
     let is_replicate = del_query.split('&').any(|p| p == "type=replicate");
@@ -3324,7 +3346,7 @@ pub async fn delete_handler(
             let store = state.store.read().unwrap();
             store
                 .find_volume(vid)
-                .is_some_and(|(_, v)| v.super_block.replica_placement.get_copy_count() > 1)
+                .is_some_and(|(_, v)| v.super_block().replica_placement.get_copy_count() > 1)
         };
         if needs_replication
             && let Err(e) = do_replicated_request(
@@ -3411,7 +3433,7 @@ pub async fn status_handler(
             );
             vol_info.insert(
                 "CompactRevision".to_string(),
-                serde_json::Value::from(vol.super_block.compaction_revision),
+                serde_json::Value::from(vol.super_block().compaction_revision),
             );
             vol_info.insert(
                 "ModifiedAtSecond".to_string(),
@@ -3422,7 +3444,7 @@ pub async fn status_handler(
                 serde_json::Value::from(loc.disk_type.to_string()),
             );
 
-            let replica = &vol.super_block.replica_placement;
+            let replica = &vol.super_block().replica_placement;
             let mut replica_value = serde_json::Map::new();
             if replica.diff_data_center_count > 0 {
                 replica_value.insert(
@@ -3447,7 +3469,7 @@ pub async fn status_handler(
                 serde_json::Value::Object(replica_value),
             );
 
-            let ttl = vol.super_block.ttl;
+            let ttl = vol.super_block().ttl;
             let mut ttl_value = serde_json::Map::new();
             if ttl.count > 0 {
                 ttl_value.insert("Count".to_string(), serde_json::Value::from(ttl.count));
@@ -4114,20 +4136,12 @@ async fn read_chunk_needle(
     enum Placement {
         Ec,
         Remote,
+        Local(std::sync::Arc<crate::storage::volume::Volume>),
     }
     let placement = {
         let store = state.store.read().unwrap();
-        if store.find_volume(vid).is_some() {
-            let mut n = Needle {
-                id: nid,
-                cookie,
-                ..Needle::default()
-            };
-            return store
-                .read_volume_needle(vid, &mut n)
-                .map_err(|e| format!("{}", e))
-                .and_then(|_| cookie_checked_chunk(n, cookie))
-                .map(|data| ChunkBytes { origin: 0, data });
+        if let Some((_, vol)) = store.find_volume(vid) {
+            Placement::Local(vol)
         } else if store.find_ec_volume(vid).is_some() {
             Placement::Ec
         } else {
@@ -4136,6 +4150,17 @@ async fn read_chunk_needle(
     };
 
     match placement {
+        Placement::Local(vol) => {
+            let mut n = Needle {
+                id: nid,
+                cookie,
+                ..Needle::default()
+            };
+            vol.read_needle(&mut n)
+                .map_err(|e| format!("{}", e))
+                .and_then(|_| cookie_checked_chunk(n, cookie))
+                .map(|data| ChunkBytes { origin: 0, data })
+        }
         Placement::Ec => {
             match crate::server::store_ec::read_ec_shard_needle_distributed(state, vid, nid).await {
                 Ok(Some(n)) => {
@@ -5476,7 +5501,7 @@ mod tests {
         use crate::storage::volume::{Volume, VolumeSpec};
 
         let dir = tmp.path().to_str().unwrap();
-        let mut v = Volume::new(
+        let v = Volume::new(
             dir,
             dir,
             VolumeId(2),
@@ -5782,7 +5807,7 @@ mod tests {
         let state = volume_test_state(&tmp);
         let dir = tmp.path().to_str().unwrap();
         let data: Vec<u8> = (0..4096u32).map(|i| (i % 251) as u8).collect();
-        let mut v = Volume::new(
+        let v = Volume::new(
             dir,
             dir,
             VolumeId(2),
@@ -7059,8 +7084,8 @@ mod tests {
         );
 
         {
-            let mut store = state.store.write().unwrap();
-            let (_, vol) = store.find_volume_mut(VolumeId(1)).unwrap();
+            let store = state.store.read().unwrap();
+            let (_, vol) = store.find_volume(VolumeId(1)).unwrap();
             vol.fail_next_fsync_for_test(true);
             vol.fail_next_truncate_for_test(true);
             let mut n = Needle {
@@ -7073,7 +7098,7 @@ mod tests {
             store
                 .write_volume_needle(VolumeId(1), &mut n, true)
                 .unwrap_err();
-            let (_, vol) = store.find_volume_mut(VolumeId(1)).unwrap();
+            let (_, vol) = store.find_volume(VolumeId(1)).unwrap();
             vol.fail_next_fsync_for_test(false);
             vol.fail_next_truncate_for_test(false);
             assert!(vol.unavailable_error().is_some());
@@ -7091,8 +7116,9 @@ mod tests {
         }
     }
 
-    /// A vacuum committed mid-stream must not change the bytes served: the
-    /// stream's handle pins the inode its offset was resolved against.
+    /// A leased stream blocks the compaction commit instead of racing the
+    /// file swap, and still delivers the needle's bytes; the commit goes
+    /// through once the stream releases its read lease.
     #[cfg(unix)]
     #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
     async fn test_streaming_body_survives_a_vacuum_that_moves_the_needle() {
@@ -7138,13 +7164,28 @@ mod tests {
         {
             let mut store = state.store.write().unwrap();
             store.compact_volume(VolumeId(1), 0, 0, |_| true).unwrap();
-            store.commit_compact_volume(VolumeId(1)).unwrap();
+            let err = store.commit_compact_volume(VolumeId(1)).unwrap_err();
+            assert!(
+                err.to_string()
+                    .contains(crate::storage::volume::COMMIT_WAITS_FOR_READS),
+                "the leased stream must hold the commit back: {err}"
+            );
         }
-        assert_ne!(data_offset(), before, "the vacuum must move the needle");
 
         while let Some(chunk) = stream.next().await {
             got.extend_from_slice(&chunk.expect("the stream failed across a vacuum"));
         }
         assert_eq!(got, data);
+        drop(stream);
+
+        {
+            let mut store = state.store.write().unwrap();
+            store.commit_compact_volume(VolumeId(1)).unwrap();
+        }
+        assert_ne!(
+            data_offset(),
+            before,
+            "the vacuum moves the needle once the stream drains"
+        );
     }
 }

seaweed-volume/src/server/heartbeat.rs

@@ -966,7 +966,7 @@ fn apply_volume_actions_to(store: &mut Store, actions: VolumeActions) {
         if let Some(vol) = store
             .locations
             .get_mut(disk_id)
-            .and_then(|loc| loc.find_volume_mut(vid))
+            .and_then(|loc| loc.find_volume(vid))
             && vol.should_quarantine()
         {
             vol.set_no_write_or_delete(true);
@@ -1207,10 +1207,10 @@ fn build_heartbeat_with_ec_status(
                     deleted_byte_count: vol.deleted_size(),
                     read_only: vol.is_read_only(),
                     read_only_can_delete: vol.is_no_write_can_delete(),
-                    replica_placement: vol.super_block.replica_placement.to_byte() as u32,
-                    version: vol.super_block.version.0 as u32,
-                    ttl: vol.super_block.ttl.to_u32(),
-                    compact_revision: vol.super_block.compaction_revision as u32,
+                    replica_placement: vol.super_block().replica_placement.to_byte() as u32,
+                    version: vol.super_block().version.0 as u32,
+                    ttl: vol.super_block().ttl.to_u32(),
+                    compact_revision: vol.super_block().compaction_revision as u32,
                     // The .dat mtime, as Go reports: the shell's quiet-period
                     // gates read this as "last touched", which a delete has to
                     // count towards even though the TTL clock ignores it.
@@ -1878,7 +1878,7 @@ mod tests {
             .store(true, Ordering::Relaxed);
 
         {
-            let (_, volume) = store.find_volume_mut(VolumeId(17)).unwrap();
+            let (_, volume) = store.find_volume(VolumeId(17)).unwrap();
             volume.set_read_only().unwrap();
             volume
                 .update_remote_files(|files| files.push(Default::default()))
@@ -1977,7 +1977,7 @@ mod tests {
             )
             .unwrap();
         {
-            let (_, volume) = store.find_volume_mut(VolumeId(21)).unwrap();
+            let (_, volume) = store.find_volume(VolumeId(21)).unwrap();
             volume.set_read_only().unwrap();
         }
 
@@ -2155,7 +2155,7 @@ mod tests {
             )
             .unwrap();
         let dat_path = {
-            let (_, volume) = store.find_volume_mut(VolumeId(41)).unwrap();
+            let (_, volume) = store.find_volume(VolumeId(41)).unwrap();
             volume.set_last_io_error_for_test(None);
             volume.set_last_modified_ts_for_test(
                 SystemTime::now()
@@ -2209,7 +2209,7 @@ mod tests {
                 },
             )
             .unwrap();
-        let (_, volume) = store.find_volume_mut(VolumeId(51)).unwrap();
+        let (_, volume) = store.find_volume(VolumeId(51)).unwrap();
         volume.set_last_io_error_for_test(Some("input/output error"));
 
         let heartbeat = build_heartbeat(&test_config(), &mut store);
@@ -2219,7 +2219,7 @@ mod tests {
         // advertised to the master.
         assert!(heartbeat.volumes.is_empty());
         assert!(store.has_volume(VolumeId(51)));
-        let (_, volume) = store.find_volume_mut(VolumeId(51)).unwrap();
+        let (_, volume) = store.find_volume(VolumeId(51)).unwrap();
         assert!(volume.is_no_write_or_delete());
     }
 
@@ -2515,7 +2515,7 @@ mod tests {
                     },
                 )
                 .unwrap();
-            let (_, volume) = store.find_volume_mut(VolumeId(id)).unwrap();
+            let (_, volume) = store.find_volume(VolumeId(id)).unwrap();
             volume.set_last_io_error_for_test(None);
             volume.set_last_modified_ts_for_test(now.saturating_sub(60 * 60));
             std::fs::OpenOptions::new()
@@ -2535,7 +2535,7 @@ mod tests {
                 },
             )
             .unwrap();
-        let (_, volume) = store.find_volume_mut(VolumeId(64)).unwrap();
+        let (_, volume) = store.find_volume(VolumeId(64)).unwrap();
         volume.set_last_io_error_for_test(Some("input/output error"));
 
         let (heartbeat, _, actions) =
@@ -2787,7 +2787,7 @@ mod tests {
                 },
             )
             .unwrap();
-        let (_, volume) = store.find_volume_mut(VolumeId(71)).unwrap();
+        let (_, volume) = store.find_volume(VolumeId(71)).unwrap();
         volume
             .update_remote_files(|files| {
                 files.push(crate::storage::volume::PbRemoteFile {

seaweed-volume/src/server/ui.rs

@@ -400,7 +400,7 @@ fn collect_ui_data(
                 file_count: volume.file_count(),
                 delete_count: volume.deleted_count(),
                 deleted_byte_count: volume.deleted_size(),
-                ttl: volume.super_block.ttl.to_string(),
+                ttl: volume.super_block().ttl.to_string(),
                 read_only: volume.is_read_only(),
                 version: volume.version().0 as u32,
                 remote_storage_name,

seaweed-volume/src/server/write_queue.rs

@@ -1,14 +1,18 @@
 //! Async batched write processing for the volume server.
 //!
 //! Instead of each upload handler directly calling `write_needle`, writes are
-//! submitted to a queue. A background worker drains the queue in batches (up to
-//! 128 entries), groups them by volume ID, and processes them together under a
-//! single store lock. Durable writes to a volume share their .dat and .idx
-//! flushes (see `Volume::write_needles_grouped`).
+//! submitted to a queue. Each volume gets its own FIFO worker (Go's
+//! volume_write.go keeps one batching worker per volume), which drains its
+//! channel in batches of up to 128 entries or 4 MiB and shares the .dat/.idx
+//! flushes across durable writes (see `Volume::write_needles_grouped`).
+//! Writes to different volumes proceed in parallel; a volume's ordering and
+//! exclusion still come from its data-file write lease.
 
-use std::sync::Arc;
+use std::collections::HashMap;
+use std::collections::hash_map::Entry;
+use std::sync::{Arc, Mutex};
 
-use tokio::sync::{mpsc, oneshot};
+use tokio::sync::{OwnedSemaphorePermit, Semaphore, mpsc, oneshot};
 use tracing::debug;
 
 use crate::storage::needle::needle::Needle;
@@ -29,6 +33,10 @@ pub struct WriteRequest {
     pub needle: Needle,
     pub fsync: bool,
     pub response_tx: oneshot::Sender<WriteResult>,
+    /// The global queued-slot permit, owned by the request itself: a caller
+    /// that cancels `submit` after the send must not free the slot while the
+    /// worker still holds the needle.
+    pub queue_permit: Option<OwnedSemaphorePermit>,
 }
 
 /// Maximum number of write requests to batch together.
@@ -38,102 +46,205 @@ const MAX_BATCH_SIZE: usize = 128;
 /// This prevents large writes from accumulating unbounded latency.
 const MAX_BATCH_BYTES: usize = 4 * 1024 * 1024;
 
-/// Handle for submitting write requests to the background worker.
+/// Cap on volumes doing grouped I/O at once, so many writers cannot crowd
+/// readers off a slow disk the way unlimited parallel fsyncs would.
+const MAX_CONCURRENT_VOLUME_WRITES: usize = 8;
+
+/// How often an idle worker re-checks that its volume is still mounted.
+/// Workers live for their volume's queue lifetime (Go parity), so the only
+/// retirement is the volume leaving the store.
+const IDLE_WORKER_POLL: std::time::Duration = std::time::Duration::from_secs(60);
+
+/// Handle for submitting write requests to the per-volume workers.
 #[derive(Clone)]
 pub struct WriteQueue {
-    tx: mpsc::Sender<WriteRequest>,
+    state: Arc<VolumeServerState>,
+    /// Per-volume channel capacity; backpressure kicks in when full.
+    capacity: usize,
+    io_permits: Arc<Semaphore>,
+    /// Bounds queued+in-flight writes across all volumes — per-volume channels
+    /// alone would multiply the old single queue's capacity by volume count.
+    queue_permits: Arc<Semaphore>,
+    workers: Arc<Mutex<HashMap<VolumeId, mpsc::Sender<WriteRequest>>>>,
 }
 
 impl WriteQueue {
-    /// Create a new write queue and spawn the background worker.
-    ///
-    /// `capacity` controls the channel buffer size (backpressure kicks in when full).
-    /// The worker holds a reference to `state` for accessing the store.
+    /// Create a new write queue. Per-volume workers are spawned lazily on the
+    /// first submission for that volume.
     pub fn new(state: Arc<VolumeServerState>, capacity: usize) -> Self {
-        let (tx, rx) = mpsc::channel(capacity);
-        let worker = WriteQueueWorker { rx, state };
-        tokio::spawn(worker.run());
-        WriteQueue { tx }
+        WriteQueue {
+            state,
+            capacity,
+            io_permits: Arc::new(Semaphore::new(MAX_CONCURRENT_VOLUME_WRITES)),
+            queue_permits: Arc::new(Semaphore::new(capacity)),
+            workers: Arc::new(Mutex::new(HashMap::new())),
+        }
+    }
+
+    /// Sender for the volume's worker, spawning one on first use.
+    fn worker_for(&self, volume_id: VolumeId) -> mpsc::Sender<WriteRequest> {
+        let mut workers = self.workers.lock().unwrap();
+        match workers.entry(volume_id) {
+            Entry::Occupied(e) => e.get().clone(),
+            Entry::Vacant(e) => {
+                let (tx, rx) = mpsc::channel(self.capacity);
+                tokio::spawn(
+                    VolumeWriteWorker {
+                        rx,
+                        vid: volume_id,
+                        state: self.state.clone(),
+                        io_permits: self.io_permits.clone(),
+                        workers: self.workers.clone(),
+                    }
+                    .run(),
+                );
+                e.insert(tx.clone());
+                tx
+            }
+        }
     }
 
     /// Submit a write request and wait for the result.
     ///
     /// Returns `Err` if the worker has shut down or the response channel was dropped.
     pub async fn submit(&self, volume_id: VolumeId, needle: Needle, fsync: bool) -> WriteResult {
+        // Workers spawn only for mounted volumes — a bogus id never creates
+        // a channel, so volume-id sprays can't grow the worker map.
+        {
+            let store = self.state.store.read().unwrap();
+            if store.find_volume(volume_id).is_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.
+        let queue_permit = self
+            .queue_permits
+            .clone()
+            .acquire_owned()
+            .await
+            .map_err(|_| broken_pipe("write queue shut down"))?;
         let (response_tx, response_rx) = oneshot::channel();
-        let request = WriteRequest {
+        let mut request = WriteRequest {
             volume_id,
             needle,
             fsync,
             response_tx,
+            queue_permit: Some(queue_permit),
         };
 
-        // Send to queue; this awaits if the channel is full (backpressure).
-        if self.tx.send(request).await.is_err() {
-            return Err(VolumeError::Io(std::io::Error::new(
-                std::io::ErrorKind::BrokenPipe,
-                "write queue worker has shut down",
-            )));
+        // A send fails only against a retiring worker's closed channel; the
+        // request was never accepted, so retrying it behind a fresh worker
+        // keeps same-volume order. The volume recheck bounds the spin.
+        loop {
+            let tx = self.worker_for(volume_id);
+            match tx.send(request).await {
+                Ok(()) => break,
+                Err(mpsc::error::SendError(returned)) => {
+                    request = returned;
+                    {
+                        let store = self.state.store.read().unwrap();
+                        if store.find_volume(volume_id).is_none() {
+                            return Err(VolumeError::NotFound);
+                        }
+                    }
+                    tokio::task::yield_now().await;
+                }
+            }
         }
 
         // Wait for the worker to process our request.
         match response_rx.await {
             Ok(result) => result,
-            Err(_) => Err(VolumeError::Io(std::io::Error::new(
-                std::io::ErrorKind::BrokenPipe,
-                "write queue worker dropped response channel",
-            ))),
+            Err(_) => Err(broken_pipe("write queue worker dropped response channel")),
         }
     }
 }
 
-/// Background worker that drains write requests and processes them in batches.
-struct WriteQueueWorker {
-    rx: mpsc::Receiver<WriteRequest>,
-    state: Arc<VolumeServerState>,
+fn broken_pipe(msg: &'static str) -> VolumeError {
+    VolumeError::Io(std::io::Error::new(std::io::ErrorKind::BrokenPipe, msg))
 }
 
-impl WriteQueueWorker {
-    async fn run(mut self) {
-        debug!("write queue worker started");
+/// FIFO worker draining one volume's write channel in batches. Like Go's
+/// per-volume async writer it runs for the volume's queue lifetime — exiting
+/// only when the channel closes or its volume leaves the store — so no
+/// handoff can ever split one volume's write order across two workers.
+struct VolumeWriteWorker {
+    rx: mpsc::Receiver<WriteRequest>,
+    vid: VolumeId,
+    state: Arc<VolumeServerState>,
+    io_permits: Arc<Semaphore>,
+    workers: Arc<Mutex<HashMap<VolumeId, mpsc::Sender<WriteRequest>>>>,
+}
 
+impl VolumeWriteWorker {
+    async fn run(mut self) {
         loop {
-            // Wait for the first request (blocks until one arrives or channel closes).
-            let first = match self.rx.recv().await {
-                Some(req) => req,
-                None => {
-                    debug!("write queue channel closed, worker exiting");
+            let first = match tokio::time::timeout(IDLE_WORKER_POLL, self.rx.recv()).await {
+                Ok(Some(req)) => req,
+                // recv() yields None only once every sender is gone — the map
+                // holds a sender per volume, so that is exactly queue shutdown.
+                Ok(None) => return,
+                Err(_) if self.volume_gone() => {
+                    self.retire().await;
                     return;
                 }
+                Err(_) => continue,
             };
-
-            // Drain as many additional requests as available, up to MAX_BATCH_SIZE
-            // or MAX_BATCH_BYTES (matches Go: 128 requests or 4MB, whichever comes first).
-            let mut batch = Vec::with_capacity(MAX_BATCH_SIZE);
-            let mut batch_bytes: usize = first.needle.data.len();
-            batch.push(first);
-
-            while batch.len() < MAX_BATCH_SIZE && batch_bytes < MAX_BATCH_BYTES {
-                match self.rx.try_recv() {
-                    Ok(req) => {
-                        batch_bytes += req.needle.data.len();
-                        batch.push(req);
-                    }
-                    Err(_) => break,
-                }
-            }
-
-            let batch_size = batch.len();
-            debug!("processing write batch of {} requests", batch_size);
-
-            // Process the batch in spawn_blocking since write_needle does file I/O.
-            let state = self.state.clone();
-            let _ = tokio::task::spawn_blocking(move || {
-                process_batch(state, batch);
-            })
-            .await;
+            self.process_from(first).await;
         }
     }
+
+    /// The volume's handle left the store, so the worker can go too; an idle
+    /// worker for a mounted volume stays (Go keeps the writer for the mount).
+    fn volume_gone(&self) -> bool {
+        let store = self.state.store.read().unwrap();
+        store.find_volume(self.vid).is_none()
+    }
+
+    /// Retire in a fixed order that keeps same-volume FIFO: close the channel
+    /// so racing senders fail fast (they retry and wait for a fresh worker),
+    /// drain what was already accepted, and only then give up the map slot —
+    /// a replacement worker can never start while this one still holds queued
+    /// writes.
+    async fn retire(&mut self) {
+        self.rx.close();
+        while let Some(first) = self.rx.recv().await {
+            self.process_from(first).await;
+        }
+        self.workers.lock().unwrap().remove(&self.vid);
+    }
+
+    /// Collect a batch around `first` and process it under the I/O bound.
+    async fn process_from(&mut self, first: WriteRequest) {
+        // Drain as many additional requests as available, up to MAX_BATCH_SIZE
+        // or MAX_BATCH_BYTES (matches Go: 128 requests or 4MB, whichever comes first).
+        let mut batch = Vec::with_capacity(MAX_BATCH_SIZE);
+        let mut batch_bytes: usize = first.needle.data.len();
+        batch.push(first);
+        while batch.len() < MAX_BATCH_SIZE && batch_bytes < MAX_BATCH_BYTES {
+            match self.rx.try_recv() {
+                Ok(req) => {
+                    batch_bytes += req.needle.data.len();
+                    batch.push(req);
+                }
+                Err(_) => break,
+            }
+        }
+        debug!("processing write batch of {} requests", batch.len());
+
+        // Bound the number of volumes doing grouped I/O at once, then run
+        // the batch in spawn_blocking since write_needle does file I/O.
+        let Ok(_permit) = self.io_permits.clone().acquire_owned().await else {
+            return;
+        };
+        let state = self.state.clone();
+        let _ = tokio::task::spawn_blocking(move || {
+            process_batch(state, batch);
+        })
+        .await;
+    }
 }
 
 /// Process a batch of write requests, grouped by volume ID.
@@ -155,15 +266,28 @@ fn process_batch(state: Arc<VolumeServerState>, batch: Vec<WriteRequest>) {
         }
     }
 
-    // Process each volume group under a single store lock.
-    let mut store = state.store.write().unwrap();
-
+    // Resolve each group's volume handle under a brief store read, then run
+    // 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 results = store.write_volume_needles(vid, &mut writes);
+        let vol = {
+            let store = state.store.read().unwrap();
+            store.writable_volume(vid)
+        };
+        let results = match vol {
+            Ok(vol) => vol.write_needles_grouped(&mut writes),
+            Err(e) => writes
+                .iter()
+                .map(|_| match e {
+                    VolumeError::ReadOnly(vid) => Err(VolumeError::ReadOnly(vid)),
+                    _ => Err(VolumeError::NotFound),
+                })
+                .collect(),
+        };
         for (response_tx, result) in senders.into_iter().zip(results) {
             // Send result back; ignore error if receiver dropped.
             let _ = response_tx.send(result);
@@ -365,6 +489,7 @@ mod tests {
                     },
                     fsync: true,
                     response_tx,
+                    queue_permit: None,
                 }
             })
             .collect();
@@ -399,4 +524,199 @@ mod tests {
         let result = queue2.submit(VolumeId(1), needle, false).await;
         assert!(result.is_err()); // NotFound is fine -- the point is it doesn't panic
     }
+
+    /// Issue #791: a write parked inside its .dat append must not hold the
+    /// store guard, so another volume still resolves and writes. The hook
+    /// runs under the volume's data-file lease, so the assertion is that the
+    /// store — not the per-volume serialization — stays open.
+    #[test]
+    fn test_process_batch_does_not_block_other_volumes() {
+        use crate::config::MinFreeSpace;
+        use crate::storage::types::{DiskType, NeedleId};
+        use crate::storage::volume::{VolumeSpec, needle_write_hook};
+        use std::sync::Barrier;
+
+        let tmp = tempfile::TempDir::new().unwrap();
+        let dir = tmp.path().to_str().unwrap();
+        let state = make_test_state();
+        {
+            let mut store = state.store.write().unwrap();
+            store
+                .add_location(
+                    dir,
+                    dir,
+                    10,
+                    DiskType::HardDrive,
+                    MinFreeSpace::Percent(1.0),
+                    Vec::new(),
+                )
+                .unwrap();
+            for vid in [VolumeId(1), VolumeId(2)] {
+                store
+                    .add_volume(vid, DiskType::HardDrive, &VolumeSpec::default())
+                    .unwrap();
+            }
+        }
+
+        let in_append = Arc::new(Barrier::new(2));
+        let release = Arc::new(Barrier::new(2));
+        let _reg = needle_write_hook::register(NeedleId(101), {
+            let in_append = in_append.clone();
+            let release = release.clone();
+            move |_| {
+                in_append.wait();
+                release.wait();
+            }
+        });
+
+        let (response_tx, mut response_rx) = oneshot::channel();
+        let batch = vec![WriteRequest {
+            volume_id: VolumeId(1),
+            needle: Needle {
+                id: NeedleId(101),
+                cookie: 0x1111.into(),
+                data: vec![7u8; 8],
+                data_size: 8,
+                ..Needle::default()
+            },
+            fsync: false,
+            response_tx,
+            queue_permit: None,
+        }];
+        let writer = {
+            let state = state.clone();
+            std::thread::spawn(move || process_batch(state, batch))
+        };
+
+        in_append.wait(); // volume 1's write is inside its .dat append.
+
+        // The store still admits a lifecycle writer and a volume-2 write.
+        drop(state.store.try_write().unwrap());
+        let vol2 = state
+            .store
+            .read()
+            .unwrap()
+            .writable_volume(VolumeId(2))
+            .unwrap();
+        let mut n2 = Needle {
+            id: NeedleId(9),
+            cookie: 0x2222.into(),
+            data: vec![9u8; 8],
+            data_size: 8,
+            ..Needle::default()
+        };
+        vol2.write_needle(&mut n2, true, false).unwrap();
+
+        // Same-volume exclusion is intact: volume 1's data-file lease is held.
+        let (_, vol1) = state.store.read().unwrap().find_volume(VolumeId(1)).unwrap();
+        assert!(vol1.instance().try_write_lock().is_none());
+
+        release.wait();
+        writer.join().unwrap();
+        assert!(response_rx.try_recv().unwrap().is_ok());
+        assert!(vol1.instance().try_write_lock().is_some());
+    }
+
+    /// A write parked inside volume 1's .dat append must not keep later
+    /// batches for other volumes queued behind it: each volume drains through
+    /// its own FIFO worker, while same-volume writes still serialize.
+    /// NeedleId(102) is reserved for this test — hooks are keyed by needle id
+    /// process-wide and parallel tests must not share one.
+    #[tokio::test]
+    async fn test_write_queue_does_not_block_other_volumes() {
+        use crate::config::MinFreeSpace;
+        use crate::storage::types::{DiskType, NeedleId};
+        use crate::storage::volume::{VolumeSpec, needle_write_hook};
+        use std::sync::Barrier;
+
+        let tmp = tempfile::TempDir::new().unwrap();
+        let dir = tmp.path().to_str().unwrap();
+        let state = make_test_state();
+        {
+            let mut store = state.store.write().unwrap();
+            store
+                .add_location(
+                    dir,
+                    dir,
+                    10,
+                    DiskType::HardDrive,
+                    MinFreeSpace::Percent(1.0),
+                    Vec::new(),
+                )
+                .unwrap();
+            for vid in [VolumeId(1), VolumeId(2)] {
+                store
+                    .add_volume(vid, DiskType::HardDrive, &VolumeSpec::default())
+                    .unwrap();
+            }
+        }
+        let queue = WriteQueue::new(state, MAX_BATCH_SIZE);
+
+        let in_append = Arc::new(Barrier::new(2));
+        let release = Arc::new(Barrier::new(2));
+        let _reg = needle_write_hook::register(NeedleId(102), {
+            let in_append = in_append.clone();
+            let release = release.clone();
+            move |_| {
+                in_append.wait();
+                release.wait();
+            }
+        });
+
+        let needle_a = Needle {
+            id: NeedleId(102),
+            cookie: 0x1111.into(),
+            data: vec![7u8; 8],
+            data_size: 8,
+            ..Needle::default()
+        };
+        let a1 = tokio::spawn({
+            let queue = queue.clone();
+            async move { queue.submit(VolumeId(1), needle_a, false).await }
+        });
+        // The hook parks on a blocking thread, so wait on the barrier off the
+        // runtime worker.
+        let parked = {
+            let in_append = in_append.clone();
+            tokio::task::spawn_blocking(move || in_append.wait())
+        };
+        parked.await.unwrap(); // volume 1's write is inside its .dat append.
+
+        // Volume 2's worker is independent: its submit completes now.
+        let needle_b = Needle {
+            id: NeedleId(9),
+            cookie: 0x2222.into(),
+            data: vec![9u8; 8],
+            data_size: 8,
+            ..Needle::default()
+        };
+        queue.submit(VolumeId(2), needle_b, true).await.unwrap();
+
+        // Same-volume FIFO is preserved: a second volume-1 write stays queued.
+        let needle_a2 = Needle {
+            id: NeedleId(2),
+            cookie: 0x3333.into(),
+            data: vec![8u8; 8],
+            data_size: 8,
+            ..Needle::default()
+        };
+        let mut a2 = tokio::spawn({
+            let queue = queue.clone();
+            async move { queue.submit(VolumeId(1), needle_a2, false).await }
+        });
+        assert!(
+            tokio::time::timeout(std::time::Duration::from_millis(200), &mut a2)
+                .await
+                .is_err(),
+            "a same-volume write must queue behind the stalled one"
+        );
+
+        let released = {
+            let release = release.clone();
+            tokio::task::spawn_blocking(move || release.wait())
+        };
+        released.await.unwrap();
+        assert!(a1.await.unwrap().is_ok());
+        assert!(a2.await.unwrap().is_ok());
+    }
 }

seaweed-volume/src/storage/disk_location.rs

@@ -37,7 +37,7 @@ pub struct DiskLocation {
     pub tags: Vec<String>,
     pub max_volume_count: AtomicI32,
     pub original_max_volume_count: i32,
-    volumes: HashMap<VolumeId, Volume>,
+    volumes: HashMap<VolumeId, Arc<Volume>>,
     ec_volumes: HashMap<VolumeId, EcVolume>,
     pub is_disk_space_low: Arc<AtomicBool>,
     pub available_space: AtomicU64,
@@ -237,6 +237,7 @@ impl DiskLocation {
         }
 
         for (collection, vid, v) in self.open_volumes(to_load, needle_map_kind) {
+            let v = Arc::new(v);
             crate::metrics::VOLUME_GAUGE
                 .with_label_values(&[&collection, "volume"])
                 .inc();
@@ -549,20 +550,23 @@ impl DiskLocation {
         None
     }
 
-    /// Find a volume by ID.
-    pub fn find_volume(&self, vid: VolumeId) -> Option<&Volume> {
-        self.volumes.get(&vid)
-    }
-
-    /// Find a volume by ID (mutable).
-    pub fn find_volume_mut(&mut self, vid: VolumeId) -> Option<&mut Volume> {
-        self.volumes.get_mut(&vid)
+    /// Find a volume by ID. Returns a shared handle like Go's findVolume:
+    /// the lookup borrows the map briefly and I/O runs on the handle after.
+    pub fn find_volume(&self, vid: VolumeId) -> Option<Arc<Volume>> {
+        self.volumes.get(&vid).cloned()
     }
 
     /// Add a volume to this location.
     pub fn set_volume(&mut self, vid: VolumeId, volume: Volume) {
         let collection = volume.collection.clone();
-        self.volumes.insert(vid, volume);
+        if let Some(old) = self.volumes.insert(vid, Arc::new(volume)) {
+            // A held handle would keep writing the same files against a stale
+            // in-memory index; close it so the stale handle fails closed.
+            crate::metrics::VOLUME_GAUGE
+                .with_label_values(&[&old.collection, "volume"])
+                .dec();
+            old.close();
+        }
         crate::metrics::VOLUME_GAUGE
             .with_label_values(&[&collection, "volume"])
             .inc();
@@ -586,13 +590,20 @@ impl DiskLocation {
         crate::metrics::VOLUME_GAUGE
             .with_label_values(&[spec.collection, "volume"])
             .inc();
-        self.volumes.insert(vid, v);
+        if let Some(old) = self.volumes.insert(vid, Arc::new(v)) {
+            // Same fail-closed rule as set_volume: a stale handle must not
+            // keep writing the files the replacement just reopened.
+            crate::metrics::VOLUME_GAUGE
+                .with_label_values(&[&old.collection, "volume"])
+                .dec();
+            old.close();
+        }
         Ok(())
     }
 
     /// Remove and close a volume.
-    pub fn unload_volume(&mut self, vid: VolumeId) -> Option<Volume> {
-        if let Some(mut v) = self.volumes.remove(&vid) {
+    pub fn unload_volume(&mut self, vid: VolumeId) -> Option<Arc<Volume>> {
+        if let Some(v) = self.volumes.remove(&vid) {
             crate::metrics::VOLUME_GAUGE
                 .with_label_values(&[&v.collection, "volume"])
                 .dec();
@@ -619,7 +630,7 @@ impl DiskLocation {
         {
             return Err(v.compacting_error());
         }
-        if let Some(mut v) = self.volumes.remove(&vid) {
+        if let Some(v) = self.volumes.remove(&vid) {
             match v.destroy(only_empty, only_garbage, keep_remote_data) {
                 Ok(()) => {
                     crate::metrics::VOLUME_GAUGE
@@ -649,7 +660,7 @@ impl DiskLocation {
             .collect();
 
         for vid in vids {
-            if let Some(mut v) = self.volumes.remove(&vid) {
+            if let Some(v) = self.volumes.remove(&vid) {
                 if let Err(e) = v.destroy(false, false, false) {
                     warn!(volume_id = vid.0, error = %e, "delete collection: failed to destroy volume");
                     self.volumes.insert(vid, v);
@@ -697,7 +708,7 @@ impl DiskLocation {
 
     /// Iterate over all volumes.
     pub fn iter_volumes(&self) -> impl Iterator<Item = (&VolumeId, &Volume)> {
-        self.volumes.iter()
+        self.volumes.iter().map(|(k, v)| (k, v.as_ref()))
     }
 
     /// Number of free volume slots.
@@ -718,12 +729,7 @@ impl DiskLocation {
 
     /// Iterate over all volumes.
     pub fn volumes(&self) -> impl Iterator<Item = (&VolumeId, &Volume)> {
-        self.volumes.iter()
-    }
-
-    /// Iterate over all volumes (mutable).
-    pub fn volumes_mut(&mut self) -> impl Iterator<Item = (&VolumeId, &mut Volume)> {
-        self.volumes.iter_mut()
+        self.volumes.iter().map(|(k, v)| (k, v.as_ref()))
     }
 
     /// Sum of unused space in writable volumes (volumeSizeLimit - actual size per volume).

seaweed-volume/src/storage/erasure_coding/ec_decoder.rs

@@ -502,7 +502,7 @@ mod tests {
         let dir = tmp.path().to_str().unwrap();
 
         // Create volume with data
-        let mut v = Volume::new(
+        let v = Volume::new(
             dir,
             dir,
             VolumeId(1),

seaweed-volume/src/storage/erasure_coding/ec_encoder.rs

@@ -848,7 +848,7 @@ mod tests {
         let dir = tmp.path().to_str().unwrap();
 
         // Create a volume with some data
-        let mut v = Volume::new(
+        let v = Volume::new(
             dir,
             dir,
             VolumeId(1),
@@ -895,7 +895,7 @@ mod tests {
     fn make_volume_with_needles(n: u64) -> TempDir {
         let tmp = TempDir::new().unwrap();
         let dir = tmp.path().to_str().unwrap();
-        let mut v = Volume::new(
+        let v = Volume::new(
             dir,
             dir,
             VolumeId(1),
@@ -970,7 +970,7 @@ mod tests {
     // the dir path so a test can drop/truncate shards and rebuild.
     fn encode_sample_volume(tmp: &TempDir) -> String {
         let dir = tmp.path().to_str().unwrap().to_string();
-        let mut v = Volume::new(
+        let v = Volume::new(
             &dir,
             &dir,
             VolumeId(1),
@@ -1167,7 +1167,7 @@ mod tests {
         use crate::storage::volume::{Volume, VolumeSpec};
         let tmp = TempDir::new().unwrap();
         let dir = tmp.path().to_str().unwrap().to_string();
-        let mut v = Volume::new(
+        let v = Volume::new(
             &dir,
             &dir,
             VolumeId(2),
@@ -1216,7 +1216,7 @@ mod tests {
         use crate::storage::volume::{Volume, VolumeSpec};
         let tmp = TempDir::new().unwrap();
         let dir = tmp.path().to_str().unwrap().to_string();
-        let mut v = Volume::new(
+        let v = Volume::new(
             &dir,
             &dir,
             VolumeId(3),
@@ -1316,7 +1316,7 @@ mod tests {
         let idx_dir = idx_tmp.path().to_str().unwrap();
 
         // Create a volume with separate data and index directories
-        let mut v = Volume::new(
+        let v = Volume::new(
             dat_dir,
             idx_dir,
             VolumeId(1),
@@ -1396,7 +1396,7 @@ mod tests {
         let idx_dir = idx_tmp.path().to_str().unwrap();
         let wrong_dir = wrong_tmp.path().to_str().unwrap();
 
-        let mut v = Volume::new(
+        let v = Volume::new(
             dat_dir,
             idx_dir,
             VolumeId(1),
@@ -1430,7 +1430,7 @@ mod tests {
     /// a caller-chosen directory, which is what a split-disk test needs: the
     /// shards have to be scattered out of the directory they were encoded into.
     fn seed_encoded_volume(dir: &str, vid: VolumeId) {
-        let mut v = Volume::new(
+        let v = Volume::new(
             dir,
             dir,
             vid,

seaweed-volume/src/storage/erasure_coding/ec_volume.rs

@@ -2382,7 +2382,7 @@ mod tests {
 
         let tmp = TempDir::new().unwrap();
         let dir = tmp.path().to_str().unwrap();
-        let mut v = Volume::new(
+        let v = Volume::new(
             dir,
             dir,
             VolumeId(2074),
@@ -2457,7 +2457,7 @@ mod tests {
 
         let tmp = TempDir::new().unwrap();
         let dir = tmp.path().to_str().unwrap();
-        let mut v = Volume::new(
+        let v = Volume::new(
             dir,
             dir,
             VolumeId(1),
@@ -2511,7 +2511,7 @@ mod tests {
 
         let tmp = TempDir::new().unwrap();
         let dir = tmp.path().to_str().unwrap();
-        let mut v = Volume::new(
+        let v = Volume::new(
             dir,
             dir,
             VolumeId(1),
@@ -2600,7 +2600,7 @@ mod tests {
 
         let tmp = TempDir::new().unwrap();
         let dir = tmp.path().to_str().unwrap();
-        let mut v = Volume::new(
+        let v = Volume::new(
             dir,
             dir,
             VolumeId(1),
@@ -2688,7 +2688,7 @@ mod tests {
 
         let tmp = TempDir::new().unwrap();
         let dir = tmp.path().to_str().unwrap();
-        let mut v = Volume::new(
+        let v = Volume::new(
             dir,
             dir,
             VolumeId(1),
@@ -2783,7 +2783,7 @@ mod tests {
 
         let tmp = TempDir::new().unwrap();
         let dir = tmp.path().to_str().unwrap();
-        let mut v = Volume::new(
+        let v = Volume::new(
             dir,
             dir,
             VolumeId(1),
@@ -4201,7 +4201,7 @@ mod tests {
         use crate::storage::needle_map::NeedleMapKind;
         use crate::storage::volume::{Volume, VolumeSpec};
 
-        let mut v = Volume::new(
+        let v = Volume::new(
             dir,
             dir,
             vid,
@@ -5014,7 +5014,7 @@ mod uniform_layout_tests {
             let dir = tmp.path().to_str().unwrap();
             let vid = VolumeId(8);
 
-            let mut v = Volume::new(
+            let v = Volume::new(
                 dir,
                 dir,
                 vid,

seaweed-volume/src/storage/store.rs

@@ -6,8 +6,8 @@
 
 use std::collections::HashSet;
 use std::io;
-use std::sync::Mutex;
 use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
+use std::sync::{Arc, Mutex};
 
 use crate::config::MinFreeSpace;
 use crate::pb::master_pb;
@@ -43,12 +43,12 @@ fn ensure_compact_volume_space(v: &Volume, preallocate: u64) -> Result<(), Volum
 
     if dir_idx.is_empty() || dir_idx == dir {
         check(dir, data_bytes + index_bytes)
-    } else if !same_filesystem(dir, dir_idx) {
+    } else if !same_filesystem(dir, &dir_idx) {
         check(dir, data_bytes)?;
-        check(dir_idx, index_bytes)
+        check(&dir_idx, index_bytes)
     } else {
         check(dir, data_bytes + index_bytes)?;
-        check(dir_idx, index_bytes)
+        check(&dir_idx, index_bytes)
     }
 }
 
@@ -170,8 +170,13 @@ impl Store {
 
     // ---- Volume lookup ----
 
-    /// Find which location contains a volume.
-    pub fn find_volume(&self, vid: VolumeId) -> Option<(usize, &crate::storage::volume::Volume)> {
+    /// Find which location contains a volume, returning a shared handle.
+    /// Mirroring Go's findVolume: the lookup only borrows the volume map,
+    /// so callers can run volume I/O after releasing the store guard.
+    pub fn find_volume(
+        &self,
+        vid: VolumeId,
+    ) -> Option<(usize, Arc<crate::storage::volume::Volume>)> {
         for (i, loc) in self.locations.iter().enumerate() {
             if let Some(v) = loc.find_volume(vid) {
                 return Some((i, v));
@@ -180,19 +185,6 @@ impl Store {
         None
     }
 
-    /// Find which location contains a volume (mutable).
-    pub fn find_volume_mut(
-        &mut self,
-        vid: VolumeId,
-    ) -> Option<(usize, &mut crate::storage::volume::Volume)> {
-        for (i, loc) in self.locations.iter_mut().enumerate() {
-            if let Some(v) = loc.find_volume_mut(vid) {
-                return Some((i, v));
-            }
-        }
-        None
-    }
-
     /// Check if a volume exists.
     pub fn has_volume(&self, vid: VolumeId) -> bool {
         self.find_volume(vid).is_some()
@@ -202,10 +194,10 @@ impl Store {
     /// the volume in place. A no-op when the location has no separate index
     /// directory. Mirrors Go's `Store::ConsolidateVolumeIndex`.
     pub fn consolidate_volume_index(&mut self, vid: VolumeId) -> Result<(), VolumeError> {
-        for loc in self.locations.iter_mut() {
+        for loc in self.locations.iter() {
             let idx_dir = loc.idx_directory.clone();
             let data_dir = loc.directory.clone();
-            if let Some(v) = loc.find_volume_mut(vid) {
+            if let Some(v) = loc.find_volume(vid) {
                 if idx_dir == data_dir {
                     return Ok(());
                 }
@@ -775,22 +767,22 @@ impl Store {
     /// Write a needle to a volume. With `fsync` the volume flushes its .dat
     /// before returning, so the caller can ack a durable write.
     pub fn write_volume_needle(
-        &mut self,
+        &self,
         vid: VolumeId,
         n: &mut Needle,
         fsync: bool,
     ) -> Result<(u64, Size, bool), VolumeError> {
-        self.writable_volume_mut(vid)?.write_needle(n, true, fsync)
+        self.writable_volume(vid)?.write_needle(n, true, fsync)
     }
 
     /// Write a batch of needles to one volume, sharing the syncs of its
     /// durable writes. See `Volume::write_needles_grouped`.
     pub fn write_volume_needles(
-        &mut self,
+        &self,
         vid: VolumeId,
         writes: &mut [(Needle, bool)],
     ) -> Vec<Result<(u64, Size, bool), VolumeError>> {
-        match self.writable_volume_mut(vid) {
+        match self.writable_volume(vid) {
             Ok(vol) => vol.write_needles_grouped(writes),
             // The lookup fails only with NotFound or the disk-space ReadOnly.
             Err(e) => writes
@@ -803,9 +795,12 @@ impl Store {
         }
     }
 
-    fn writable_volume_mut(&mut self, vid: VolumeId) -> Result<&mut Volume, VolumeError> {
+    /// Resolve a mounted volume handle for writing, checking disk space like
+    /// Go's writableVolume. The returned Arc stays valid for I/O after the
+    /// store guard is released; if the volume is unmounted meanwhile the I/O
+    /// fails on the volume's closed state, not on a stale pointer.
+    pub(crate) fn writable_volume(&self, vid: VolumeId) -> Result<Arc<Volume>, VolumeError> {
         // Check disk space on the location containing this volume.
-        // We do this before the mutable borrow to avoid borrow conflicts.
         let loc_idx = self
             .find_volume(vid)
             .map(|(i, _)| i)
@@ -817,13 +812,13 @@ impl Store {
             return Err(VolumeError::ReadOnly(vid));
         }
 
-        let (_, vol) = self.find_volume_mut(vid).ok_or(VolumeError::NotFound)?;
+        let (_, vol) = self.find_volume(vid).ok_or(VolumeError::NotFound)?;
         Ok(vol)
     }
 
     /// Delete a needle from a volume.
     pub fn delete_volume_needle(
-        &mut self,
+        &self,
         vid: VolumeId,
         n: &mut Needle,
     ) -> Result<Size, VolumeError> {
@@ -832,8 +827,6 @@ impl Store {
         if vol.is_no_write_or_delete() {
             return Err(VolumeError::ReadOnly(vid));
         }
-
-        let (_, vol) = self.find_volume_mut(vid).ok_or(VolumeError::NotFound)?;
         vol.delete_needle(n)
     }
 
@@ -1604,10 +1597,10 @@ impl Store {
         let (_, v) = self
             .find_volume(vid)
             .ok_or(VolumeError::VolumeNotFound(vid))?;
-        ensure_compact_volume_space(v, preallocate)?;
+        ensure_compact_volume_space(&v, preallocate)?;
 
         let (_, v) = self
-            .find_volume_mut(vid)
+            .find_volume(vid)
             .ok_or(VolumeError::VolumeNotFound(vid))?;
         v.begin_compact_by_index()
     }
@@ -1625,8 +1618,8 @@ impl Store {
             collection,
             ..VolumeSpec::default()
         };
-        let mut v = Volume::new(dir, dir_idx, vid, needle_map_kind, &spec)?;
-        let mut compact = || -> Result<(), VolumeError> {
+        let v = Volume::new(dir, dir_idx, vid, needle_map_kind, &spec)?;
+        let compact = || -> Result<(), VolumeError> {
             ensure_compact_volume_space(&v, 0)?;
             v.compact_by_index(0, 0, |_| true)?;
             v.commit_compact()
@@ -1646,7 +1639,7 @@ impl Store {
     /// Commit a completed compaction: swap files and reload.
     pub fn commit_compact_volume(&mut self, vid: VolumeId) -> Result<(bool, u64), VolumeError> {
         let (_, v) = self
-            .find_volume_mut(vid)
+            .find_volume(vid)
             .ok_or(VolumeError::VolumeNotFound(vid))?;
         let is_read_only = v.is_read_only();
         v.commit_compact()?;
@@ -1657,7 +1650,7 @@ impl Store {
     /// Clean up leftover compaction files.
     pub fn cleanup_compact_volume(&mut self, vid: VolumeId) -> Result<(), VolumeError> {
         let (_, v) = self
-            .find_volume_mut(vid)
+            .find_volume(vid)
             .ok_or(VolumeError::VolumeNotFound(vid))?;
         v.cleanup_compact()
     }
@@ -2790,7 +2783,7 @@ mod tests {
         }
         let revision = {
             let (_, v) = store.find_volume(vid).unwrap();
-            v.super_block.compaction_revision
+            v.super_block().compaction_revision
         };
 
         let job = store
@@ -2803,20 +2796,20 @@ mod tests {
         assert!(store.delete_volume(vid, false, false, false).is_err());
         store.delete_collection("").unwrap();
         assert!(store.cleanup_compact_volume(vid).is_err());
-        let (_, v) = store.find_volume_mut(vid).unwrap();
+        let (_, v) = store.find_volume(vid).unwrap();
         assert!(v.relocate_index_to(other.path().to_str().unwrap()).is_err());
         // Go parity: a commit that finds the volume compacting is a no-op.
         store.commit_compact_volume(vid).unwrap();
 
         let (_, v) = store.find_volume(vid).expect("still mounted");
         assert!(v.is_compacting());
-        assert_eq!(v.super_block.compaction_revision, revision);
+        assert_eq!(v.super_block().compaction_revision, revision);
 
         job.run(|_| true).unwrap();
         assert!(!store.find_volume(vid).unwrap().1.is_compacting());
         store.commit_compact_volume(vid).unwrap();
         let (_, v) = store.find_volume(vid).unwrap();
-        assert_eq!(v.super_block.compaction_revision, revision + 1);
+        assert_eq!(v.super_block().compaction_revision, revision + 1);
         assert!(store.unmount_volume(vid).unwrap());
     }
 
@@ -2856,7 +2849,7 @@ mod tests {
         store.delete_volume_needle(vid, &mut del).unwrap();
         let revision = {
             let (_, v) = store.find_volume(vid).unwrap();
-            v.super_block.compaction_revision
+            v.super_block().compaction_revision
         };
 
         let job = store.begin_compact_volume(vid, 0).unwrap().unwrap();
@@ -2885,7 +2878,7 @@ mod tests {
         store.commit_compact_volume(vid).unwrap();
 
         let (_, v) = store.find_volume(vid).unwrap();
-        assert_eq!(v.super_block.compaction_revision, revision + 1);
+        assert_eq!(v.super_block().compaction_revision, revision + 1);
         assert_eq!(read_test_needle(&store, vid, 99).unwrap(), b"late-write");
         assert_eq!(read_test_needle(&store, vid, 3).unwrap(), b"overwritten");
         assert!(read_test_needle(&store, vid, 4).is_err());

seaweed-volume/src/storage/volume_idx_rebuild.rs

@@ -142,7 +142,7 @@ mod tests {
         let old_idx = old_idx_dir.to_str().unwrap();
         let new_idx = new_idx_dir.to_str().unwrap();
 
-        let mut v = Volume::new(
+        let v = Volume::new(
             data,
             old_idx,
             VolumeId(1),
@@ -196,7 +196,7 @@ mod tests {
         let root = TempDir::new().unwrap();
         let dir = root.path().to_str().unwrap();
 
-        let mut v = Volume::new(
+        let v = Volume::new(
             dir,
             dir,
             VolumeId(1),
@@ -242,7 +242,7 @@ mod tests {
         let root = TempDir::new().unwrap();
         let dir = root.path().to_str().unwrap();
 
-        let mut v = Volume::new(
+        let v = Volume::new(
             dir,
             dir,
             VolumeId(1),
@@ -291,7 +291,7 @@ mod tests {
         let root = TempDir::new().unwrap();
         let dir = root.path().to_str().unwrap();
 
-        let mut v = Volume::new(
+        let v = Volume::new(
             dir,
             dir,
             VolumeId(1),

seaweed-volume/src/storage/volume_idx_repair.rs

@@ -12,7 +12,7 @@ use crate::storage::idx;
 use crate::storage::needle::Needle;
 use crate::storage::needle::needle::needle_body_length;
 use crate::storage::types::*;
-use crate::storage::volume::{Volume, VolumeError, fsync_dir};
+use crate::storage::volume::{Volume, VolumeError, VolumeIo, fsync_dir};
 
 /// Needles found in the head of .dat, keyed by id, plus the ids in .dat order.
 type DatHeadNeedles = (HashMap<NeedleId, (Offset, Size)>, Vec<NeedleId>);
@@ -45,15 +45,15 @@ impl Volume {
     /// restores .idx to .dat append order, which several readers lean on: the
     /// fingerprint is gone so a later load stops after one row, and the last row
     /// is the .dat-tail needle again.
-    pub(crate) fn repair_idx_head_tombstones(&self) -> Result<usize, VolumeError> {
-        if !self.has_data_backend() {
+    pub(crate) fn repair_idx_head_tombstones(&self, io: &VolumeIo) -> Result<usize, VolumeError> {
+        if !self.has_data_backend_io(io) {
             return Ok(0);
         }
         let version = self.version();
         if !version.is_supported() {
             return Ok(0);
         }
-        let first_needle_offset = self.super_block.block_size() as i64;
+        let first_needle_offset = self.super_block().block_size() as i64;
         let idx_path = self.file_name(".idx");
 
         let clobbered = idx_head_tombstone_count(&idx_path)?;
@@ -68,7 +68,7 @@ impl Volume {
             "idx starts with offset-0 tombstones, recovering the rows they overwrote from .dat"
         );
 
-        let (mut lost, order) = self.scan_dat_head(version, first_needle_offset, clobbered)?;
+        let (mut lost, order) = self.scan_dat_head(io, version, first_needle_offset, clobbered)?;
         drop_indexed_keys(&idx_path, &mut lost)?;
         if lost.is_empty() {
             return Ok(0);
@@ -93,6 +93,7 @@ impl Volume {
     /// being resurrected.
     fn scan_dat_head(
         &self,
+        io: &VolumeIo,
         version: Version,
         first_needle_offset: i64,
         limit: usize,
@@ -103,7 +104,7 @@ impl Volume {
 
         for _ in 0..limit {
             let mut header = [0u8; NEEDLE_HEADER_SIZE];
-            match self.read_exact_at_backend(&mut header, offset as u64) {
+            match self.read_exact_at_backend(io, &mut header, offset as u64) {
                 Ok(()) => {}
                 Err(VolumeError::Io(ref e)) if e.kind() == io::ErrorKind::UnexpectedEof => break,
                 Err(e) => return Err(e),
@@ -245,7 +246,7 @@ mod tests {
     }
 
     fn write_test_volume(dir: &str, needle_count: u64) -> Vec<Vec<u8>> {
-        let mut v = open_volume(dir);
+        let v = open_volume(dir);
         let mut written = Vec::new();
         for i in 1..=needle_count {
             let data = format!("needle-{}-payload", i).into_bytes();
@@ -355,7 +356,7 @@ mod tests {
         let rows = idx_rows(&idx_path);
         assert_eq!(
             rows[0].1,
-            v.super_block.block_size() as i64,
+            v.super_block().block_size() as i64,
             "first row does not index the first needle in .dat"
         );
         for i in 1..4 {
@@ -380,7 +381,7 @@ mod tests {
         let idx_path = format!("{}/1.idx", dir);
 
         write_test_volume(dir, 6);
-        let mut v = open_volume(dir);
+        let v = open_volume(dir);
         v.delete_needle(&mut Needle {
             id: NeedleId(3),
             ..Needle::default()
@@ -415,7 +416,7 @@ mod tests {
         let idx_path = format!("{}/1.idx", dir);
 
         write_test_volume(dir, 8);
-        let mut v = open_volume(dir);
+        let v = open_volume(dir);
         v.delete_needle(&mut Needle {
             id: NeedleId(2),
             ..Needle::default()
@@ -456,8 +457,10 @@ mod tests {
         drop(dat);
 
         let v = open_volume(dir);
-        let first = v.super_block.block_size() as i64;
-        let err = v.scan_dat_head(v.version(), first, 10).unwrap_err();
+        let first = v.super_block().block_size() as i64;
+        let err = v
+            .scan_dat_head(&v.io_read_for_test(), v.version(), first, 10)
+            .unwrap_err();
         assert!(
             matches!(&err, VolumeError::Io(e) if e.kind() == io::ErrorKind::InvalidData),
             "expected a corrupt-data error, got {err:?}"

seaweed-volume/tests/http_integration.rs

@@ -511,7 +511,7 @@ async fn write_and_delete_on_read_only_volume_say_is_read_only() {
         .store
         .write()
         .unwrap()
-        .find_volume_mut(VolumeId(1))
+        .find_volume(VolumeId(1))
         .unwrap()
         .1
         .set_no_write_or_delete(true);
@@ -539,7 +539,7 @@ async fn write_and_delete_on_read_only_volume_say_is_read_only() {
         .store
         .write()
         .unwrap()
-        .find_volume_mut(VolumeId(1))
+        .find_volume(VolumeId(1))
         .unwrap()
         .1
         .set_no_write_or_delete(false);
@@ -1044,7 +1044,7 @@ async fn chunk_manifest_expands_chunk_stored_on_ec_volume() {
     // Build regular volume 2 holding the chunk, then EC-encode it and mount all
     // 14 shards locally so reconstruct-on-read is a pure local read.
     {
-        let mut v = Volume::new(
+        let v = Volume::new(
             dir,
             dir,
             VolumeId(2),
@@ -1084,7 +1084,7 @@ async fn chunk_manifest_expands_chunk_stored_on_ec_volume() {
     let manifest_nid = NeedleId(0x7);
     let manifest_cookie = Cookie(0x55667788);
     {
-        let mut store = state.store.write().unwrap();
+        let store = state.store.read().unwrap();
         let mut n = Needle {
             id: manifest_nid,
             cookie: manifest_cookie,
@@ -1148,7 +1148,7 @@ async fn delete_on_ec_volume_succeeds_when_the_needles_shard_is_not_mounted() {
     // Build regular volume 4, then EC-encode it. The volume is standalone and
     // never registered in the store, so afterwards it exists only as shards.
     {
-        let mut v = Volume::new(
+        let v = Volume::new(
             dir,
             dir,
             VolumeId(4),

seaweed-volume/tests/store_lock_repro.rs

@@ -0,0 +1,89 @@
+//! Issue #791: the shared `RwLock<Store>` guards the volume map only.
+//! `find_volume` hands out an `Arc<Volume>` so needle I/O runs without the
+//! guard; a handle that outlives its mount must fail closed rather than I/O
+//! a torn-down volume — the same end Go reaches through nil needle-map/dat
+//! checks under the per-volume dataFileAccessLock.
+
+use std::sync::{Arc, RwLock};
+
+use seaweed_volume::config::MinFreeSpace;
+use seaweed_volume::storage::needle::needle::Needle;
+use seaweed_volume::storage::needle_map::NeedleMapKind;
+use seaweed_volume::storage::store::Store;
+use seaweed_volume::storage::types::{DiskType, NeedleId, VolumeId};
+use seaweed_volume::storage::volume::VolumeSpec;
+
+use tempfile::TempDir;
+
+fn store_with_two_volumes() -> (Arc<RwLock<Store>>, TempDir) {
+    let tmp = TempDir::new().unwrap();
+    let dir = tmp.path().to_str().unwrap();
+    let mut store = Store::new(NeedleMapKind::InMemory);
+    store
+        .add_location(dir, dir, 10, DiskType::HardDrive, MinFreeSpace::Percent(1.0), Vec::new())
+        .unwrap();
+    store
+        .add_volume(VolumeId(1), DiskType::HardDrive, &VolumeSpec::default())
+        .unwrap();
+    store
+        .add_volume(VolumeId(2), DiskType::HardDrive, &VolumeSpec::default())
+        .unwrap();
+    (Arc::new(RwLock::new(store)), tmp)
+}
+
+fn needle(id: u64) -> Needle {
+    Needle {
+        id: NeedleId(id),
+        cookie: 0x1234.into(),
+        data: vec![7u8; 16],
+        data_size: 16,
+        ..Needle::default()
+    }
+}
+
+#[test]
+fn store_read_guard_admits_cross_volume_io() {
+    let (store, _tmp) = store_with_two_volumes();
+
+    // Read-guarded callers can write on different volumes concurrently —
+    // what used to require store.write() now only resolves a handle.
+    let t = {
+        let store = store.clone();
+        std::thread::spawn(move || {
+            let store = store.read().unwrap();
+            store
+                .write_volume_needle(VolumeId(1), &mut needle(1), false)
+                .unwrap();
+        })
+    };
+    {
+        let store = store.read().unwrap();
+        store
+            .write_volume_needle(VolumeId(2), &mut needle(9), false)
+            .unwrap();
+    }
+    t.join().unwrap();
+
+    let store = store.read().unwrap();
+    let mut n = needle(9);
+    n.data.clear();
+    assert_eq!(store.read_volume_needle(VolumeId(2), &mut n).unwrap(), 16);
+    let mut n = needle(1);
+    n.data.clear();
+    assert_eq!(store.read_volume_needle(VolumeId(1), &mut n).unwrap(), 16);
+}
+
+#[test]
+fn stale_volume_handle_fails_after_unmount() {
+    let (store, _tmp) = store_with_two_volumes();
+
+    // A handle resolved before unmount keeps the Volume alive but must not
+    // be able to write into its closed files.
+    let (_, vol1) = store.read().unwrap().find_volume(VolumeId(1)).unwrap();
+    store.write().unwrap().unmount_volume(VolumeId(1)).unwrap();
+
+    assert!(vol1.write_needle(&mut needle(1), true, false).is_err());
+    let mut n = needle(1);
+    n.data.clear();
+    assert!(vol1.read_needle(&mut n).is_err());
+}