volume tier: accept deletes on a volume as soon as it is tiered (#11894)

* volume tier: accept deletes on a volume as soon as it is tiered

volume.tier.upload marks the volume read-only without deletes before the
upload (markVolumeReplicasWritable(..., false, false), i.e. noWriteOrDelete),
and nothing clears that until the volume server restarts, when load() gives
every tiered volume noWriteCanDelete. Every delete in between is refused with
"volume N is read only": the needle stays live in the index, and its bytes in
the remote object, until someone notices.

After the remote file is loaded, the upload handler now gives the volume the
flags a restart would (MarkTieredDeletable: writes refused, deletes accepted),
with two exceptions that keep the current behaviour:
- a replicated volume: the shell copies the uploaded replica's .idx and .vif
  to the other replicas after the upload, and a tombstone written before that
  copy would reach one replica only;
- a volume in I/O quarantine: that flag is not the upload's to lift.

Tests: TestTierUploadAcceptsDeletesOnASingleCopyVolume fails without the
change ("volume 1 is read only") and checks writes stay refused;
TestTierUploadKeepsRefusingDeletesOnAReplicatedVolume; TestMarkTieredDeletable
(quarantine kept).

* volume tier: check the I/O quarantine under noWriteLock

MarkTieredDeletable read the quarantine state before taking noWriteLock.
CollectHeartbeat could quarantine the volume and set noWriteOrDelete in
between, and the upload then cleared that refusal, so a quarantined volume
accepted deletes until the next heartbeat. Take noWriteLock first: the
heartbeat marks the quarantine before it takes noWriteLock, so either the
check sees it or the heartbeat's refusal lands afterwards and stays.

Also leave alone a volume past IoErrorTolerance that the next heartbeat
will quarantine, as the heartbeat itself does.

* volume: read noWriteOrDelete under noWriteLock in DeleteVolumeNeedle

DeleteVolumeNeedle checked v.noWriteOrDelete without the lock that
MarkVolumeReadonly, CollectHeartbeat and now MarkTieredDeletable hold when
they write it, so a delete racing a tier upload was a data race. Copy the
flag under the read lock; the delete path is otherwise unchanged.

* volume tier: mirror the delete acceptance rules in the Rust volume server

The Rust tier upload already promoted a freshly tiered volume to
no-write-can-delete via refresh_remote_write_mode, and did so for every
volume -- including the replicated and I/O-quarantined ones the Go
MarkTieredDeletable deliberately refuses. A live tier-up now goes
through mark_tiered_deletable (replicated volumes and quarantines keep
their upload-time mark until the next load), while load() keeps
promoting every tiered volume unconditionally, as Go does.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* volume: keep a tiered volume delete-only when marked writable

MarkVolumeWritable preserved noWriteCanDelete on a remote-tiered volume
only when it was already set. A tiered volume still holding the
upload-time mark (a replicated volume, or one whose tier upload ran
before MarkTieredDeletable existed) had it clear, so marking writable
published the volume as fully writable while its .dat lives remotely --
writes then failed against a backend that cannot append, after the
master was told the volume is writable again.

load() gives every remote-tiered volume noWriteCanDelete; deriving the
same flag here keeps mark-writable consistent with a restart: "writable"
on a remote volume resolves to can-delete.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* volume: keep a tiered volume delete-only when marked writable (rust)

set_writable preserved the remote delete-only mode only when the flag was
already set, so a tiered volume holding the upload-time read-only mark
(replicated, or quarantined under mark_tiered_deletable) was published
fully writable with its .dat remote. Derive no_write_can_delete from
has_remote_file() instead, matching load() and the Go MarkVolumeWritable.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* volume tier: serialize the quarantine check with the flag swap (rust)

A heartbeat refusal landing between should_quarantine() and the store
could be clobbered until the next heartbeat re-asserted it — the same
race the Go side closed by checking under noWriteLock. A
compare-exchange gives the same ordering: the swap fails, and the retry
sees the quarantine mark the heartbeat sets first.

* volume tier: MarkTieredDeletable refuses a non-tiered volume

A stray call on a local single-copy volume would pin it
read-only-can-delete. The only caller promotes right after
LoadRemoteFile, but the guard makes the method safe to reuse; the Rust
mark_tiered_deletable gets the same check, though its caller already
only reaches it under has_remote.

---------

Co-authored-by: Javier de la Torre <jatorre@lakehousebox.com>
Co-authored-by: chrislusf <chris.lu@gmail.com>
Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

seaweed-volume/src/storage/volume.rs

@@ -4185,8 +4185,10 @@ impl Volume {
         let _sync_guard = self.data_file_sync_lock.lock().unwrap();
         let _guard = self.data_file_access_control.write_lock();
         let (was_no_write_or_delete, was_no_write_can_delete) = self.no_write_flags();
-        // Remote-tiered volumes must stay no_write_can_delete regardless of marks.
-        let no_write_can_delete = was_no_write_can_delete && self.has_remote_file();
+        // A remote-tiered volume keeps the delete-only mode a load derives: its
+        // .dat cannot take appends, so "writable" resolves to can-delete rather
+        // than advertising writable with no local data file.
+        let no_write_can_delete = self.has_remote_file();
         self.no_write.store(
             if no_write_can_delete {
                 NO_WRITE_CAN_DELETE
@@ -4300,21 +4302,70 @@ impl Volume {
         }
         f(&mut self.volume_info.write().unwrap().files);
         let mut io = self.io.write().unwrap();
-        self.refresh_remote_write_mode(&mut io)
+        self.refresh_remote_write_mode(&mut io, true)
+    }
+
+    /// Go's MarkTieredDeletable: a live tier-up lifts the upload-time
+    /// read-only mark, except on a replicated volume — the shell copies the
+    /// uploaded replica's .idx and .vif to the other replicas after the
+    /// upload, and a tombstone written before that copy would reach one
+    /// replica only — or under an I/O quarantine, which is not the tier
+    /// upload's to lift. A refused volume keeps its flags until the next
+    /// load, as the Go volume does.
+    fn mark_tiered_deletable(&self) -> bool {
+        if self
+            .super_block
+            .read()
+            .unwrap()
+            .replica_placement
+            .has_replication()
+        {
+            return false;
+        }
+        if !self.has_remote_file() {
+            return false;
+        }
+        // Go takes noWriteLock around the quarantine check and the flag
+        // change; here a compare-exchange gives the same ordering — a
+        // heartbeat refusal landing between the check and the swap fails the
+        // swap, and the retry sees the quarantine mark the heartbeat set
+        // first.
+        loop {
+            if self.should_quarantine() {
+                return false;
+            }
+            let flags = self.no_write.load(Ordering::Acquire);
+            let promoted = (flags & !NO_WRITE_OR_DELETE) | NO_WRITE_CAN_DELETE;
+            if self
+                .no_write
+                .compare_exchange(flags, promoted, Ordering::AcqRel, Ordering::Acquire)
+                .is_ok()
+            {
+                return true;
+            }
+        }
     }
 
     /// Recompute the Go-style write/delete mode from the current remote tier
     /// state, and bring the needle map in line with it — a volume that stops
     /// being remote also stops using the read-only sorted index.
     ///
+    /// `live` is a tier transition on a running volume: the promotion then
+    /// goes through mark_tiered_deletable, Go's MarkTieredDeletable, while a
+    /// load promotes every tiered volume unconditionally as Go's load() does.
+    ///
     /// If the map cannot be rebuilt the volume is pinned read-only rather than
     /// published as writable with an index that rejects every put; a restart
     /// recovers it.
-    fn refresh_remote_write_mode(&self, io: &mut VolumeIo) -> Result<(), VolumeError> {
+    fn refresh_remote_write_mode(&self, io: &mut VolumeIo, live: bool) -> Result<(), VolumeError> {
         let has_remote = self.has_remote_file();
         if has_remote {
             // remote-tiered: no writes, deletes allowed
-            self.no_write.store(NO_WRITE_CAN_DELETE, Ordering::Release);
+            if !live {
+                self.no_write.store(NO_WRITE_CAN_DELETE, Ordering::Release);
+            } else {
+                self.mark_tiered_deletable();
+            }
         } else if !self.volume_info.read().unwrap().read_only_can_delete {
             // only clear the remoteness-derived flag, not an operator mark
             self.set_no_write_flag(NO_WRITE_CAN_DELETE, false);
@@ -4363,7 +4414,7 @@ impl Volume {
                     }
                 }
                 *self.volume_info.write().unwrap() = pb_info;
-                self.refresh_remote_write_mode(io)?;
+                self.refresh_remote_write_mode(io, false)?;
                 {
                     let mut vi = self.volume_info.write().unwrap();
                     if vi.version == 0 {
@@ -4394,7 +4445,7 @@ impl Volume {
                     self.set_no_write_flag(NO_WRITE_OR_DELETE, true);
                 }
                 *self.volume_info.write().unwrap() = pb_info;
-                self.refresh_remote_write_mode(io)?;
+                self.refresh_remote_write_mode(io, false)?;
                 {
                     let mut vi = self.volume_info.write().unwrap();
                     if vi.version == 0 {
@@ -9701,6 +9752,72 @@ mod tests {
         assert!(!v.is_no_write_or_delete());
     }
 
+    /// Go's MarkTieredDeletable leaves a replicated volume's upload-time
+    /// read-only mark alone: the shell still has to copy the uploaded
+    /// replica's .idx to the others, and a tombstone written before that copy
+    /// would reach one replica only.
+    #[test]
+    fn test_update_remote_files_keeps_a_replicated_volume_refusing_deletes() {
+        let tmp = TempDir::new().unwrap();
+        let dir = tmp.path().to_str().unwrap();
+        let v = make_test_volume(dir);
+        v.super_block.write().unwrap().replica_placement =
+            crate::storage::super_block::ReplicaPlacement::from_string("001").unwrap();
+        v.set_no_write_or_delete(true); // the shell's pre-upload mark
+
+        v.update_remote_files(|files| files.push(Default::default()))
+            .unwrap();
+
+        assert!(v.has_remote_file());
+        assert!(
+            v.is_no_write_or_delete(),
+            "a replicated volume keeps refusing deletes until its next load"
+        );
+        assert!(!v.is_no_write_can_delete());
+    }
+
+    /// An I/O quarantine is not the tier upload's to lift: a live tier-up
+    /// leaves a quarantined volume's read-only mark alone, as Go's
+    /// MarkTieredDeletable does.
+    #[test]
+    fn test_update_remote_files_keeps_an_io_quarantine() {
+        let tmp = TempDir::new().unwrap();
+        let dir = tmp.path().to_str().unwrap();
+        let v = make_test_volume(dir);
+        v.set_no_write_or_delete(true);
+        v.mark_io_quarantined();
+
+        v.update_remote_files(|files| files.push(Default::default()))
+            .unwrap();
+
+        assert!(v.has_remote_file());
+        assert!(v.is_no_write_or_delete());
+        assert!(!v.is_no_write_can_delete());
+    }
+
+    /// A refused tiered volume keeps OR_DELETE without CAN_DELETE; marking it
+    /// writable must resolve to the delete-only mode a load derives — the
+    /// remote .dat cannot take appends — not drop every protection.
+    #[test]
+    fn test_set_writable_on_a_refused_tiered_volume_stays_delete_only() {
+        let tmp = TempDir::new().unwrap();
+        let dir = tmp.path().to_str().unwrap();
+        let v = make_test_volume(dir);
+        v.super_block.write().unwrap().replica_placement =
+            crate::storage::super_block::ReplicaPlacement::from_string("001").unwrap();
+        v.set_no_write_or_delete(true);
+
+        v.update_remote_files(|files| files.push(Default::default()))
+            .unwrap();
+        assert!(v.is_no_write_or_delete());
+
+        v.set_writable().unwrap();
+
+        assert!(v.is_read_only());
+        assert!(!v.no_write_flags().0);
+        assert!(v.no_write_flags().1);
+    }
+
     // Tier-down clears the remote mode and publishes the volume as writable. The
     // map it booted with is the read-only sorted one, whose put always fails, so
     // without a rebuild the first write appends to .dat and then cannot be

weed/server/volume_grpc_tier_upload.go

@@ -110,6 +110,8 @@ func (vs *VolumeServer) VolumeTierMoveDatToRemote(req *volume_server_pb.VolumeTi
 	if err := v.LoadRemoteFile(); err != nil {
 		return fmt.Errorf("volume %d failed to load remote file: %v", v.Id, err)
 	}
+	// Deletes must reach the index now, not only after the next restart.
+	v.MarkTieredDeletable()
 
 	if !req.KeepLocalDatFile {
 		os.Remove(v.FileName(".dat"))

weed/server/volume_grpc_tier_upload_deletes_test.go

@@ -0,0 +1,77 @@
+package weed_server
+
+import (
+	"bytes"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
+	"github.com/seaweedfs/seaweedfs/weed/stats"
+	"github.com/seaweedfs/seaweedfs/weed/storage"
+	"github.com/seaweedfs/seaweedfs/weed/storage/backend"
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	"github.com/seaweedfs/seaweedfs/weed/storage/types"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+)
+
+// tierUploadForDeletes writes 10 needles into volume 1 of a new store, marks it
+// read-only as volume.tier.upload does, and moves its .dat to the tier.
+func tierUploadForDeletes(t *testing.T, replication string) *storage.Store {
+	t.Helper()
+	backend.BackendStorages[tierTimestampTestBackendName] = &tierTimestampTestBackend{root: t.TempDir()}
+	t.Cleanup(func() { delete(backend.BackendStorages, tierTimestampTestBackendName) })
+	store := storage.NewStore(nil, "localhost", 8080, 18080, "http://localhost:8080", "store-id",
+		[]string{t.TempDir()}, []int32{10}, []util.MinFreeSpace{{}}, "", storage.NeedleMapInMemory,
+		[]types.DiskType{types.HardDriveType}, nil, 0, stats.DefaultDiskIOProbeConfig())
+	t.Cleanup(store.Close)
+	if err := store.AddVolume(1, "", storage.NeedleMapInMemory, replication, "", 0, needle.Version3, 0, types.HardDriveType, 0); err != nil {
+		t.Fatalf("add volume: %v", err)
+	}
+	for i := 1; i <= 10; i++ {
+		n := &needle.Needle{Id: types.Uint64ToNeedleId(uint64(i)), Cookie: 7, Data: bytes.Repeat([]byte{byte(i)}, 1024)}
+		n.Checksum = needle.NewCRC(n.Data)
+		if _, err := store.WriteVolumeNeedle(1, n, false, false); err != nil {
+			t.Fatalf("write %d: %v", i, err)
+		}
+	}
+	if err := store.MarkVolumeReadonly(1, false, false); err != nil {
+		t.Fatalf("mark read-only: %v", err)
+	}
+	vs := &VolumeServer{store: store}
+	if err := vs.VolumeTierMoveDatToRemote(&volume_server_pb.VolumeTierMoveDatToRemoteRequest{
+		VolumeId: 1, DestinationBackendName: tierTimestampTestBackendName,
+	}, &discardServerStream[volume_server_pb.VolumeTierMoveDatToRemoteResponse]{}); err != nil {
+		t.Fatalf("tier upload: %v", err)
+	}
+	return store
+}
+
+// A single-copy volume takes deletes as soon as it is tiered, as it would
+// after a restart. Before, they were refused ("volume 1 is read only") until
+// then, leaving the needles live in the index.
+func TestTierUploadAcceptsDeletesOnASingleCopyVolume(t *testing.T) {
+	store := tierUploadForDeletes(t, "000")
+	if _, err := store.DeleteVolumeNeedle(1, &needle.Needle{Id: 2}); err != nil {
+		t.Fatalf("delete on the freshly tiered volume: %v", err)
+	}
+	if _, err := store.ReadVolumeNeedle(1, &needle.Needle{Id: 2}, nil, nil); err == nil {
+		t.Fatalf("the deleted needle still reads")
+	}
+	if _, err := store.ReadVolumeNeedle(1, &needle.Needle{Id: 3}, nil, nil); err != nil {
+		t.Fatalf("a live needle does not read: %v", err)
+	}
+	n := &needle.Needle{Id: 99, Cookie: 7, Data: []byte("x")}
+	n.Checksum = needle.NewCRC(n.Data)
+	if _, err := store.WriteVolumeNeedle(1, n, false, false); err == nil {
+		t.Fatalf("a write to a tiered volume succeeded")
+	}
+}
+
+// A replicated volume keeps refusing deletes: the shell copies the uploaded
+// replica's .idx and .vif to the others after the upload, and a tombstone
+// written before that copy would reach one replica only.
+func TestTierUploadKeepsRefusingDeletesOnAReplicatedVolume(t *testing.T) {
+	store := tierUploadForDeletes(t, "001")
+	if _, err := store.DeleteVolumeNeedle(1, &needle.Needle{Id: 2}); err == nil {
+		t.Fatalf("a replicated volume accepted a delete right after the tier upload")
+	}
+}

weed/storage/store.go

@@ -880,7 +880,10 @@ func (s *Store) WriteVolumeNeedle(i needle.VolumeId, n *needle.Needle, checkCook
 
 func (s *Store) DeleteVolumeNeedle(i needle.VolumeId, n *needle.Needle) (Size, error) {
 	if v := s.findVolume(i); v != nil {
-		if v.noWriteOrDelete {
+		v.noWriteLock.RLock()
+		noWriteOrDelete := v.noWriteOrDelete
+		v.noWriteLock.RUnlock()
+		if noWriteOrDelete {
 			return 0, fmt.Errorf("volume %d is read only", i)
 		}
 		return v.deleteNeedle2(n)
@@ -986,10 +989,10 @@ func (s *Store) MarkVolumeWritable(i needle.VolumeId) error {
 	prevNoWriteOrDelete := v.noWriteOrDelete
 	prevNoWriteCanDelete := v.noWriteCanDelete
 	v.noWriteOrDelete = false
-	// Remote-tiered volumes must stay noWriteCanDelete regardless of marks.
-	if !v.HasRemoteFile() {
-		v.noWriteCanDelete = false
-	}
+	// A remote-tiered volume keeps the delete-only mode load() derives: its
+	// .dat cannot take appends, so "writable" resolves to can-delete rather
+	// than advertising writable with no local data file.
+	v.noWriteCanDelete = v.HasRemoteFile()
 	persistErr := v.PersistReadOnly(false, false)
 	if persistErr != nil {
 		var ndErr *volume_info.NotCrashDurableError

weed/storage/volume_mark_writable_test.go

@@ -68,6 +68,33 @@ func TestMarkVolumeWritable_ReopensPersistedReadOnly(t *testing.T) {
 	}
 }
 
+// A remote-tiered volume marked writable keeps the delete-only mode load()
+// derives — the remote .dat cannot take appends, so "writable" resolves to
+// can-delete even when the mark left noWriteOrDelete without noWriteCanDelete.
+func TestMarkVolumeWritableOnATieredVolumeKeepsDeletesOnly(t *testing.T) {
+	b := newLocalDirBackend(t)
+	registerTestBackend(t, b)
+
+	v, _ := tierUpVolumeLive(t, t.TempDir(), 86, b)
+	defer v.Close()
+	// the upload-time mark as a refused MarkTieredDeletable leaves it
+	v.noWriteOrDelete = true
+	v.noWriteCanDelete = false
+
+	location := newTestLocation(10, false, 0)
+	location.SetVolume(86, v)
+	s := &Store{Locations: []*DiskLocation{location}}
+	if err := s.MarkVolumeWritable(86); err != nil {
+		t.Fatalf("mark writable: %v", err)
+	}
+
+	readOnly, orDelete, canDelete, _ := v.ReadOnlyReasons()
+	if orDelete || !canDelete || !readOnly {
+		t.Fatalf("a tiered volume resolves writable to delete-only, got readOnly=%v orDelete=%v canDelete=%v",
+			readOnly, orDelete, canDelete)
+	}
+}
+
 // reopenIdxForWrite must be a no-op when the volume already has a writable
 // needle map — otherwise repeated MarkVolumeWritable calls would churn the
 // index file handle for no reason.

weed/storage/volume_tier.go

@@ -87,6 +87,41 @@ func (v *Volume) LoadRemoteFile() error {
 	return v.loadRemoteFileLocked()
 }
 
+// MarkTieredDeletable gives a volume whose data has just moved to the remote
+// tier the flags a restart gives it (load(): a tiered volume refuses writes and
+// accepts deletes). volume.tier.upload marks the volume read-only without
+// deletes before the upload; left that way, every delete until the next
+// restart is refused, the filer drops it, and the needle's bytes stay in the
+// remote object where no vacuum can find them.
+//
+// Only for a single-copy volume: the shell copies the uploaded replica's .idx
+// and .vif to the other replicas after the upload, and a tombstone written
+// before that copy would reach one replica only; a replicated volume keeps
+// refusing deletes until its next load, as before. An I/O quarantine is left
+// alone, and so is a volume past the I/O error tolerance that the next
+// heartbeat will quarantine. Otherwise the flags are the ones load() gives
+// every tiered volume.
+func (v *Volume) MarkTieredDeletable() bool {
+	if rp := v.SuperBlock.ReplicaPlacement; rp != nil && rp.GetCopyCount() > 1 {
+		return false
+	}
+	// Check the I/O state under noWriteLock: CollectHeartbeat marks the
+	// quarantine before it takes noWriteLock to refuse deletes, so either the
+	// check below sees the quarantine, or the heartbeat's refusal lands after
+	// this change and stays.
+	v.noWriteLock.Lock()
+	defer v.noWriteLock.Unlock()
+	if !v.HasRemoteFile() {
+		return false
+	}
+	if ioErr, ioCount, quarantined := v.getIoErrorState(); quarantined || (ioErr != nil && ioCount >= IoErrorTolerance) {
+		return false
+	}
+	v.noWriteOrDelete = false
+	v.noWriteCanDelete = true
+	return true
+}
+
 // loadRemoteFileLocked swaps the data backend to the remote tier object. The
 // caller must hold dataFileAccessLock or be single-threaded (load() during
 // construction or a compaction-commit reload). It marks the volume tiered in the

weed/storage/volume_tier_deletable_test.go

@@ -0,0 +1,123 @@
+package storage
+
+import (
+	"syscall"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
+	"github.com/seaweedfs/seaweedfs/weed/storage/types"
+	"github.com/stretchr/testify/require"
+)
+
+// MarkTieredDeletable gives a freshly tiered single-copy volume the flags a
+// restart gives it, and leaves an I/O quarantine alone.
+func TestMarkTieredDeletable(t *testing.T) {
+	b := newLocalDirBackend(t)
+	registerTestBackend(t, b)
+
+	v, _ := tierUpVolumeLive(t, t.TempDir(), 81, b)
+	defer v.Close()
+	v.noWriteOrDelete = true // as volume.tier.upload's mark leaves it
+	require.True(t, v.MarkTieredDeletable())
+	require.False(t, v.noWriteOrDelete)
+	require.True(t, v.noWriteCanDelete)
+
+	w, _ := tierUpVolumeLive(t, t.TempDir(), 82, b)
+	defer w.Close()
+	w.noWriteOrDelete = true
+	w.markIoQuarantined()
+	require.False(t, w.MarkTieredDeletable())
+	require.True(t, w.noWriteOrDelete, "an I/O quarantine is not the tier upload's to lift")
+}
+
+// MarkTieredDeletable is only for a tiered volume: a stray call on a local
+// volume must not pin it read-only-can-delete.
+func TestMarkTieredDeletableLeavesANonTieredVolume(t *testing.T) {
+	dir := t.TempDir()
+	v, err := NewVolume(dir, dir, "", 86, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0)
+	require.NoError(t, err)
+	defer v.Close()
+
+	require.False(t, v.MarkTieredDeletable())
+	require.False(t, v.noWriteOrDelete)
+	require.False(t, v.noWriteCanDelete)
+}
+
+// A heartbeat that quarantines the volume while MarkTieredDeletable waits for
+// noWriteLock must keep its refusal: the quarantine check and the flag change
+// happen under the same lock.
+func TestMarkTieredDeletableKeepsAConcurrentQuarantine(t *testing.T) {
+	b := newLocalDirBackend(t)
+	registerTestBackend(t, b)
+
+	v, _ := tierUpVolumeLive(t, t.TempDir(), 83, b)
+	defer v.Close()
+	v.noWriteOrDelete = true
+
+	v.noWriteLock.Lock()
+	done := make(chan bool)
+	go func() { done <- v.MarkTieredDeletable() }()
+	time.Sleep(100 * time.Millisecond) // let it reach noWriteLock
+	// What CollectHeartbeat does: mark the quarantine, then refuse writes and deletes.
+	v.markIoQuarantined()
+	v.noWriteOrDelete = true
+	v.noWriteLock.Unlock()
+
+	require.False(t, <-done, "a volume quarantined before the flags changed must not be made deletable")
+	v.noWriteLock.RLock()
+	defer v.noWriteLock.RUnlock()
+	require.True(t, v.noWriteOrDelete, "the heartbeat's refusal was overwritten")
+}
+
+// A volume past the I/O error tolerance is about to be quarantined by the next
+// heartbeat; it is not made deletable in the meantime.
+func TestMarkTieredDeletableRefusesAVolumePastIoErrorTolerance(t *testing.T) {
+	b := newLocalDirBackend(t)
+	registerTestBackend(t, b)
+
+	v, _ := tierUpVolumeLive(t, t.TempDir(), 84, b)
+	defer v.Close()
+	v.noWriteOrDelete = true
+	for i := 0; i < IoErrorTolerance; i++ {
+		v.noteIoError(syscall.EIO)
+	}
+	require.False(t, v.MarkTieredDeletable())
+	require.True(t, v.noWriteOrDelete)
+
+	v.clearIoError() // positive control: below the tolerance it is made deletable
+	require.True(t, v.MarkTieredDeletable())
+	require.False(t, v.noWriteOrDelete)
+}
+
+// DeleteVolumeNeedle reads the read-only flags under noWriteLock, so a delete
+// that races the tier upload's flag change is not a data race (run with -race).
+func TestDeleteVolumeNeedleRacesMarkTieredDeletable(t *testing.T) {
+	b := newLocalDirBackend(t)
+	registerTestBackend(t, b)
+
+	v, _ := tierUpVolumeLive(t, t.TempDir(), 85, b)
+	defer v.Close()
+	v.noWriteOrDelete = true
+	location := newTestLocation(10, false, 0)
+	location.SetVolume(85, v)
+	s := &Store{Locations: []*DiskLocation{location}}
+
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		for i := 0; i < 100; i++ {
+			v.MarkTieredDeletable()
+		}
+	}()
+	for i := 0; i < 100; i++ {
+		_, _ = s.DeleteVolumeNeedle(85, &needle.Needle{Id: types.Uint64ToNeedleId(1000)})
+	}
+	<-done
+
+	// Positive control: once deletable, a delete of a live needle goes through.
+	size, err := s.DeleteVolumeNeedle(85, &needle.Needle{Id: types.Uint64ToNeedleId(1)})
+	require.NoError(t, err)
+	require.NotZero(t, size)
+}