volume: never compact a volume tiered to remote storage (#11895)

* volume: never compact a volume tiered to remote storage

A tiered volume's data backend is the remote object, but compaction reads
the local .dat path. When tiering removed the local .dat, every vacuum of a
tiered volume with garbage fails with "no such file or directory", and the
master's sweep and the worker's vacuum task keep selecting it. When the
local .dat was kept, the compaction succeeds against the stale copy and the
commit reloads the volume to read the remote object (old layout) through the
compacted index, so live needles stop reading.

Mirror the Rust volume server (#11545):
- CompactByIndex and CompactByVolumeData refuse a volume with a remote file
  before creating .cpd/.cpx ("volume N is tiered to remote storage, cannot
  compact").
- CommitCompact refuses under the data-file locks when the volume became
  tiered during the copy, and removes .cpd/.cpx.
- Store.CheckCompactVolume reports no garbage for a tiered volume, so the
  master's vacuum and volume.vacuum do not select it.
- The worker's vacuum detection skips a volume with a remote copy, as balance
  and EC detection already do.

Tests (each fails on master): TestCompactRefusesTieredVolume,
TestCommitCompactRefusesVolumeTieredDuringCopy (on master the commit
succeeds and a surviving needle no longer reads),
TestPickPrimaryReplicaSkipsTieredVolumes.

* volume: skip tiered replicas in vacuum selection, keep their real ratio

Reporting a zero garbage ratio for a tiered volume hid it from
volume.tier.compact, whose preflight reads that ratio while the volume is
still remote, and did not exclude it from a vacuum with a zero threshold.
Report the real ratio again, as the Rust volume server does, and have the
master's vacuum check skip a replica whose volume info names remote
storage, whatever the threshold or request.

When CommitCompact refuses a volume tiered during the copy, also close the
temporary needle map and remove .cpldb: nothing else would, since no later
vacuum selects the volume.

The tiered-compaction test no longer depends on random needle sizes
reaching a 0.3 garbage ratio.

* volume tier: do not attach a copy of a .dat a compaction replaced

The upload copies the .dat, then records the remote file in the .vif and
switches the backend. A compaction commit in between swaps in a compacted
.dat and index; the upload then served the old-layout remote object through
the new index, so live needles stopped reading. CommitCompact's tiered
check cannot see this: the volume is not tiered until the switch.

AttachRemoteDatFile records the remote file and switches the backend under
the data-file locks, and refuses when the backend is no longer the file
that was copied. The upload then removes the unused remote copy.

* volume tier: keep a committed .vif entry when only the dir fsync fails

AttachRemoteDatFile rolled back the in-memory remote entry on any
SaveVolumeInfo error. A NotCrashDurableError comes after the rename, so
the .vif on disk already names the copy; with the entry gone, the upload
handler deleted that object, and a reload would serve from a missing
remote file. Keep the entry and switch to the copy as the .vif says,
and still return the error so the caller keeps the local .dat.

* vacuum worker: filter tiered replicas instead of skipping the volume

A partially tiered volume is the leftover of an interrupted
volume.tier.upload; skipping the whole volume left garbage on its local
replicas unreclaimed. Per-replica filtering matches the master's
batchVacuumVolumeCheck and is safe: a tiered replica refuses both
compaction and commit.

---------

Co-authored-by: Javier de la Torre <jatorre@lakehousebox.com>
Co-authored-by: Chris Lu <chrislusf@users.noreply.github.com>

weed/server/volume_grpc_tier_upload.go

@@ -5,6 +5,7 @@ import (
 	"os"
 	"time"
 
+	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
 	"github.com/seaweedfs/seaweedfs/weed/storage/backend"
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
@@ -92,8 +93,9 @@ func (vs *VolumeServer) VolumeTierMoveDatToRemote(req *volume_server_pb.VolumeTi
 		return fmt.Errorf("backend %s copy file %s: %v", req.DestinationBackendName, diskFile.Name(), err)
 	}
 
-	// save the remote file to volume tier info
-	v.GetVolumeInfo().Files = append(v.GetVolumeInfo().GetFiles(), &volume_server_pb.RemoteFile{
+	// record the remote file and switch to it, unless a compaction commit
+	// replaced the .dat while it was being copied
+	if err := v.AttachRemoteDatFile(diskFile, &volume_server_pb.RemoteFile{
 		BackendType:  backendType,
 		BackendId:    backendId,
 		Key:          key,
@@ -101,14 +103,20 @@ func (vs *VolumeServer) VolumeTierMoveDatToRemote(req *volume_server_pb.VolumeTi
 		FileSize:     uint64(size),
 		ModifiedTime: uint64(modTime.Unix()),
 		Extension:    ".dat",
-	})
-
-	if err := v.SaveVolumeInfo(); err != nil {
-		return fmt.Errorf("volume %d failed to save remote file info: %v", v.Id, err)
-	}
-
-	if err := v.LoadRemoteFile(); err != nil {
-		return fmt.Errorf("volume %d failed to load remote file: %v", v.Id, err)
+	}); err != nil {
+		// Remove the copy unless the .vif already names it.
+		recorded := false
+		for _, f := range v.GetVolumeInfo().GetFiles() {
+			if f.Key == key && f.BackendType == backendType && f.BackendId == backendId {
+				recorded = true
+			}
+		}
+		if !recorded {
+			if delErr := backendStorage.DeleteFile(key); delErr != nil {
+				glog.Warningf("volume %d: remove unused remote copy %s: %v", v.Id, key, delErr)
+			}
+		}
+		return fmt.Errorf("volume %d failed to attach remote file: %v", v.Id, err)
 	}
 	// Deletes must reach the index now, not only after the next restart.
 	v.MarkTieredDeletable()

weed/server/volume_grpc_tier_upload_compact_test.go

@@ -0,0 +1,107 @@
+package weed_server
+
+import (
+	"bytes"
+	"os"
+	"path/filepath"
+	"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"
+)
+
+// afterCopyBackend runs a hook once the .dat has been copied, before the
+// upload handler records the remote file.
+type afterCopyBackend struct {
+	tierTimestampTestBackend
+	afterCopy func()
+}
+
+func (b *afterCopyBackend) CopyFile(file *os.File, fn func(progressed int64, percentage float32) error, concurrency int) (string, int64, error) {
+	key, size, err := b.tierTimestampTestBackend.CopyFile(file, fn, concurrency)
+	if err == nil && b.afterCopy != nil {
+		b.afterCopy()
+	}
+	return key, size, err
+}
+
+// A compaction committed after the tier upload copied the .dat, but before the
+// upload switched the volume to the remote object, leaves that object with the
+// old layout and the volume with the compacted index. The upload must not
+// attach it.
+func TestTierUploadRefusesADatReplacedByACompactionCommit(t *testing.T) {
+	root := t.TempDir()
+	b := &afterCopyBackend{tierTimestampTestBackend: tierTimestampTestBackend{root: root}}
+	backend.BackendStorages[tierTimestampTestBackendName] = b
+	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, "000", "", 0, needle.Version3, 0, types.HardDriveType, 0); err != nil {
+		t.Fatalf("add volume: %v", err)
+	}
+	written := map[uint64][]byte{}
+	for i := uint64(1); i <= 10; i++ {
+		n := &needle.Needle{Id: types.Uint64ToNeedleId(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)
+		}
+		written[i] = n.Data
+	}
+	for i := uint64(1); i <= 5; i++ {
+		if _, err := store.DeleteVolumeNeedle(1, &needle.Needle{Id: types.Uint64ToNeedleId(i)}); err != nil {
+			t.Fatalf("delete %d: %v", i, err)
+		}
+	}
+	v := store.GetVolume(1)
+	if err := v.CompactByIndex(nil); err != nil {
+		t.Fatalf("compact: %v", err)
+	}
+	if err := store.MarkVolumeReadonly(1, false, false); err != nil {
+		t.Fatalf("mark read-only: %v", err)
+	}
+
+	committed := false
+	b.afterCopy = func() {
+		if err := v.CommitCompact(); err != nil {
+			t.Errorf("commit: %v", err)
+			return
+		}
+		committed = true
+	}
+	vs := &VolumeServer{store: store}
+	err := vs.VolumeTierMoveDatToRemote(&volume_server_pb.VolumeTierMoveDatToRemoteRequest{
+		VolumeId: 1, DestinationBackendName: tierTimestampTestBackendName,
+	}, &discardServerStream[volume_server_pb.VolumeTierMoveDatToRemoteResponse]{})
+	if !committed {
+		t.Fatal("the compaction did not commit between the copy and the attach") // positive control
+	}
+	if err == nil {
+		t.Errorf("the upload attached a remote copy of a .dat that a compaction had replaced")
+	}
+	if v.HasRemoteFile() {
+		t.Errorf("the volume is tiered to the stale remote copy")
+	}
+	if entries, _ := os.ReadDir(root); len(entries) != 0 {
+		t.Errorf("the abandoned remote copy was left behind: %v", entries)
+	}
+	if !util.FileExists(filepath.Join(v.FileName(".dat"))) {
+		t.Fatalf("the local .dat is gone")
+	}
+	for i := uint64(6); i <= 10; i++ {
+		n := &needle.Needle{Id: types.Uint64ToNeedleId(i), Cookie: 7}
+		if _, err := store.ReadVolumeNeedle(1, n, nil, nil); err != nil {
+			t.Errorf("needle %d does not read after the upload: %v", i, err)
+		} else if !bytes.Equal(n.Data, written[i]) {
+			t.Errorf("needle %d reads different data", i)
+		}
+	}
+}

weed/storage/volume_tier.go

@@ -1,6 +1,7 @@
 package storage
 
 import (
+	"errors"
 	"fmt"
 
 	"github.com/seaweedfs/seaweedfs/weed/glog"
@@ -142,6 +143,41 @@ func (v *Volume) loadRemoteFileLocked() error {
 	return nil
 }
 
+// AttachRemoteDatFile records remoteFile, a copy of the .dat read from source,
+// in the volume's .vif and switches the data backend to it. It refuses when
+// source is no longer the volume's data backend: a compaction commit since the
+// copy replaced the .dat and the index, and the copy has the old layout. The
+// check, the .vif write and the switch happen under the data-file locks, so a
+// commit either lands before (and is seen here) or after (and refuses a tiered
+// volume).
+func (v *Volume) AttachRemoteDatFile(source backend.BackendStorageFile, remoteFile *volume_server_pb.RemoteFile) error {
+	v.dataFileSyncLock.Lock()
+	defer v.dataFileSyncLock.Unlock()
+	v.dataFileAccessLock.Lock()
+	defer v.dataFileAccessLock.Unlock()
+	if v.DataBackend != source {
+		return fmt.Errorf("volume %d data file changed during the copy (compacted or reloaded)", v.Id)
+	}
+	files := v.volumeInfo.GetFiles()
+	v.volumeInfo.Files = append(files, remoteFile)
+	if err := v.SaveVolumeInfo(); err != nil {
+		// A NotCrashDurableError means the rename committed: the .vif names
+		// the copy, so keep the entry and serve from the copy, and still
+		// report the error so the caller keeps the local .dat. Any other
+		// failure left the old .vif in place.
+		var ndErr *volume_info.NotCrashDurableError
+		if !errors.As(err, &ndErr) {
+			v.volumeInfo.Files = files
+			return fmt.Errorf("save remote file info: %w", err)
+		}
+		if loadErr := v.loadRemoteFileLocked(); loadErr != nil {
+			return fmt.Errorf("save remote file info: %w; load remote file: %v", err, loadErr)
+		}
+		return fmt.Errorf("save remote file info: %w", err)
+	}
+	return v.loadRemoteFileLocked()
+}
+
 func (v *Volume) SaveVolumeInfo() error {
 
 	tierFileName := v.FileName(".vif")
@@ -149,6 +185,9 @@ func (v *Volume) SaveVolumeInfo() error {
 		v.volumeInfo.ExpireAtSec = expireAtSec
 	}
 
-	return volume_info.SaveVolumeInfo(tierFileName, v.volumeInfo)
+	return saveVolumeInfoFile(tierFileName, v.volumeInfo)
 
 }
+
+// saveVolumeInfoFile writes a .vif; tests replace it to inject failures.
+var saveVolumeInfoFile = volume_info.SaveVolumeInfo

weed/storage/volume_tier_attach_test.go

@@ -0,0 +1,70 @@
+package storage
+
+import (
+	"errors"
+	"syscall"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
+	"github.com/seaweedfs/seaweedfs/weed/storage/backend"
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
+	"github.com/seaweedfs/seaweedfs/weed/storage/types"
+	"github.com/seaweedfs/seaweedfs/weed/storage/volume_info"
+)
+
+// When the .vif rename commits but the directory fsync fails, the saved .vif
+// names the remote copy. AttachRemoteDatFile must keep that entry (the upload
+// handler deletes a copy the volume info does not name) and serve from the
+// copy the .vif points at, while still reporting the error.
+func TestAttachRemoteDatFileKeepsACommittedVifEntry(t *testing.T) {
+	b := newLocalDirBackend(t)
+	registerTestBackend(t, b)
+
+	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()
+	for i := 1; i <= 5; i++ {
+		_, _, _, err := v.writeNeedle2(newRandomNeedle(uint64(i)), true, false, false)
+		require.NoError(t, err)
+	}
+	diskFile, ok := v.DataBackend.(*backend.DiskFile)
+	require.True(t, ok)
+	key, size, err := b.CopyFile(diskFile.File, nil, 0)
+	require.NoError(t, err)
+
+	saved := saveVolumeInfoFile
+	saveVolumeInfoFile = func(fileName string, info *volume_server_pb.VolumeInfo) error {
+		if err := saved(fileName, info); err != nil {
+			return err
+		}
+		return &volume_info.NotCrashDurableError{FileName: fileName, Err: syscall.EIO}
+	}
+	defer func() { saveVolumeInfoFile = saved }()
+
+	bType, bId := backend.BackendNameToTypeId(testBackendName)
+	err = v.AttachRemoteDatFile(diskFile, &volume_server_pb.RemoteFile{
+		BackendType: bType, BackendId: bId, Key: key, FileSize: uint64(size),
+		ModifiedTime: uint64(time.Now().Unix()), Extension: ".dat",
+	})
+	var ndErr *volume_info.NotCrashDurableError
+	require.True(t, errors.As(err, &ndErr), "the fsync failure is still reported: %v", err)
+
+	onDisk, _, _, err := volume_info.MaybeLoadVolumeInfo(v.FileName(".vif"))
+	require.NoError(t, err)
+	require.Len(t, onDisk.GetFiles(), 1)
+	require.Equal(t, key, onDisk.GetFiles()[0].Key)
+	require.Len(t, v.GetVolumeInfo().GetFiles(), 1, "the in-memory entry was rolled back while the .vif names the copy")
+	require.Equal(t, key, v.GetVolumeInfo().GetFiles()[0].Key)
+	require.True(t, v.HasRemoteFile())
+	_, isDisk := v.DataBackend.(*backend.DiskFile)
+	require.False(t, isDisk, "the live backend must be the copy the .vif names")
+
+	n := &needle.Needle{Id: types.Uint64ToNeedleId(3)}
+	_, err = v.readNeedle(n, &ReadOption{}, nil)
+	require.NoError(t, err)
+}

weed/storage/volume_vacuum.go

@@ -62,6 +62,14 @@ func (v *Volume) garbageLevel() float64 {
 	return float64(deletedSize) / float64(fileSize)
 }
 
+// errTieredVolumeCompact refuses a compaction of a volume whose .dat is in the
+// remote tier: the copy reads the local .dat, which tiering removes (or leaves
+// behind as a stale copy), and the reload after a commit would read the remote
+// object through the compacted index.
+func errTieredVolumeCompact(vid needle.VolumeId) error {
+	return fmt.Errorf("volume %d is tiered to remote storage, cannot compact", vid)
+}
+
 type CompactOptions struct {
 	PreallocateBytes  int64
 	MaxBytesPerSecond int64
@@ -87,6 +95,9 @@ func (v *Volume) CompactByVolumeData(opts *CompactOptions) error {
 		return nil
 	}
 	glog.V(3).Infof("Compacting volume %d ...", v.Id)
+	if v.HasRemoteFile() {
+		return errTieredVolumeCompact(v.Id)
+	}
 	//no need to lock for copy on write
 	//v.accessLock.Lock()
 	//defer v.accessLock.Unlock()
@@ -132,6 +143,9 @@ func (v *Volume) CompactByIndex(opts *CompactOptions) error {
 		return nil
 	}
 	glog.V(3).Infof("Compact2 volume %d ...", v.Id)
+	if v.HasRemoteFile() {
+		return errTieredVolumeCompact(v.Id)
+	}
 
 	if !v.isCompactionInProgress.CompareAndSwap(false, true) {
 		glog.V(0).Infof("Volume %d is already compacting2 ...", v.Id)
@@ -183,6 +197,21 @@ func (v *Volume) CommitCompact() error {
 	defer v.dataFileAccessLock.Unlock()
 
 	glog.V(3).Infof("Got volume %d committing lock...", v.Id)
+	if v.HasRemoteFile() {
+		// The volume was tiered after the copy started: the reload would read
+		// the remote object through the compacted index. Drop the copy and its
+		// temporary needle map; no later vacuum of this volume would.
+		if v.tmpNm != nil {
+			v.tmpNm.Close()
+			v.tmpNm = nil
+		}
+		for _, ext := range []string{".cpd", ".cpx", ".cpldb"} {
+			if e := os.RemoveAll(v.FileName(ext)); e != nil {
+				glog.V(0).Infof("remove %s: %v", v.FileName(ext), e)
+			}
+		}
+		return errTieredVolumeCompact(v.Id)
+	}
 	if v.nm != nil {
 		v.nm.Close()
 		v.nm = nil

weed/storage/volume_vacuum_tiered_test.go

@@ -0,0 +1,111 @@
+package storage
+
+import (
+	"bytes"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
+	"github.com/seaweedfs/seaweedfs/weed/storage/backend"
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
+	"github.com/seaweedfs/seaweedfs/weed/storage/types"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+)
+
+// A tiered volume that accepted deletes has garbage, but its .dat is remote:
+// the compaction must be refused before any copy, and the store must not
+// report the volume as a vacuum candidate.
+func TestCompactRefusesTieredVolume(t *testing.T) {
+	b := newLocalDirBackend(t)
+	registerTestBackend(t, b)
+	dir := t.TempDir()
+	const vid = needle.VolumeId(34)
+	v, _ := tierUpVolumeLive(t, dir, vid, b)
+	defer v.Close()
+	require.True(t, v.HasRemoteFile())
+	for i := 1; i <= 4; i++ {
+		_, err := v.deleteNeedle2(newRandomNeedle(uint64(i)))
+		require.NoError(t, err)
+	}
+	garbage := v.garbageLevel()
+	require.Greater(t, garbage, 0.0, "precondition: the deletes left garbage")
+
+	require.ErrorContains(t, v.CompactByIndex(nil), "tiered")
+	require.ErrorContains(t, v.CompactByVolumeData(nil), "tiered")
+	require.False(t, util.FileExists(v.FileName(".cpd")))
+	require.False(t, util.FileExists(v.FileName(".cpx")))
+
+	loc := &DiskLocation{Directory: dir, volumes: map[needle.VolumeId]*Volume{}}
+	loc.SetVolume(vid, v)
+	s := &Store{Locations: []*DiskLocation{loc}}
+	ratio, _, err := s.CheckCompactVolume(vid)
+	require.NoError(t, err)
+	// volume.tier.compact reads this ratio while the volume is still remote to
+	// decide whether to download and compact it; vacuum selection skips tiered
+	// volumes by their remote storage name instead.
+	require.Equal(t, garbage, ratio, "a tiered volume reports its real garbage ratio")
+}
+
+// A volume tiered while its compaction copy ran (keeping the local .dat) must
+// not commit: the reload would read the remote object, which has the old
+// layout, through the compacted index.
+func TestCommitCompactRefusesVolumeTieredDuringCopy(t *testing.T) {
+	t.Run("memory", func(t *testing.T) { testCommitCompactRefusesVolumeTieredDuringCopy(t, NeedleMapInMemory) })
+	t.Run("leveldb", func(t *testing.T) { testCommitCompactRefusesVolumeTieredDuringCopy(t, NeedleMapLevelDb) })
+}
+
+func testCommitCompactRefusesVolumeTieredDuringCopy(t *testing.T, kind NeedleMapKind) {
+	b := newLocalDirBackend(t)
+	registerTestBackend(t, b)
+	dir := t.TempDir()
+	v, err := NewVolume(dir, dir, "", 35, kind, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0)
+	require.NoError(t, err)
+	defer v.Close()
+
+	written := map[uint64][]byte{}
+	for i := uint64(1); i <= 6; i++ {
+		n := newRandomNeedle(i)
+		n.Cookie = 7
+		_, _, _, err := v.writeNeedle2(n, true, false, false)
+		require.NoError(t, err)
+		written[i] = n.Data
+	}
+	for i := uint64(1); i <= 3; i++ {
+		_, err := v.deleteNeedle2(newEmptyNeedle(i))
+		require.NoError(t, err)
+	}
+	require.NoError(t, v.CompactByIndex(nil))
+	require.True(t, util.FileExists(v.FileName(".cpd")))
+	require.NotNil(t, v.tmpNm, "precondition: the copy left its temporary needle map")
+	if kind == NeedleMapLevelDb {
+		require.True(t, util.FileExists(v.FileName(".cpldb")), "precondition: the copy left its temporary LevelDB")
+	}
+
+	// tier the volume up, keeping the local .dat
+	diskFile, ok := v.DataBackend.(*backend.DiskFile)
+	require.True(t, ok)
+	key, size, err := b.CopyFile(diskFile.File, nil, 0)
+	require.NoError(t, err)
+	bType, bId := backend.BackendNameToTypeId(testBackendName)
+	v.GetVolumeInfo().Files = append(v.GetVolumeInfo().GetFiles(), &volume_server_pb.RemoteFile{
+		BackendType: bType, BackendId: bId, Key: key, FileSize: uint64(size),
+		ModifiedTime: uint64(time.Now().Unix()), Extension: ".dat",
+	})
+	require.NoError(t, v.SaveVolumeInfo())
+	require.NoError(t, v.LoadRemoteFile())
+
+	require.ErrorContains(t, v.CommitCompact(), "tiered")
+	require.False(t, util.FileExists(v.FileName(".cpd")))
+	require.False(t, util.FileExists(v.FileName(".cpx")))
+	require.False(t, util.FileExists(v.FileName(".cpldb")), "the temporary LevelDB is left on disk")
+	require.Nil(t, v.tmpNm, "the temporary needle map is kept")
+	for i := uint64(4); i <= 6; i++ {
+		got := &needle.Needle{Id: types.Uint64ToNeedleId(i), Cookie: 7}
+		_, err := v.readNeedle(got, nil, nil)
+		require.NoError(t, err, "needle %d", i)
+		require.True(t, bytes.Equal(written[i], got.Data), "needle %d data", i)
+	}
+}

weed/topology/topology_vacuum.go

@@ -39,6 +39,13 @@ func (t *Topology) batchVacuumVolumeCheck(grpcDialOption grpc.DialOption, vid ne
 	defer cancel()
 	for index, dn := range locationlist.list {
 		go func(index int, dn *DataNode, url pb.ServerAddress, vid needle.VolumeId) {
+			// A replica tiered to remote storage cannot be compacted in place,
+			// whatever its garbage ratio; volume.tier.compact reclaims it.
+			if v, lookErr := dn.GetVolumesById(vid); lookErr == nil && v.IsRemote() {
+				glog.V(1).Infof("skip vacuuming volume %d on %s: tiered to %s", vid, url, v.RemoteStorageName)
+				ch <- -1
+				return
+			}
 			err := operation.WithVolumeServerClient(false, url, grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
 				resp, err := volumeServerClient.VacuumVolumeCheck(ctx, &volume_server_pb.VacuumVolumeCheckRequest{
 					VolumeId: uint32(vid),

weed/topology/topology_vacuum_test.go

@@ -481,3 +481,52 @@ func TestVacuumStalledVolumeServerReleasesGuard(t *testing.T) {
 		})
 	}
 }
+
+// A replica tiered to remote storage is never compacted by a vacuum, not even
+// on an explicit request with a zero threshold: its .dat is the remote object.
+// volume.tier.compact is the way to reclaim its garbage.
+func TestVacuumSkipsTieredReplica(t *testing.T) {
+	fake := &fakeVacuumServer{
+		checks: map[uint32]*volume_server_pb.VacuumVolumeCheckResponse{
+			1: {GarbageRatio: 0.9},
+			2: {GarbageRatio: 0.9},
+		},
+	}
+	grpcPort, dialOption := startFakeVolumeServer(t, fake)
+
+	topo := NewTopology("weedfs", sequence.NewMemorySequencer(), 32*1024, 5, false)
+	dn := topo.GetOrCreateDataCenter("dc1").GetOrCreateRack("rack1").
+		GetOrCreateDataNode("127.0.0.1", 8080, grpcPort, "127.0.0.1", "dn", map[string]uint32{"": 10})
+	newVolumeInfo := func(vid needle.VolumeId, remote string) storage.VolumeInfo {
+		return storage.VolumeInfo{
+			Id:                vid,
+			Size:              1 << 20,
+			Collection:        "c",
+			ReadOnly:          remote != "",
+			ModifiedAtSecond:  time.Now().Unix(),
+			Version:           needle.GetCurrentVersion(),
+			ReplicaPlacement:  &super_block.ReplicaPlacement{},
+			Ttl:               needle.EMPTY_TTL,
+			RemoteStorageName: remote,
+		}
+	}
+	vl := topo.GetVolumeLayout("c", &super_block.ReplicaPlacement{}, needle.EMPTY_TTL, types.ToDiskType(""))
+	volumes := []storage.VolumeInfo{newVolumeInfo(1, "s3.default"), newVolumeInfo(2, "")}
+	dn.UpdateVolumes(volumes)
+	for _, v := range volumes {
+		topo.RegisterVolumeLayout(v, dn)
+	}
+	c := NewCollection("c", topo.volumeSizeLimit, false)
+	for _, vid := range []needle.VolumeId{1, 2} {
+		vl.accessLock.RLock()
+		ll := vl.vid2location[vid].Copy()
+		vl.accessLock.RUnlock()
+		topo.vacuumOneVolumeId(dialOption, vl, c, 0, ll, vid, 0, false)
+	}
+
+	fake.mu.Lock()
+	defer fake.mu.Unlock()
+	if len(fake.committed) != 1 || fake.committed[0] != 2 {
+		t.Fatalf("committed %v, want only the local volume 2", fake.committed) // volume 2 is the positive control
+	}
+}

weed/worker/tasks/vacuum/detection.go

@@ -109,10 +109,14 @@ func Detection(metrics []*types.VolumeHealthMetrics, clusterInfo *types.ClusterI
 }
 
 // pickPrimaryReplica returns the eligible replica with the highest garbage
-// ratio. Returns nil if no replica satisfies both gates.
+// ratio. Returns nil if no replica satisfies both gates. Replicas tiered to
+// remote storage are skipped: the volume server cannot compact them.
 func pickPrimaryReplica(replicas []*types.VolumeHealthMetrics, garbageThreshold float64, minAge time.Duration) *types.VolumeHealthMetrics {
 	var best *types.VolumeHealthMetrics
 	for _, m := range replicas {
+		if m.HasRemoteCopy {
+			continue
+		}
 		if m.GarbageRatio < garbageThreshold {
 			continue
 		}
@@ -144,7 +148,7 @@ func recordSkipReasons(replicas []*types.VolumeHealthMetrics, garbageThreshold f
 }
 
 // createVacuumTaskParams creates typed parameters for a vacuum task whose
-// Sources list contains every replica of the volume. Worker-side
+// Sources list contains every non-tiered replica of the volume. Worker-side
 // performVacuum iterates over Sources and vacuums each replica.
 func createVacuumTaskParams(task *types.TaskDetectionResult, replicas []*types.VolumeHealthMetrics, vacuumConfig *Config, clusterInfo *types.ClusterInfo) *worker_pb.TaskParams {
 	garbageThreshold := 0.3
@@ -164,6 +168,9 @@ func createVacuumTaskParams(task *types.TaskDetectionResult, replicas []*types.V
 	sources := make([]*worker_pb.TaskSource, 0, len(replicas))
 	seen := make(map[string]struct{}, len(replicas))
 	for _, m := range replicas {
+		if m.HasRemoteCopy {
+			continue
+		}
 		address, err := util.ResolveServerAddress(m.Server, clusterInfo.ActiveTopology)
 		if err != nil {
 			glog.Warningf("Failed to resolve address for server %s for vacuum task on volume %d, dropping replica: %v", m.Server, task.VolumeID, err)

weed/worker/tasks/vacuum/detection_tiered_test.go

@@ -0,0 +1,25 @@
+package vacuum
+
+import (
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/worker/types"
+)
+
+func TestPickPrimaryReplicaSkipsTieredReplicas(t *testing.T) {
+	m := func(id uint32, server string, remote bool) *types.VolumeHealthMetrics {
+		return &types.VolumeHealthMetrics{VolumeID: id, Server: server, GarbageRatio: 0.9, Age: 72 * time.Hour,
+			HasRemoteCopy: remote, IsReadOnly: remote}
+	}
+	if p := pickPrimaryReplica([]*types.VolumeHealthMetrics{m(1, "a:8080", false)}, 0.3, 24*time.Hour); p == nil {
+		t.Fatal("a local volume at 90% garbage must be picked")
+	}
+	if p := pickPrimaryReplica([]*types.VolumeHealthMetrics{m(2, "a:8080", true)}, 0.3, 24*time.Hour); p != nil {
+		t.Fatalf("tiered replica of volume %d picked for vacuum", p.VolumeID)
+	}
+	p := pickPrimaryReplica([]*types.VolumeHealthMetrics{m(3, "a:8080", true), m(3, "b:8080", false)}, 0.3, 24*time.Hour)
+	if p == nil || p.Server != "b:8080" {
+		t.Fatalf("partially tiered volume should pick its local replica, got %+v", p)
+	}
+}