feat(master): drain pending size before marking volume readonly (#9036)

* feat(master): drain pending size before marking volume readonly

When vacuum, volume move, or EC encoding marks a volume readonly,
in-flight assigned bytes may still be pending. This adds a drain step:
immediately remove from writable list (stop new assigns), then wait
for pending to decay below 4MB or 30s timeout.

- Add volumeSizeTracking struct consolidating effectiveSize,
  reportedSize, and compactRevision into a single map
- Add GetPendingSize, waitForPendingDrain, DrainAndRemoveFromWritable,
  DrainAndSetVolumeReadOnly to VolumeLayout
- UpdateVolumeSize detects compaction via compactRevision change and
  resets effectiveSize instead of decaying
- Wire drain into vacuum (topology_vacuum.go) and volume mark readonly
  (master_grpc_server_volume.go)

* fix: use 2MB pending size drain threshold

* fix: check crowded state on initial UpdateVolumeSize registration

* fix: respect context cancellation in drain, relax test timing

- DrainAndSetVolumeReadOnly now accepts context.Context and returns
  early on cancellation (for gRPC handler timeout/cancel)
- waitForPendingDrain uses select on ctx.Done instead of time.Sleep
- Increase concurrent heartbeat test timeout from 10s to 15s for CI

* fix: use time-based dedup so decay runs even when reported size is unchanged

The value-based dedup (same reportedSize + compactRevision = skip) prevented
decay from running when pending bytes existed but no writes had landed on
disk yet. The reported size stayed the same across heartbeats, so the excess
never decayed.

Fix: dedup replicas within the same heartbeat cycle using a 2-second time
window instead of comparing values. This allows decay to run once per
heartbeat cycle even when the reported size is unchanged.

Also confirmed finding 1 (draining re-add race) is a false positive:
- Vacuum: ensureCorrectWritables only runs for ReadOnly-changed volumes
- Move/EC: readonlyVolumes flag prevents re-adding during drain

* fix: make VolumeMarkReadonly non-blocking to fix EC integration test timeout

The DrainAndSetVolumeReadOnly call in VolumeMarkReadonly gRPC blocked up
to 30s waiting for pending bytes to decay. In integration tests (and
real clusters during EC encoding), this caused timeouts because multiple
volumes are marked readonly sequentially and heartbeats may not arrive
fast enough to decay pending within the drain window.

Fix: VolumeMarkReadonly now calls SetVolumeReadOnly immediately (stops
new assigns) and only logs a warning if pending bytes remain. The drain
wait is kept only for vacuum (DrainAndRemoveFromWritable) which runs
inside the master's own goroutine pool.

Remove DrainAndSetVolumeReadOnly as it's no longer used.

* fix: relax test timing, rename test, add post-condition assert

* test: add vacuum integration tests with CI workflow

Full-cluster integration test for vacuum, modeled on the EC integration
tests. Starts a real master + 2 volume servers, uploads data, deletes
entries to create garbage, runs volume.vacuum via shell command, and
verifies garbage cleanup and data integrity.

Test flow:
1. Start cluster (master + 2 volume servers)
2. Upload 10 files to create volume with data
3. Delete 5 files to create ~50% garbage
4. Verify garbage ratio > 10%
5. Run volume.vacuum command
6. Verify garbage cleaned up
7. Verify remaining 5 files are still accessible

CI workflow runs on push/PR to master with 15-minute timeout.
Log collection on failure via artifact upload.

* fix: use 500KB files and delete 75% to exceed vacuum garbage threshold

* fix: add shell lock before vacuum command, fix compilation error

* fix: strengthen vacuum integration test assertions

- waitForServer: use net.DialTimeout instead of grpc.NewClient for
  real TCP readiness check
- verify_garbage_before_vacuum: t.Fatal instead of warning when no
  garbage detected
- verify_cleanup_after_vacuum: t.Fatal if no server reported the
  volume or cleanup wasn't verified
- verify_remaining_data: read actual file contents via HTTP and
  compare byte-for-byte against original uploaded payloads

* fix: use http.Client with timeout and close body before retry

.github/workflows/vacuum-integration-tests.yml

@@ -0,0 +1,56 @@
+name: "Vacuum Integration Tests"
+
+on:
+  push:
+    branches: [ master ]
+  pull_request:
+    branches: [ master ]
+
+permissions:
+  contents: read
+
+jobs:
+  vacuum-integration-tests:
+    name: Vacuum Integration Tests
+    runs-on: ubuntu-22.04
+    timeout-minutes: 15
+    steps:
+    - name: Set up Go 1.x
+      uses: actions/setup-go@v6
+      with:
+        go-version: ^1.25
+      id: go
+
+    - name: Check out code into the Go module directory
+      uses: actions/checkout@v6
+
+    - name: Build weed binary
+      run: |
+        cd weed && go build -o weed .
+
+    - name: Run Vacuum Integration Tests
+      working-directory: test/vacuum
+      run: |
+        go test -v -timeout 10m
+
+    - name: Collect server logs on failure
+      if: failure()
+      run: |
+        echo "Collecting server logs from temp directories..."
+        mkdir -p /tmp/vacuum-test-logs
+        find /tmp -maxdepth 1 -type d -name "TestVacuum*" 2>/dev/null | while read dir; do
+          if [ -d "$dir" ]; then
+            echo "Found test directory: $dir"
+            cp -r "$dir" /tmp/vacuum-test-logs/ 2>/dev/null || true
+          fi
+        done
+        echo "Collected logs:"
+        find /tmp/vacuum-test-logs -type f -name "*.log" 2>/dev/null || echo "No logs found"
+
+    - name: Archive logs
+      if: failure()
+      uses: actions/upload-artifact@v7
+      with:
+        name: vacuum-integration-test-logs
+        path: /tmp/vacuum-test-logs/
+        retention-days: 14

test/vacuum/vacuum_integration_test.go

@@ -0,0 +1,402 @@
+package vacuum
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/operation"
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
+	"github.com/seaweedfs/seaweedfs/weed/shell"
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+)
+
+type TestCluster struct {
+	masterCmd     *exec.Cmd
+	volumeServers []*exec.Cmd
+}
+
+func (c *TestCluster) Stop() {
+	for _, cmd := range c.volumeServers {
+		if cmd != nil && cmd.Process != nil {
+			cmd.Process.Kill()
+			cmd.Wait()
+		}
+	}
+	if c.masterCmd != nil && c.masterCmd.Process != nil {
+		c.masterCmd.Process.Kill()
+		c.masterCmd.Wait()
+	}
+}
+
+func startCluster(ctx context.Context, dataDir string) (*TestCluster, error) {
+	weedBinary := findWeedBinary()
+	if weedBinary == "" {
+		return nil, fmt.Errorf("weed binary not found - build with 'cd weed && go build' first")
+	}
+
+	cluster := &TestCluster{}
+
+	masterDir := filepath.Join(dataDir, "master")
+	os.MkdirAll(masterDir, 0755)
+
+	// Empty security.toml to disable JWT in tests
+	os.WriteFile(filepath.Join(dataDir, "security.toml"), []byte("# test\n"), 0644)
+
+	// Start master
+	masterCmd := exec.CommandContext(ctx, weedBinary, "master",
+		"-port", "9333",
+		"-mdir", masterDir,
+		"-volumeSizeLimitMB", "10",
+		"-ip", "127.0.0.1",
+	)
+	masterCmd.Dir = dataDir
+	masterLog, _ := os.Create(filepath.Join(masterDir, "master.log"))
+	masterCmd.Stdout = masterLog
+	masterCmd.Stderr = masterLog
+	if err := masterCmd.Start(); err != nil {
+		return nil, fmt.Errorf("start master: %v", err)
+	}
+	cluster.masterCmd = masterCmd
+	time.Sleep(2 * time.Second)
+
+	// Start 2 volume servers (enough for vacuum testing)
+	for i := 0; i < 2; i++ {
+		volumeDir := filepath.Join(dataDir, fmt.Sprintf("volume%d", i))
+		os.MkdirAll(volumeDir, 0755)
+
+		port := fmt.Sprintf("808%d", i)
+		volumeCmd := exec.CommandContext(ctx, weedBinary, "volume",
+			"-port", port,
+			"-dir", volumeDir,
+			"-max", "10",
+			"-master", "127.0.0.1:9333",
+			"-ip", "127.0.0.1",
+		)
+		volumeCmd.Dir = dataDir
+		volumeLog, _ := os.Create(filepath.Join(volumeDir, "volume.log"))
+		volumeCmd.Stdout = volumeLog
+		volumeCmd.Stderr = volumeLog
+		if err := volumeCmd.Start(); err != nil {
+			cluster.Stop()
+			return nil, fmt.Errorf("start volume server %d: %v", i, err)
+		}
+		cluster.volumeServers = append(cluster.volumeServers, volumeCmd)
+	}
+
+	time.Sleep(5 * time.Second)
+	return cluster, nil
+}
+
+func findWeedBinary() string {
+	candidates := []string{
+		"../../weed/weed",
+		"../weed/weed",
+		"./weed",
+	}
+	for _, c := range candidates {
+		if _, err := os.Stat(c); err == nil {
+			if abs, err := filepath.Abs(c); err == nil {
+				return abs
+			}
+			return c
+		}
+	}
+	if path, err := exec.LookPath("weed"); err == nil {
+		return path
+	}
+	return ""
+}
+
+func waitForServer(address string, timeout time.Duration) error {
+	start := time.Now()
+	for time.Since(start) < timeout {
+		if conn, err := net.DialTimeout("tcp", address, 1*time.Second); err == nil {
+			conn.Close()
+			return nil
+		}
+		time.Sleep(500 * time.Millisecond)
+	}
+	return fmt.Errorf("timeout waiting for server %s", address)
+}
+
+func uploadData(masterAddr, collection string, data []byte) (string, needle.VolumeId, error) {
+	assignResult, err := operation.Assign(context.Background(), func(ctx context.Context) pb.ServerAddress {
+		return pb.ServerAddress(masterAddr)
+	}, grpc.WithInsecure(), &operation.VolumeAssignRequest{
+		Count:      1,
+		Collection: collection,
+	})
+	if err != nil {
+		return "", 0, fmt.Errorf("assign: %v", err)
+	}
+
+	uploader, err := operation.NewUploader()
+	if err != nil {
+		return "", 0, fmt.Errorf("new uploader: %v", err)
+	}
+
+	uploadResult, err, _ := uploader.Upload(context.Background(), bytes.NewReader(data), &operation.UploadOption{
+		UploadUrl: "http://" + assignResult.Url + "/" + assignResult.Fid,
+		Filename:  "testfile.txt",
+		MimeType:  "text/plain",
+	})
+	if err != nil {
+		return "", 0, fmt.Errorf("upload: %v", err)
+	}
+	if uploadResult.Error != "" {
+		return "", 0, fmt.Errorf("upload error: %s", uploadResult.Error)
+	}
+
+	fid, err := needle.ParseFileIdFromString(assignResult.Fid)
+	if err != nil {
+		return "", 0, err
+	}
+	return assignResult.Fid, fid.VolumeId, nil
+}
+
+func deleteFile(masterAddr string, fid string) error {
+	results := operation.DeleteFileIds(func(ctx context.Context) pb.ServerAddress {
+		return pb.ServerAddress(masterAddr)
+	}, false, grpc.WithInsecure(), []string{fid})
+	for _, r := range results {
+		if r.Error != "" {
+			return fmt.Errorf("delete %s: %s", fid, r.Error)
+		}
+	}
+	return nil
+}
+
+func getGarbageRatio(volumeServerAddr string, volumeId uint32) (float64, error) {
+	var ratio float64
+	err := operation.WithVolumeServerClient(false, pb.ServerAddress(volumeServerAddr), grpc.WithInsecure(),
+		func(client volume_server_pb.VolumeServerClient) error {
+			resp, err := client.VacuumVolumeCheck(context.Background(), &volume_server_pb.VacuumVolumeCheckRequest{
+				VolumeId: volumeId,
+			})
+			if err != nil {
+				return err
+			}
+			ratio = resp.GarbageRatio
+			return nil
+		})
+	return ratio, err
+}
+
+// TestVacuumIntegration tests the full vacuum flow:
+// upload data → delete some → verify garbage → vacuum → verify cleanup
+func TestVacuumIntegration(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping integration test in short mode")
+	}
+
+	testDir := t.TempDir()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second)
+	defer cancel()
+
+	cluster, err := startCluster(ctx, testDir)
+	require.NoError(t, err)
+	defer cluster.Stop()
+
+	require.NoError(t, waitForServer("127.0.0.1:9333", 30*time.Second))
+	require.NoError(t, waitForServer("127.0.0.1:8080", 30*time.Second))
+	require.NoError(t, waitForServer("127.0.0.1:8081", 30*time.Second))
+
+	masterAddr := "127.0.0.1:9333"
+	collection := "vactest"
+
+	// Upload files large enough that deleting most creates significant garbage.
+	// With volumeSizeLimitMB=10, we need several MB of garbage to exceed the
+	// 10% threshold passed to vacuum.
+	const fileSize = 500 * 1024 // 500 KB per file
+	const totalFiles = 16
+	const filesToDelete = 12 // delete 75% → ~6 MB garbage out of ~8 MB
+
+	var fids []string
+	var payloads [][]byte
+	var volumeId needle.VolumeId
+	for i := 0; i < totalFiles; i++ {
+		data := bytes.Repeat([]byte{byte('A' + i%26)}, fileSize)
+		fid, vid, err := uploadData(masterAddr, collection, data)
+		require.NoError(t, err, "upload %d", i)
+		fids = append(fids, fid)
+		payloads = append(payloads, data)
+		volumeId = vid
+	}
+	t.Logf("Uploaded %d files (%d KB each) to volume %d", totalFiles, fileSize/1024, volumeId)
+
+	// Wait for heartbeat to report sizes
+	time.Sleep(6 * time.Second)
+
+	// Delete most files to create garbage well above the threshold
+	for i := 0; i < filesToDelete; i++ {
+		err := deleteFile(masterAddr, fids[i])
+		require.NoError(t, err, "delete %s", fids[i])
+	}
+	t.Logf("Deleted %d of %d files to create garbage", filesToDelete, totalFiles)
+
+	// Wait for heartbeat to report deletions
+	time.Sleep(6 * time.Second)
+
+	// Verify garbage exists
+	t.Run("verify_garbage_before_vacuum", func(t *testing.T) {
+		for _, addr := range []string{"127.0.0.1:8080", "127.0.0.1:8081"} {
+			ratio, err := getGarbageRatio(addr, uint32(volumeId))
+			if err != nil {
+				continue
+			}
+			t.Logf("Garbage ratio on %s: %.2f%%", addr, ratio*100)
+			if ratio > 0.1 {
+				return // sufficient garbage found
+			}
+		}
+		t.Fatal("No server reported garbage > 10% — test data setup failed")
+	})
+
+	// Execute vacuum via shell command
+	t.Run("run_vacuum", func(t *testing.T) {
+		options := &shell.ShellOptions{
+			Masters:        stringPtr(masterAddr),
+			GrpcDialOption: grpc.WithInsecure(),
+			FilerGroup:     stringPtr("default"),
+		}
+		commandEnv := shell.NewCommandEnv(options)
+
+		shellCtx, shellCancel := context.WithTimeout(context.Background(), 60*time.Second)
+		defer shellCancel()
+		go commandEnv.MasterClient.KeepConnectedToMaster(shellCtx)
+		commandEnv.MasterClient.WaitUntilConnected(shellCtx)
+		time.Sleep(2 * time.Second)
+
+		// Acquire lock (required by shell commands)
+		locked, unlock := tryLock(t, commandEnv, 30*time.Second)
+		require.True(t, locked, "could not acquire shell lock")
+		defer unlock()
+
+		// Find and execute vacuum command
+		var output bytes.Buffer
+		var found bool
+		var err error
+		for _, cmd := range shell.Commands {
+			if cmd.Name() == "volume.vacuum" {
+				err = cmd.Do(
+					[]string{"-garbageThreshold", "0.1", "-collection", collection},
+					commandEnv, &output,
+				)
+				found = true
+				break
+			}
+		}
+		require.True(t, found, "volume.vacuum command not found")
+		t.Logf("Vacuum output: %s", output.String())
+		require.NoError(t, err, "vacuum command failed")
+		t.Log("Vacuum completed successfully")
+	})
+
+	// Wait for vacuum effects to settle
+	time.Sleep(6 * time.Second)
+
+	// Verify garbage was cleaned
+	t.Run("verify_cleanup_after_vacuum", func(t *testing.T) {
+		var volumeFound, cleanupVerified bool
+		for _, addr := range []string{"127.0.0.1:8080", "127.0.0.1:8081"} {
+			ratio, err := getGarbageRatio(addr, uint32(volumeId))
+			if err != nil {
+				continue
+			}
+			volumeFound = true
+			t.Logf("Garbage ratio after vacuum on %s: %.2f%%", addr, ratio*100)
+			if ratio < 0.05 {
+				cleanupVerified = true
+			}
+		}
+		if !volumeFound {
+			t.Fatal("No server reported volume after vacuum")
+		}
+		if !cleanupVerified {
+			t.Fatal("Garbage was not cleaned up after vacuum")
+		}
+	})
+
+	// Verify remaining files are still readable with correct contents
+	t.Run("verify_remaining_data", func(t *testing.T) {
+		for i := filesToDelete; i < totalFiles; i++ {
+			fid := fids[i]
+			expected := payloads[i]
+
+			// Read file via HTTP from volume server
+			client := &http.Client{Timeout: 5 * time.Second}
+			url := fmt.Sprintf("http://127.0.0.1:8080/%s", fid)
+			resp, err := client.Get(url)
+			if err != nil || resp.StatusCode == http.StatusNotFound {
+				if resp != nil {
+					resp.Body.Close()
+				}
+				url = fmt.Sprintf("http://127.0.0.1:8081/%s", fid)
+				resp, err = client.Get(url)
+			}
+			require.NoError(t, err, "read fid %s", fid)
+			body, err := io.ReadAll(resp.Body)
+			resp.Body.Close()
+			require.NoError(t, err, "read body of fid %s", fid)
+			require.Equal(t, http.StatusOK, resp.StatusCode, "fid %s returned %d", fid, resp.StatusCode)
+			require.Equal(t, len(expected), len(body), "fid %s size mismatch", fid)
+			require.True(t, bytes.Equal(expected, body), "fid %s content mismatch", fid)
+			t.Logf("File %s verified (%d bytes)", fid, len(body))
+		}
+	})
+}
+
+func stringPtr(s string) *string {
+	return &s
+}
+
+func tryLock(t *testing.T, commandEnv *shell.CommandEnv, timeout time.Duration) (locked bool, unlock func()) {
+	t.Helper()
+	type result struct {
+		err error
+	}
+	done := make(chan result, 1)
+	go func() {
+		for _, cmd := range shell.Commands {
+			if cmd.Name() == "lock" {
+				var out bytes.Buffer
+				done <- result{err: cmd.Do([]string{}, commandEnv, &out)}
+				return
+			}
+		}
+		done <- result{err: fmt.Errorf("lock command not found")}
+	}()
+
+	select {
+	case res := <-done:
+		if res.err != nil {
+			t.Logf("lock failed: %v", res.err)
+			return false, nil
+		}
+		return true, func() {
+			for _, cmd := range shell.Commands {
+				if cmd.Name() == "unlock" {
+					var out bytes.Buffer
+					cmd.Do([]string{}, commandEnv, &out)
+					return
+				}
+			}
+		}
+	case <-time.After(timeout):
+		t.Log("lock timed out")
+		return false, nil
+	}
+}

