fix(vacuum): writable volume re-notification after worker VACUUM (#9732)

* fix(vacuum): notify master writable after worker vacuum commit

Add Phase 3 (markWritableOne) that walks vacuumTargets and calls
VolumeMarkWritable on each replica's volume server, mirroring
batchVacuumVolumeCommit's per-replica SetVolumeAvailable. Failures are
logged at WARN; the task does not fail because the vacuum itself
already succeeded. See upstream seaweedfs#9685.

* fix(vacuum): delay Phase 3 to let post-commit heartbeats settle

Phase 3's VolumeMarkWritable can race with the volume server's first
post-commit heartbeat. SetVolumeWritable adds the vid to writables,
but a racing heartbeat whose ReadOnly value changed re-runs
EnsureCorrectWritables against the master's per-replica cache, and any
replica still cached as ReadOnly=true silently removes the vid again
— with no further heartbeat change to trigger another recovery.

Sleep 30s after Phase 2 (Commit) so every replica's post-vacuum
heartbeat has reached the master before Phase 3 fires. Cancel cleanly
on ctx.Done so a shutdown during the wait still exits.

* fix(vacuum): reduce post-commit settle from 30s to 10s

VolumePulsePeriod is 5s, so 10s (2x) is enough margin for every
replica's post-commit heartbeat to reach the master before Phase 3
fires. 30s was overly conservative and made TestVacuumExecutionIntegration
hit its 30s context deadline.

* fix(vacuum): use flat 1m timeout for VolumeMarkWritable RPC

VolumeMarkWritable on the volume server is a metadata operation
(reopen idx + flags + master ReadOnly=false heartbeat), independent
of volume size. Scaling via vacuumTimeout(time.Minute) gave it tens
of minutes — even hours on TB volumes — so a single unresponsive
replica could block Phase 3 indefinitely. Use a flat 1m cap.

* fix(vacuum): gate post-vacuum mark-writable on commit read-only state

Phase 3 force-called VolumeMarkWritable on every replica unconditionally,
clearing the read-only flag and persisting ReadOnly=false even for a
replica left read-only by an operator, an EIO quarantine, or low disk.
That overrode states the master deliberately keeps out of writables;
master built-in vacuum gates the same step on the commit's IsReadOnly via
SetVolumeAvailable.

Capture the VacuumVolumeCommit response and skip Phase 3 when any replica
came back read-only, letting it recover on its own ReadOnly=false
heartbeat. Drop the 10s post-commit settle sleep: the heartbeat race it
guarded needed a replica cached read-only at the master, which the gate
now excludes.

---------

Co-authored-by: Chris Lu <chris.lu@gmail.com>

test/plugin_workers/fake_volume_server.go

@@ -30,21 +30,22 @@ type VolumeServer struct {
 	address  string
 	baseDir  string
 
-	mu                  sync.Mutex
-	receivedFiles       map[string]uint64
-	mountRequests       []*volume_server_pb.VolumeEcShardsMountRequest
-	deleteRequests      []*volume_server_pb.VolumeDeleteRequest
-	markReadonlyCalls   int
-	markWritableCalls   int
-	readFileStatusCalls int
-	vacuumGarbageRatio  float64
-	vacuumCheckCalls    int
-	vacuumCompactCalls  int
-	vacuumCommitCalls   int
-	vacuumCleanupCalls  int
-	volumeCopyCalls     int
-	volumeMountCalls    int
-	tailReceiverCalls   int
+	mu                   sync.Mutex
+	receivedFiles        map[string]uint64
+	mountRequests        []*volume_server_pb.VolumeEcShardsMountRequest
+	deleteRequests       []*volume_server_pb.VolumeDeleteRequest
+	markReadonlyCalls    int
+	markWritableCalls    int
+	readFileStatusCalls  int
+	vacuumGarbageRatio   float64
+	vacuumCommitReadOnly bool
+	vacuumCheckCalls     int
+	vacuumCompactCalls   int
+	vacuumCommitCalls    int
+	vacuumCleanupCalls   int
+	volumeCopyCalls      int
+	volumeMountCalls     int
+	tailReceiverCalls    int
 }
 
 // NewVolumeServer starts a test volume server using the provided base directory.
@@ -115,6 +116,13 @@ func (v *VolumeServer) SetVacuumGarbageRatio(ratio float64) {
 	v.vacuumGarbageRatio = ratio
 }
 
+// SetVacuumCommitReadOnly sets the IsReadOnly value returned by VacuumVolumeCommit.
+func (v *VolumeServer) SetVacuumCommitReadOnly(readOnly bool) {
+	v.mu.Lock()
+	defer v.mu.Unlock()
+	v.vacuumCommitReadOnly = readOnly
+}
+
 // VacuumStats returns the vacuum RPC call counts.
 func (v *VolumeServer) VacuumStats() (check, compact, commit, cleanup int) {
 	v.mu.Lock()
@@ -428,8 +436,9 @@ func (v *VolumeServer) VacuumVolumeCompact(req *volume_server_pb.VacuumVolumeCom
 func (v *VolumeServer) VacuumVolumeCommit(ctx context.Context, req *volume_server_pb.VacuumVolumeCommitRequest) (*volume_server_pb.VacuumVolumeCommitResponse, error) {
 	v.mu.Lock()
 	v.vacuumCommitCalls++
+	readOnly := v.vacuumCommitReadOnly
 	v.mu.Unlock()
-	return &volume_server_pb.VacuumVolumeCommitResponse{}, nil
+	return &volume_server_pb.VacuumVolumeCommitResponse{IsReadOnly: readOnly}, nil
 }
 
 func (v *VolumeServer) VacuumVolumeCleanup(ctx context.Context, req *volume_server_pb.VacuumVolumeCleanupRequest) (*volume_server_pb.VacuumVolumeCleanupResponse, error) {

test/plugin_workers/vacuum/execution_test.go

@@ -15,6 +15,24 @@ import (
 	"google.golang.org/grpc/credentials/insecure"
 )
 
+func vacuumJobSpec(volumeID uint32, server string) *plugin_pb.JobSpec {
+	return &plugin_pb.JobSpec{
+		JobId:   fmt.Sprintf("vacuum-job-%d", volumeID),
+		JobType: "vacuum",
+		Parameters: map[string]*plugin_pb.ConfigValue{
+			"volume_id": {
+				Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: int64(volumeID)},
+			},
+			"server": {
+				Kind: &plugin_pb.ConfigValue_StringValue{StringValue: server},
+			},
+			"collection": {
+				Kind: &plugin_pb.ConfigValue_StringValue{StringValue: "vac-test"},
+			},
+		},
+	}
+}
+
 func TestVacuumExecutionIntegration(t *testing.T) {
 	volumeID := uint32(202)
 
@@ -31,26 +49,10 @@ func TestVacuumExecutionIntegration(t *testing.T) {
 	source := pluginworkers.NewVolumeServer(t, "")
 	source.SetVacuumGarbageRatio(0.6)
 
-	job := &plugin_pb.JobSpec{
-		JobId:   fmt.Sprintf("vacuum-job-%d", volumeID),
-		JobType: "vacuum",
-		Parameters: map[string]*plugin_pb.ConfigValue{
-			"volume_id": {
-				Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: int64(volumeID)},
-			},
-			"server": {
-				Kind: &plugin_pb.ConfigValue_StringValue{StringValue: source.Address()},
-			},
-			"collection": {
-				Kind: &plugin_pb.ConfigValue_StringValue{StringValue: "vac-test"},
-			},
-		},
-	}
-
 	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
 	defer cancel()
 
-	result, err := harness.Plugin().ExecuteJob(ctx, job, nil, 1)
+	result, err := harness.Plugin().ExecuteJob(ctx, vacuumJobSpec(volumeID, source.Address()), nil, 1)
 	require.NoError(t, err)
 	require.NotNil(t, result)
 	require.True(t, result.Success)
@@ -66,4 +68,41 @@ func TestVacuumExecutionIntegration(t *testing.T) {
 	// topology.vacuumOneVolumeId which only calls batchVacuumVolumeCleanup
 	// on the Compact-failure branch.
 	require.Equal(t, 0, cleanupCalls)
+	// Phase 3 marks each replica writable so master returns it to the
+	// writables layout. See upstream seaweedfs#9685.
+	require.GreaterOrEqual(t, source.MarkWritableCount(), 1)
+}
+
+// A replica that commits still read-only (operator-set, EIO-quarantined,
+// disk-space-low) must not be force-marked writable: master built-in vacuum
+// skips it via SetVolumeAvailable, and it recovers on its own ReadOnly=false
+// heartbeat.
+func TestVacuumExecutionSkipsMarkWritableWhenReadOnly(t *testing.T) {
+	volumeID := uint32(203)
+
+	dialOption := grpc.WithTransportCredentials(insecure.NewCredentials())
+	handler := vacuum.NewVacuumHandler(dialOption, 1)
+	harness := pluginworkers.NewHarness(t, pluginworkers.HarnessConfig{
+		WorkerOptions: pluginworker.WorkerOptions{
+			GrpcDialOption: dialOption,
+		},
+		Handlers: []pluginworker.JobHandler{handler},
+	})
+	harness.WaitForJobType("vacuum")
+
+	source := pluginworkers.NewVolumeServer(t, "")
+	source.SetVacuumGarbageRatio(0.6)
+	source.SetVacuumCommitReadOnly(true)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+
+	result, err := harness.Plugin().ExecuteJob(ctx, vacuumJobSpec(volumeID, source.Address()), nil, 1)
+	require.NoError(t, err)
+	require.NotNil(t, result)
+	require.True(t, result.Success)
+
+	_, _, commitCalls, _ := source.VacuumStats()
+	require.GreaterOrEqual(t, commitCalls, 1)
+	require.Equal(t, 0, source.MarkWritableCount())
 }

weed/worker/tasks/vacuum/vacuum_task.go

@@ -233,20 +233,26 @@ func (t *VacuumTask) checkOneVacuumEligibility(ctx context.Context, server strin
 	return garbageRatio, err
 }
 
-// performVacuum runs the two-phase vacuum protocol that master built-in
-// vacuum uses (topology.vacuumOneVolumeId):
+// performVacuum runs the three-phase vacuum protocol that mirrors
+// master built-in vacuum (topology.vacuumOneVolumeId):
 //
-//   Phase 1 (Compact): build the new .cpd/.cpx files on every target.
-//   If any replica fails, roll back by Cleanup'ing the .cp* temp files
-//   on every target and abort — no replica has yet swapped its active
-//   files, so no replica is committed.
+//	Phase 1 (Compact): build the new .cpd/.cpx files on every target.
+//	If any replica fails, roll back by Cleanup'ing the .cp* temp files
+//	on every target and abort — no replica has yet swapped its active
+//	files, so no replica is committed.
 //
-//   Phase 2 (Commit): swap each target's active files with its .cp*
-//   files. Best-effort, matching batchVacuumVolumeCommit: per-replica
-//   errors are logged and surfaced together, but once any replica has
-//   swapped there is no clean rollback for the others, so we do not
-//   retry or undo. An operator must reconcile a partial commit
-//   failure.
+//	Phase 2 (Commit): swap each target's active files with its .cp*
+//	files. Best-effort, matching batchVacuumVolumeCommit: per-replica
+//	errors are logged and surfaced together, but once any replica has
+//	swapped there is no clean rollback for the others, so we do not
+//	retry or undo. An operator must reconcile a partial commit
+//	failure.
+//
+//	Phase 3 (Mark Writable): re-notify the master per replica so the
+//	volume re-enters the writable set, the worker analog of
+//	batchVacuumVolumeCommit's per-replica SetVolumeAvailable. Skipped
+//	when a replica came back read-only. Best-effort; never fails the
+//	task.
 //
 // Interleaving Compact→Commit→Cleanup per replica (the prior behavior)
 // could leave a committed first replica beside an uncompacted second
@@ -261,18 +267,47 @@ func (t *VacuumTask) performVacuum(ctx context.Context) error {
 		}
 	}
 
-	// Phase 2: Commit all targets.
+	// Phase 2: Commit all targets, tracking whether any replica is still
+	// read-only after the swap.
 	var commitErrors []error
+	anyReadOnly := false
 	for _, server := range t.vacuumTargets {
-		if err := t.commitOne(ctx, server); err != nil {
+		resp, err := t.commitOne(ctx, server)
+		if err != nil {
 			glog.Errorf("vacuum commit on %s for volume %d: %v", server, t.volumeID, err)
 			commitErrors = append(commitErrors, fmt.Errorf("%s: %w", server, err))
+			continue
+		}
+		if resp.GetIsReadOnly() {
+			anyReadOnly = true
 		}
 	}
 	if len(commitErrors) > 0 {
 		return fmt.Errorf("vacuum commit failed on %d/%d replicas: %v",
 			len(commitErrors), len(t.vacuumTargets), commitErrors)
 	}
+
+	// Phase 3: re-notify the master so the volume re-enters the writable
+	// set. The worker's only lever is VolumeMarkWritable, which clears the
+	// read-only flag and runs notifyMasterVolumeReadonly(false). Gate on
+	// the commit's IsReadOnly exactly as SetVolumeAvailable does: a replica
+	// still read-only (operator-set, EIO-quarantined, or disk-space-low)
+	// must stay out, and recovers on its own via the next ReadOnly=false
+	// heartbeat — force-clearing the flag here would override that. The
+	// worker is not told the master's size limit, so the isFullCapacity
+	// guard is left to the next capacity heartbeat. Best-effort: the vacuum
+	// itself already succeeded.
+	if anyReadOnly {
+		glog.V(0).Infof("post-vacuum: volume %d still read-only on a replica, leaving it out of writables", t.volumeID)
+		return nil
+	}
+	for _, server := range t.vacuumTargets {
+		if err := t.markWritableOne(ctx, server); err != nil {
+			glog.Warningf("post-vacuum mark writable on %s for volume %d: %v", server, t.volumeID, err)
+			continue
+		}
+		glog.V(0).Infof("post-vacuum marked volume %d writable on %s", t.volumeID, server)
+	}
 	return nil
 }
 
@@ -302,13 +337,15 @@ func (t *VacuumTask) compactOne(ctx context.Context, server string) error {
 		})
 }
 
-func (t *VacuumTask) commitOne(ctx context.Context, server string) error {
-	return operation.WithVolumeServerClient(false, pb.ServerAddress(server), t.grpcDialOption,
+func (t *VacuumTask) commitOne(ctx context.Context, server string) (*volume_server_pb.VacuumVolumeCommitResponse, error) {
+	var resp *volume_server_pb.VacuumVolumeCommitResponse
+	err := operation.WithVolumeServerClient(false, pb.ServerAddress(server), t.grpcDialOption,
 		func(client volume_server_pb.VolumeServerClient) error {
 			t.GetLogger().Info("Committing vacuum on %s", server)
 			commitCtx, cancel := context.WithTimeout(ctx, t.vacuumTimeout(time.Minute))
 			defer cancel()
-			_, err := client.VacuumVolumeCommit(commitCtx, &volume_server_pb.VacuumVolumeCommitRequest{
+			var err error
+			resp, err = client.VacuumVolumeCommit(commitCtx, &volume_server_pb.VacuumVolumeCommitRequest{
 				VolumeId: t.volumeID,
 			})
 			if err != nil {
@@ -316,6 +353,7 @@ func (t *VacuumTask) commitOne(ctx context.Context, server string) error {
 			}
 			return nil
 		})
+	return resp, err
 }
 
 func (t *VacuumTask) cleanupOne(ctx context.Context, server string) error {
@@ -330,6 +368,23 @@ func (t *VacuumTask) cleanupOne(ctx context.Context, server string) error {
 		})
 }
 
+func (t *VacuumTask) markWritableOne(ctx context.Context, server string) error {
+	return operation.WithVolumeServerClient(false, pb.ServerAddress(server), t.grpcDialOption,
+		func(client volume_server_pb.VolumeServerClient) error {
+			// VolumeMarkWritable is a metadata RPC (reopen idx + flags +
+			// notifyMasterVolumeReadonly heartbeat) — millisecond-scale and
+			// independent of volume size. A flat 1m cap prevents an
+			// unresponsive replica from blocking Phase 3 for hours on a
+			// TB-scale volume where vacuumTimeout() would balloon.
+			markCtx, cancel := context.WithTimeout(ctx, time.Minute)
+			defer cancel()
+			_, err := client.VolumeMarkWritable(markCtx, &volume_server_pb.VolumeMarkWritableRequest{
+				VolumeId: t.volumeID,
+			})
+			return err
+		})
+}
+
 // cleanupAll removes the .cpd/.cpx/.cpldb temp files on every target.
 // Used to roll back when Compact fails on one replica after others
 // have already created their temp files. Per-target failures are