diff --git a/test/plugin_workers/fake_volume_server.go b/test/plugin_workers/fake_volume_server.go index 6d3ce469f..574da7f6a 100644 --- a/test/plugin_workers/fake_volume_server.go +++ b/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) { diff --git a/test/plugin_workers/vacuum/execution_test.go b/test/plugin_workers/vacuum/execution_test.go index e687a2dec..7a6a69dab 100644 --- a/test/plugin_workers/vacuum/execution_test.go +++ b/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()) } diff --git a/weed/worker/tasks/vacuum/vacuum_task.go b/weed/worker/tasks/vacuum/vacuum_task.go index c56a8d5c9..e37c98c92 100644 --- a/weed/worker/tasks/vacuum/vacuum_task.go +++ b/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