weed/server/master_grpc_server_volume.go

@@ -336,7 +336,11 @@ func (ms *MasterServer) VolumeMarkReadonly(ctx context.Context, req *master_pb.V
 	for _, dn := range dataNodes {
 		if dn.Ip == req.Ip && dn.Port == int(req.Port) {
 			if req.IsReadonly {
-				vl.SetVolumeReadOnly(dn, needle.VolumeId(req.VolumeId))
+				vid := needle.VolumeId(req.VolumeId)
+				vl.SetVolumeReadOnly(dn, vid)
+				if pending := vl.GetPendingSize(vid); pending > 0 {
+					glog.V(0).Infof("volume %d marked readonly with %d pending bytes", vid, pending)
+				}
 			} else {
 				vl.SetVolumeWritable(dn, needle.VolumeId(req.VolumeId))
 			}

weed/topology/topology.go

@@ -505,7 +505,7 @@ func (t *Topology) SyncDataNodeRegistration(volumes []*master_pb.VolumeInformati
 		}
 		diskType := types.ToDiskType(v.DiskType)
 		vl := t.GetVolumeLayout(v.Collection, v.ReplicaPlacement, v.Ttl, diskType)
-		vl.UpdateVolumeSize(v.Id, v.Size)
+		vl.UpdateVolumeSize(v.Id, v.Size, v.CompactRevision)
 	}
 	return
 }

weed/topology/topology_vacuum.go

@@ -67,9 +67,7 @@ func (t *Topology) batchVacuumVolumeCheck(grpcDialOption grpc.DialOption, vid ne
 
 func (t *Topology) batchVacuumVolumeCompact(grpcDialOption grpc.DialOption, vl *VolumeLayout, vid needle.VolumeId,
 	locationlist *VolumeLocationList, preallocate int64) bool {
-	vl.accessLock.Lock()
-	vl.removeFromWritable(vid)
-	vl.accessLock.Unlock()
+	vl.DrainAndRemoveFromWritable(vid)
 
 	ch := make(chan bool, locationlist.Length())
 	for index, dn := range locationlist.list {

weed/topology/volume_layout.go

@@ -1,6 +1,7 @@
 package topology
 
 import (
+	"context"
 	"fmt"
 	"math/rand/v2"
 	"sync"
@@ -101,6 +102,14 @@ func (v *volumesBinaryState) copyState(list *VolumeLocationList) copyState {
 	return enoughCopies
 }
 
+// volumeSizeTracking holds per-volume size accounting for weighted assignment.
+type volumeSizeTracking struct {
+	effectiveSize   uint64    // reported + pending assigned bytes
+	reportedSize    uint64    // last heartbeat-reported size (dedup replicas)
+	compactRevision uint32    // detect compaction to reset instead of decay
+	lastUpdateTime  time.Time // dedup replicas within the same heartbeat cycle
+}
+
 // mapping from volume to its locations, inverted from server to volume
 type VolumeLayout struct {
 	growRequest      atomic.Bool
@@ -116,9 +125,8 @@ type VolumeLayout struct {
 	vacuumedVolumes  map[needle.VolumeId]time.Time
 	volumeSizeLimit  uint64
 	replicationAsMin bool
-	accessLock        sync.RWMutex
-	vid2size          map[needle.VolumeId]uint64 // effective size: reported + pending
-	vid2reportedSize  map[needle.VolumeId]uint64 // last heartbeat-reported size (dedup replicas)
+	accessLock       sync.RWMutex
+	sizeTracking     map[needle.VolumeId]*volumeSizeTracking
 }
 
 type VolumeLayoutStats struct {
@@ -140,8 +148,7 @@ func NewVolumeLayout(rp *super_block.ReplicaPlacement, ttl *needle.TTL, diskType
 		vacuumedVolumes:  make(map[needle.VolumeId]time.Time),
 		volumeSizeLimit:  volumeSizeLimit,
 		replicationAsMin: replicationAsMin,
-		vid2size:          make(map[needle.VolumeId]uint64),
-		vid2reportedSize:  make(map[needle.VolumeId]uint64),
+		sizeTracking:    make(map[needle.VolumeId]*volumeSizeTracking),
 	}
 }
 
@@ -159,10 +166,13 @@ func (vl *VolumeLayout) RegisterVolume(v *storage.VolumeInfo, dn *DataNode) {
 		vl.vid2location[v.Id] = NewVolumeLocationList()
 	}
 	vl.vid2location[v.Id].Set(dn)
-	// For new volumes, initialize vid2size from reported size.
-	if _, exists := vl.vid2size[v.Id]; !exists {
-		vl.vid2size[v.Id] = v.Size
-		vl.vid2reportedSize[v.Id] = v.Size
+	// For new volumes, initialize size tracking from reported size.
+	if _, exists := vl.sizeTracking[v.Id]; !exists {
+		vl.sizeTracking[v.Id] = &volumeSizeTracking{
+			effectiveSize:   v.Size,
+			reportedSize:    v.Size,
+			compactRevision: v.CompactRevision,
+		}
 	}
 	// glog.V(4).Infof("volume %d added to %s len %d copy %d", v.Id, dn.Id(), vl.vid2location[v.Id].Length(), v.ReplicaPlacement.GetCopyCount())
 	for _, dn := range vl.vid2location[v.Id].list {
@@ -197,20 +207,40 @@ func (vl *VolumeLayout) rememberOversizedVolume(v *storage.VolumeInfo, dn *DataN
 // It decays the pending size estimate toward the reported size and updates
 // crowded state. Replicated volumes report from multiple DataNodes; decay
 // runs only once per new reported size to avoid double-halving.
-func (vl *VolumeLayout) UpdateVolumeSize(vid needle.VolumeId, reportedSize uint64) {
+// If the compact revision changed, the size drop is from compaction (not
+// pending writes), so we reset effectiveSize to the reported size instead of
+// decaying.
+func (vl *VolumeLayout) UpdateVolumeSize(vid needle.VolumeId, reportedSize uint64, compactRevision uint32) {
 	vl.accessLock.Lock()
 	defer vl.accessLock.Unlock()
 
-	if reportedSize == vl.vid2reportedSize[vid] {
-		return // same size from another replica in this cycle
-	}
-	vl.vid2reportedSize[vid] = reportedSize
-	if prev := vl.vid2size[vid]; prev > reportedSize {
-		vl.vid2size[vid] = reportedSize + (prev-reportedSize)/2
+	now := time.Now()
+	st := vl.sizeTracking[vid]
+	if st == nil {
+		st = &volumeSizeTracking{
+			effectiveSize:   reportedSize,
+			reportedSize:    reportedSize,
+			compactRevision: compactRevision,
+			lastUpdateTime:  now,
+		}
+		vl.sizeTracking[vid] = st
+	} else if now.Sub(st.lastUpdateTime) < 2*time.Second {
+		return // duplicate replica in the same heartbeat cycle
 	} else {
-		vl.vid2size[vid] = reportedSize
+		st.lastUpdateTime = now
+		st.reportedSize = reportedSize
+		if compactRevision != st.compactRevision {
+			// Compaction happened — size drop is real, not pending. Reset.
+			st.compactRevision = compactRevision
+			st.effectiveSize = reportedSize
+		} else if st.effectiveSize > reportedSize {
+			st.effectiveSize = reportedSize + (st.effectiveSize-reportedSize)/2
+		} else {
+			st.effectiveSize = reportedSize
+		}
 	}
-	if float64(vl.vid2size[vid]) > float64(vl.volumeSizeLimit)*VolumeGrowStrategy.Threshold {
+
+	if float64(st.effectiveSize) > float64(vl.volumeSizeLimit)*VolumeGrowStrategy.Threshold {
 		vl.setVolumeCrowded(vid)
 	} else {
 		vl.removeFromCrowded(vid)
@@ -235,8 +265,7 @@ func (vl *VolumeLayout) UnRegisterVolume(v *storage.VolumeInfo, dn *DataNode) {
 
 		if location.Length() == 0 {
 			delete(vl.vid2location, v.Id)
-			delete(vl.vid2size, v.Id)
-			delete(vl.vid2reportedSize, v.Id)
+			delete(vl.sizeTracking, v.Id)
 			vl.removeFromCrowded(v.Id)
 		}
 
@@ -301,14 +330,62 @@ func (vl *VolumeLayout) RecordAssign(vid needle.VolumeId, pendingDelta int64) {
 	vl.accessLock.Lock()
 	defer vl.accessLock.Unlock()
 
-	if pendingDelta > 0 {
-		vl.vid2size[vid] += uint64(pendingDelta)
+	st := vl.sizeTracking[vid]
+	if st == nil {
+		return
 	}
-	if float64(vl.vid2size[vid]) > float64(vl.volumeSizeLimit)*VolumeGrowStrategy.Threshold {
+	if pendingDelta > 0 {
+		st.effectiveSize += uint64(pendingDelta)
+	}
+	if float64(st.effectiveSize) > float64(vl.volumeSizeLimit)*VolumeGrowStrategy.Threshold {
 		vl.setVolumeCrowded(vid)
 	}
 }
 
+const maxDrainWait = 30 * time.Second
+const pendingSizeThreshold uint64 = 2 * 1024 * 1024 // 2 MB
+
+// GetPendingSize returns the estimated in-flight bytes for a volume:
+// the gap between the effective tracked size and the last heartbeat-reported size.
+func (vl *VolumeLayout) GetPendingSize(vid needle.VolumeId) uint64 {
+	vl.accessLock.RLock()
+	defer vl.accessLock.RUnlock()
+	if st := vl.sizeTracking[vid]; st != nil && st.effectiveSize > st.reportedSize {
+		return st.effectiveSize - st.reportedSize
+	}
+	return 0
+}
+
+// waitForPendingDrain polls until pending bytes for the volume decay below
+// the threshold, the timeout expires, or the context is cancelled. Since the
+// volume is already removed from the writable list, no new assigns accumulate
+// — pending only decreases via heartbeat decay.
+func (vl *VolumeLayout) waitForPendingDrain(ctx context.Context, vid needle.VolumeId) {
+	deadline := time.Now().Add(maxDrainWait)
+	for time.Now().Before(deadline) {
+		if vl.GetPendingSize(vid) <= pendingSizeThreshold {
+			return
+		}
+		select {
+		case <-ctx.Done():
+			return
+		case <-time.After(1 * time.Second):
+		}
+	}
+	glog.Warningf("volume %d: %d pending bytes remain after drain timeout", vid, vl.GetPendingSize(vid))
+}
+
+// DrainAndRemoveFromWritable removes the volume from the writable list
+// immediately, then waits for pending assigned bytes to decay.
+// Used by vacuum before compaction.
+func (vl *VolumeLayout) DrainAndRemoveFromWritable(vid needle.VolumeId) {
+	vl.accessLock.Lock()
+	vl.removeFromWritable(vid)
+	vl.accessLock.Unlock()
+	vl.waitForPendingDrain(context.Background(), vid)
+}
+
+
 func (vl *VolumeLayout) isEmpty() bool {
 	vl.accessLock.RLock()
 	defer vl.accessLock.RUnlock()
@@ -432,7 +509,10 @@ func (vl *VolumeLayout) weightedPick(candidates []needle.VolumeId) (needle.Volum
 }
 
 func (vl *VolumeLayout) remainingSize(vid needle.VolumeId) uint64 {
-	size := vl.vid2size[vid]
+	var size uint64
+	if st := vl.sizeTracking[vid]; st != nil {
+		size = st.effectiveSize
+	}
 	if size < vl.volumeSizeLimit {
 		if r := vl.volumeSizeLimit - size; r > 1 {
 			return r
@@ -484,7 +564,10 @@ func (vl *VolumeLayout) ShouldGrowVolumesByDcAndRack(writables *[]needle.VolumeI
 			}
 			if _, err := dn.GetVolumesById(v); err == nil {
 				vl.accessLock.RLock()
-				size := vl.vid2size[v]
+				var size uint64
+				if st := vl.sizeTracking[v]; st != nil {
+					size = st.effectiveSize
+				}
 				vl.accessLock.RUnlock()
 				if float64(size) <= float64(vl.volumeSizeLimit)*VolumeGrowStrategy.Threshold {
 					return false

weed/topology/volume_layout_drain_test.go

@@ -0,0 +1,261 @@
+package topology
+
+import (
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
+	"github.com/seaweedfs/seaweedfs/weed/storage/types"
+)
+
+func TestGetPendingSize(t *testing.T) {
+	layout := `
+{
+  "dc1":{
+    "rack1":{
+      "server1":{
+        "volumes":[
+          {"id":1, "size":1000, "replication":"000"}
+        ],
+        "limit":10
+      }
+    }
+  }
+}
+`
+	_, vl := setupPickTest(t, layout, 10000)
+
+	// Initially no pending
+	if p := vl.GetPendingSize(1); p != 0 {
+		t.Fatalf("expected 0 pending, got %d", p)
+	}
+
+	// RecordAssign increases pending
+	vl.RecordAssign(1, 5000)
+	if p := vl.GetPendingSize(1); p != 5000 {
+		t.Fatalf("expected 5000 pending, got %d", p)
+	}
+
+	// UpdateVolumeSize (heartbeat) decays pending
+	vl.UpdateVolumeSize(1, 3000, 0)
+	// effective was 6000, reported 3000 → decay to 3000 + (6000-3000)/2 = 4500
+	// pending = 4500 - 3000 = 1500
+	if p := vl.GetPendingSize(1); p != 1500 {
+		t.Fatalf("expected 1500 pending after decay, got %d", p)
+	}
+}
+
+func TestGetPendingSize_CompactionResets(t *testing.T) {
+	layout := `
+{
+  "dc1":{
+    "rack1":{
+      "server1":{
+        "volumes":[
+          {"id":1, "size":5000, "replication":"000"}
+        ],
+        "limit":10
+      }
+    }
+  }
+}
+`
+	_, vl := setupPickTest(t, layout, 10000)
+
+	// Add large pending
+	vl.RecordAssign(1, 4000)
+	if p := vl.GetPendingSize(1); p != 4000 {
+		t.Fatalf("expected 4000 pending, got %d", p)
+	}
+
+	// Compaction happens — size drops from 5000 to 2000, revision changes.
+	// Without compaction awareness, decay would give: 2000 + (9000-2000)/2 = 5500.
+	// With compaction awareness, vid2size resets to 2000 (the real size).
+	vl.UpdateVolumeSize(1, 2000, 1) // revision 0 → 1
+
+	if p := vl.GetPendingSize(1); p != 0 {
+		t.Errorf("expected 0 pending after compaction reset, got %d", p)
+	}
+
+	// Verify vid2size is the reported size, not a decayed value
+	vl.accessLock.RLock()
+	if vl.sizeTracking[1].effectiveSize != 2000 {
+		t.Errorf("expected vid2size=2000 after compaction, got %d", vl.sizeTracking[1].effectiveSize)
+	}
+	vl.accessLock.RUnlock()
+}
+
+func TestDrainAndRemoveFromWritable_NoPending(t *testing.T) {
+	layout := `
+{
+  "dc1":{
+    "rack1":{
+      "server1":{
+        "volumes":[
+          {"id":1, "size":1000, "replication":"000"}
+        ],
+        "limit":10
+      }
+    }
+  }
+}
+`
+	_, vl := setupPickTest(t, layout, 10000)
+
+	// No pending — drain should return immediately
+	start := time.Now()
+	vl.DrainAndRemoveFromWritable(1)
+	if elapsed := time.Since(start); elapsed > 500*time.Millisecond {
+		t.Errorf("drain with no pending took %v, expected near-instant", elapsed)
+	}
+
+	// Verify volume is no longer writable
+	writable, _ := vl.GetWritableVolumeCount()
+	if writable != 0 {
+		t.Errorf("expected 0 writable after drain, got %d", writable)
+	}
+}
+
+func TestDrainAndRemoveFromWritable_WithPending(t *testing.T) {
+	layout := `
+{
+  "dc1":{
+    "rack1":{
+      "server1":{
+        "volumes":[
+          {"id":1, "size":1000, "replication":"000"},
+          {"id":2, "size":1000, "replication":"000"}
+        ],
+        "limit":10
+      }
+    }
+  }
+}
+`
+	_, vl := setupPickTest(t, layout, 10000)
+
+	// Add pending below threshold
+	vl.RecordAssign(1, int64(pendingSizeThreshold-1))
+
+	start := time.Now()
+	vl.DrainAndRemoveFromWritable(1)
+	elapsed := time.Since(start)
+
+	// Should return quickly since pending is below threshold
+	if elapsed > 500*time.Millisecond {
+		t.Errorf("drain with pending below threshold took %v", elapsed)
+	}
+
+	// Volume removed from writable
+	writables := vl.CloneWritableVolumes()
+	for _, vid := range writables {
+		if vid == 1 {
+			t.Error("volume 1 should not be writable after drain")
+		}
+	}
+	// Volume 2 still writable
+	found := false
+	for _, vid := range writables {
+		if vid == 2 {
+			found = true
+		}
+	}
+	if !found {
+		t.Error("volume 2 should still be writable")
+	}
+}
+
+func TestDrainAndRemoveFromWritable_DecaysViaConcurrentHeartbeat(t *testing.T) {
+	layout := `
+{
+  "dc1":{
+    "rack1":{
+      "server1":{
+        "volumes":[
+          {"id":1, "size":1000, "replication":"000"}
+        ],
+        "limit":10
+      }
+    }
+  }
+}
+`
+	_, vl := setupPickTest(t, layout, 10000)
+
+	// Add large pending (well above threshold)
+	vl.RecordAssign(1, 100*1024*1024) // 100 MB
+
+	// Simulate heartbeats in background that will decay the pending.
+	// Advance lastUpdateTime before each call to bypass the 2s replica dedup.
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		for i := 0; i < 10; i++ {
+			time.Sleep(500 * time.Millisecond)
+			vl.accessLock.Lock()
+			if st := vl.sizeTracking[1]; st != nil {
+				st.lastUpdateTime = time.Time{} // reset to allow update
+			}
+			vl.accessLock.Unlock()
+			vl.UpdateVolumeSize(1, 1000+uint64(i+1)*1000, 0)
+		}
+	}()
+
+	start := time.Now()
+	vl.DrainAndRemoveFromWritable(1)
+	elapsed := time.Since(start)
+
+	// Should have drained within a few seconds (heartbeats every 500ms).
+	// Use generous margin for slow CI.
+	if elapsed > 15*time.Second {
+		t.Errorf("drain took %v, expected faster with concurrent heartbeats", elapsed)
+	}
+
+	<-done
+
+	// Verify volume is no longer writable
+	writable, _ := vl.GetWritableVolumeCount()
+	if writable != 0 {
+		t.Errorf("expected 0 writable after drain, got %d", writable)
+	}
+}
+
+func TestSetVolumeReadOnly_PreservesPending(t *testing.T) {
+	layout := `
+{
+  "dc1":{
+    "rack1":{
+      "server1":{
+        "volumes":[
+          {"id":1, "size":1000, "replication":"000"}
+        ],
+        "limit":10
+      }
+    }
+  }
+}
+`
+	topo := setupWithLimit(t, layout, 10000)
+	rp, _ := super_block.NewReplicaPlacementFromString("000")
+	vl := topo.GetVolumeLayout("", rp, needle.EMPTY_TTL, types.HardDriveType)
+	dn := vl.Lookup(1)[0]
+
+	// Add some pending
+	vl.RecordAssign(1, 5000)
+
+	// SetVolumeReadOnly should succeed immediately (non-blocking)
+	result := vl.SetVolumeReadOnly(dn, 1)
+	if !result {
+		t.Error("expected SetVolumeReadOnly to return true")
+	}
+
+	// Pending is still there (not drained), but volume is readonly
+	writable, _ := vl.GetWritableVolumeCount()
+	if writable != 0 {
+		t.Errorf("expected 0 writable after readonly, got %d", writable)
+	}
+	if p := vl.GetPendingSize(1); p != 5000 {
+		t.Errorf("expected 5000 pending (not drained), got %d", p)
+	}
+}

weed/topology/volume_layout_pick_test.go

@@ -4,6 +4,7 @@ import (
 	"encoding/json"
 	"math"
 	"testing"
+	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/sequence"
 	"github.com/seaweedfs/seaweedfs/weed/storage"
@@ -334,39 +335,48 @@ func TestHeartbeatDecaysPendingSize(t *testing.T) {
 	vl.RecordAssign(1, 8000)
 
 	vl.accessLock.RLock()
-	if vl.vid2size[1] != 9000 {
-		t.Fatalf("expected vid2size=9000 after RecordAssign, got %d", vl.vid2size[1])
+	if vl.sizeTracking[1].effectiveSize != 9000 {
+		t.Fatalf("expected vid2size=9000 after RecordAssign, got %d", vl.sizeTracking[1].effectiveSize)
 	}
 	vl.accessLock.RUnlock()
 
+	// Helper to simulate a new heartbeat cycle (advance past dedup window)
+	advanceCycle := func() {
+		vl.accessLock.Lock()
+		vl.sizeTracking[1].lastUpdateTime = time.Now().Add(-3 * time.Second)
+		vl.accessLock.Unlock()
+	}
+
 	// Heartbeat: volume server reports size=3000 (some writes landed).
 	// Old effective=9000, new reported=3000 → excess=6000 → decayed to 3000.
 	// So vid2size should become 3000 + 6000/2 = 6000, not just 3000.
-	vl.UpdateVolumeSize(1, 3000)
+	vl.UpdateVolumeSize(1, 3000, 0)
 
 	vl.accessLock.RLock()
-	if vl.vid2size[1] != 6000 {
-		t.Errorf("expected vid2size=6000 after decay (3000 + 6000/2), got %d", vl.vid2size[1])
+	if vl.sizeTracking[1].effectiveSize != 6000 {
+		t.Errorf("expected vid2size=6000 after decay (3000 + 6000/2), got %d", vl.sizeTracking[1].effectiveSize)
 	}
 	vl.accessLock.RUnlock()
 
 	// Second heartbeat: size=5000. Old effective=6000 → excess=1000 → decay to 500.
 	// vid2size should become 5000 + 1000/2 = 5500.
-	vl.UpdateVolumeSize(1, 5000)
+	advanceCycle()
+	vl.UpdateVolumeSize(1, 5000, 0)
 
 	vl.accessLock.RLock()
-	if vl.vid2size[1] != 5500 {
-		t.Errorf("expected vid2size=5500 after second decay (5000 + 1000/2), got %d", vl.vid2size[1])
+	if vl.sizeTracking[1].effectiveSize != 5500 {
+		t.Errorf("expected vid2size=5500 after second decay (5000 + 1000/2), got %d", vl.sizeTracking[1].effectiveSize)
 	}
 	vl.accessLock.RUnlock()
 
 	// Third heartbeat: size=5500. Old effective=5500 → no excess.
 	// vid2size should be exactly 5500.
-	vl.UpdateVolumeSize(1, 5500)
+	advanceCycle()
+	vl.UpdateVolumeSize(1, 5500, 0)
 
 	vl.accessLock.RLock()
-	if vl.vid2size[1] != 5500 {
-		t.Errorf("expected vid2size=5500 (no excess), got %d", vl.vid2size[1])
+	if vl.sizeTracking[1].effectiveSize != 5500 {
+		t.Errorf("expected vid2size=5500 (no excess), got %d", vl.sizeTracking[1].effectiveSize)
 	}
 	vl.accessLock.RUnlock()
 
@@ -419,18 +429,18 @@ func TestHeartbeatDecayDedupReplicas(t *testing.T) {
 	vl.RecordAssign(1, 8000)
 
 	vl.accessLock.RLock()
-	if vl.vid2size[1] != 9000 {
-		t.Fatalf("expected vid2size=9000, got %d", vl.vid2size[1])
+	if vl.sizeTracking[1].effectiveSize != 9000 {
+		t.Fatalf("expected vid2size=9000, got %d", vl.sizeTracking[1].effectiveSize)
 	}
 	vl.accessLock.RUnlock()
 
 	// Both replicas report size=3000. Decay should happen once: 3000 + (9000-3000)/2 = 6000.
 	// Calling UpdateVolumeSize twice simulates two replicas reporting in the same cycle.
-	vl.UpdateVolumeSize(1, 3000)
-	vl.UpdateVolumeSize(1, 3000) // second replica, same size — should be a no-op
+	vl.UpdateVolumeSize(1, 3000, 0)
+	vl.UpdateVolumeSize(1, 3000, 0) // second replica, same size — should be a no-op
 
 	vl.accessLock.RLock()
-	got := vl.vid2size[1]
+	got := vl.sizeTracking[1].effectiveSize
 	vl.accessLock.RUnlock()
 	// Without dedup: would be 3000 + (6000-3000)/2 = 4500 (double decay).
 	// With dedup: should be 6000 (single decay).
@@ -439,6 +449,49 @@ func TestHeartbeatDecayDedupReplicas(t *testing.T) {
 	}
 }
 
+func TestUpdateVolumeSize_DecaysEvenWhenReportedSizeUnchanged(t *testing.T) {
+	layout := `
+{
+  "dc1":{
+    "rack1":{
+      "server1":{
+        "volumes":[
+          {"id":1, "size":1000, "replication":"000"}
+        ],
+        "limit":10
+      }
+    }
+  }
+}
+`
+	_, vl := setupPickTest(t, layout, 10000)
+
+	// Add pending: effective = 1000 + 8000 = 9000
+	vl.RecordAssign(1, 8000)
+	if p := vl.GetPendingSize(1); p != 8000 {
+		t.Fatalf("expected 8000 pending, got %d", p)
+	}
+
+	// First heartbeat: reported size unchanged at 1000 (writes haven't landed).
+	// Decay should still run: 1000 + (9000-1000)/2 = 5000.
+	vl.UpdateVolumeSize(1, 1000, 0)
+	if p := vl.GetPendingSize(1); p != 4000 {
+		t.Errorf("expected 4000 pending after first decay, got %d", p)
+	}
+
+	// Simulate next heartbeat cycle (>2s later) with same reported size.
+	// Need to advance lastUpdateTime — manipulate directly under lock.
+	vl.accessLock.Lock()
+	vl.sizeTracking[1].lastUpdateTime = time.Now().Add(-3 * time.Second)
+	vl.accessLock.Unlock()
+
+	// Second heartbeat: still 1000. Decay again: 1000 + (5000-1000)/2 = 3000.
+	vl.UpdateVolumeSize(1, 1000, 0)
+	if p := vl.GetPendingSize(1); p != 2000 {
+		t.Errorf("expected 2000 pending after second decay, got %d", p)
+	}
+}
+
 func TestShouldGrowVolumesByDcAndRack_WithPendingSize(t *testing.T) {
 	layout := `
 {