diff --git a/.github/workflows/vacuum-integration-tests.yml b/.github/workflows/vacuum-integration-tests.yml
new file mode 100644
index 000000000..26d6b8c75
--- /dev/null
+++ b/.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
diff --git a/test/vacuum/vacuum_integration_test.go b/test/vacuum/vacuum_integration_test.go
new file mode 100644
index 000000000..166946106
--- /dev/null
+++ b/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
+	}
+}
diff --git a/weed/server/master_grpc_server_volume.go b/weed/server/master_grpc_server_volume.go
index 512dfd2aa..dc74b5709 100644
--- a/weed/server/master_grpc_server_volume.go
+++ b/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))
 			}
diff --git a/weed/topology/topology.go b/weed/topology/topology.go
index 0c26bfd86..9d133530c 100644
--- a/weed/topology/topology.go
+++ b/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
 }
diff --git a/weed/topology/topology_vacuum.go b/weed/topology/topology_vacuum.go
index 45901f777..a5ea14d05 100644
--- a/weed/topology/topology_vacuum.go
+++ b/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 {
diff --git a/weed/topology/volume_layout.go b/weed/topology/volume_layout.go
index 938592113..39602502f 100644
--- a/weed/topology/volume_layout.go
+++ b/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
diff --git a/weed/topology/volume_layout_drain_test.go b/weed/topology/volume_layout_drain_test.go
new file mode 100644
index 000000000..6132b7a2c
--- /dev/null
+++ b/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)
+	}
+}
diff --git a/weed/topology/volume_layout_pick_test.go b/weed/topology/volume_layout_pick_test.go
index 62860a87b..f63878c9e 100644
--- a/weed/topology/volume_layout_pick_test.go
+++ b/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 := `
 {