From 1167ca8cf5aa28d2f48f3164df9e8722f52e7f98 Mon Sep 17 00:00:00 2001 From: Yip Leo Date: Sun, 15 Nov 2026 03:07:44 +0800 Subject: [PATCH] vacuum: report a request dropped because another vacuum is running (#12648) * vacuum: report a request dropped because another vacuum is running The master runs one vacuum at a time. A request that arrives while another one is running, the automatic vacuum or an earlier request, is dropped: Topology.Vacuum logs "Vacuum is already running" and returns. The VacuumVolume RPC and /vol/vacuum then answered as if the vacuum had run, so volume.vacuum printed nothing and exited 0 with no volume compacted. Where the automatic vacuum is busy for hours, an explicit volume.vacuum -volumeId could do nothing every time, and nothing said so. Topology.Vacuum now returns ErrVacuumRunning for a dropped request. The RPC returns it and /vol/vacuum answers 409 with it. volume.vacuum names the request that failed, because a list of volumes or collections is sent as several requests and a vacuum that starts in between fails only the later ones. Requests are still dropped, not queued, and the automatic vacuum ignores the error as before. The integration tests that need their /vol/vacuum request to run ask again while the master answers 409. Co-Authored-By: Claude Fable 5.2 * docs: note the 409 answer from /vol/vacuum while another vacuum runs Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Claude Fable 5.2 Co-authored-by: Chris Lu --- REST_API.md | 4 + .../s3_copy_versioning_regression_test.go | 22 ++++-- test/tus/tus_integration_test.go | 20 +++-- weed/server/master_grpc_server_volume.go | 6 +- weed/server/master_server_handlers_admin.go | 5 +- weed/shell/command_volume_vacuum.go | 20 ++++- weed/topology/topology_vacuum.go | 15 +++- weed/topology/topology_vacuum_test.go | 77 +++++++++++++++++++ 8 files changed, 153 insertions(+), 16 deletions(-) diff --git a/REST_API.md b/REST_API.md index a8315a5fd5..97df87757c 100644 --- a/REST_API.md +++ b/REST_API.md @@ -271,6 +271,10 @@ curl "http://localhost:9333/vol/vacuum?garbageThreshold=0.4" Vacuuming makes a volume read-only, copies live needles to a new volume, and swaps it in. +The master runs one vacuum at a time. A request that arrives while the +automatic vacuum or another request is running does nothing and gets +`409 Conflict` with `{"error":"another vacuum is running, try again later"}`. + ### Delete a collection ```bash diff --git a/test/s3/versioning/s3_copy_versioning_regression_test.go b/test/s3/versioning/s3_copy_versioning_regression_test.go index 92bbc7864b..1e2a60974f 100644 --- a/test/s3/versioning/s3_copy_versioning_regression_test.go +++ b/test/s3/versioning/s3_copy_versioning_regression_test.go @@ -44,12 +44,22 @@ func vacuumVolumes(t *testing.T) { require.NotEmpty(t, defaultConfig.MasterEndpoint, "vacuum needs a master endpoint; set MASTER_ENDPOINT") endpoint := strings.TrimRight(defaultConfig.MasterEndpoint, "/") + "/vol/vacuum?garbageThreshold=0.001" httpClient := &http.Client{Timeout: 30 * time.Second} - resp, err := httpClient.Get(endpoint) - require.NoError(t, err, "vacuum request to %s", endpoint) - defer resp.Body.Close() - _, err = io.Copy(io.Discard, resp.Body) - require.NoError(t, err, "reading the vacuum response") - require.Equal(t, http.StatusOK, resp.StatusCode, "vacuum request to %s", endpoint) + // The master runs one vacuum at a time and answers 409 to a request it + // dropped because its automatic vacuum was running. Ask until ours runs. + deadline := time.Now().Add(time.Minute) + for { + resp, err := httpClient.Get(endpoint) + require.NoError(t, err, "vacuum request to %s", endpoint) + _, err = io.Copy(io.Discard, resp.Body) + resp.Body.Close() + require.NoError(t, err, "reading the vacuum response") + if resp.StatusCode == http.StatusConflict && time.Now().Before(deadline) { + time.Sleep(500 * time.Millisecond) + continue + } + require.Equal(t, http.StatusOK, resp.StatusCode, "vacuum request to %s", endpoint) + return + } } func requireVersionBody(t *testing.T, client *s3.Client, bucketName, objectKey, versionId string, want []byte, msg string) { diff --git a/test/tus/tus_integration_test.go b/test/tus/tus_integration_test.go index cf9456325c..1cc3390a93 100644 --- a/test/tus/tus_integration_test.go +++ b/test/tus/tus_integration_test.go @@ -983,11 +983,21 @@ func TestTusAbortedPatchKeepsStoredChunks(t *testing.T) { require.Equal(t, http.StatusNoContent, patchResp.StatusCode) // A vacuum reclaims whatever the filer deleted, so the file survives this - // only if the chunks the session kept are still stored. - vacuumResp, err := client.Get(fmt.Sprintf("http://127.0.0.1:%s/vol/vacuum?garbageThreshold=0.001", testMasterPort)) - require.NoError(t, err) - vacuumResp.Body.Close() - require.Equal(t, http.StatusOK, vacuumResp.StatusCode) + // only if the chunks the session kept are still stored. The master answers + // 409 to a request it dropped because its automatic vacuum was running, so + // ask until ours runs. + vacuumDeadline := time.Now().Add(time.Minute) + for { + vacuumResp, err := client.Get(fmt.Sprintf("http://127.0.0.1:%s/vol/vacuum?garbageThreshold=0.001", testMasterPort)) + require.NoError(t, err) + vacuumResp.Body.Close() + if vacuumResp.StatusCode == http.StatusConflict && time.Now().Before(vacuumDeadline) { + time.Sleep(500 * time.Millisecond) + continue + } + require.Equal(t, http.StatusOK, vacuumResp.StatusCode) + break + } getResp, err := client.Get(cluster.FilerURL() + targetPath) require.NoError(t, err) diff --git a/weed/server/master_grpc_server_volume.go b/weed/server/master_grpc_server_volume.go index 96b60182f9..099541c583 100644 --- a/weed/server/master_grpc_server_volume.go +++ b/weed/server/master_grpc_server_volume.go @@ -407,7 +407,11 @@ func (ms *MasterServer) VacuumVolume(ctx context.Context, req *master_pb.VacuumV resp := &master_pb.VacuumVolumeResponse{} - ms.Topo.Vacuum(ms.grpcDialOption, float64(req.GarbageThreshold), ms.option.MaxParallelVacuumPerServer, ms.option.MaxParallelVacuumPerRack, req.VolumeId, req.Collection, ms.preallocateSize, false, 0) + // A request that lost the vacuum guard to the automatic vacuum or to another + // request did nothing. Returning success would tell the caller it ran. + if err := ms.Topo.Vacuum(ms.grpcDialOption, float64(req.GarbageThreshold), ms.option.MaxParallelVacuumPerServer, ms.option.MaxParallelVacuumPerRack, req.VolumeId, req.Collection, ms.preallocateSize, false, 0); err != nil { + return nil, err + } return resp, nil } diff --git a/weed/server/master_server_handlers_admin.go b/weed/server/master_server_handlers_admin.go index e6479ea49a..3bd8d4c863 100644 --- a/weed/server/master_server_handlers_admin.go +++ b/weed/server/master_server_handlers_admin.go @@ -61,7 +61,10 @@ func (ms *MasterServer) volumeVacuumHandler(w http.ResponseWriter, r *http.Reque } } // glog.Infoln("garbageThreshold =", gcThreshold) - ms.Topo.Vacuum(ms.grpcDialOption, gcThreshold, ms.option.MaxParallelVacuumPerServer, ms.option.MaxParallelVacuumPerRack, 0, "", ms.preallocateSize, false, 0) + if err := ms.Topo.Vacuum(ms.grpcDialOption, gcThreshold, ms.option.MaxParallelVacuumPerServer, ms.option.MaxParallelVacuumPerRack, 0, "", ms.preallocateSize, false, 0); err != nil { + writeJsonError(w, r, http.StatusConflict, err) + return + } ms.dirStatusHandler(w, r) } diff --git a/weed/shell/command_volume_vacuum.go b/weed/shell/command_volume_vacuum.go index cd93dcddd5..965f7e3b99 100644 --- a/weed/shell/command_volume_vacuum.go +++ b/weed/shell/command_volume_vacuum.go @@ -38,6 +38,10 @@ func (c *commandVacuum) Help() string { Without -volumeId this runs the same sweep as the automatic vacuum, which skips read-only volumes. Name a read-only volume with -volumeId to vacuum it anyway. + The master runs one vacuum at a time. While the automatic vacuum or another + request is running, this command fails instead of waiting for its turn. Run it + again later, or pause the automatic vacuum first with volume.vacuum.disable. + ` } @@ -157,13 +161,27 @@ func (c *commandVacuum) Do(args []string, commandEnv *CommandEnv, writer io.Writ return err }) if err != nil { - return err + // A list of volumes or collections is several requests, and a vacuum + // that starts in between only fails the ones after it. + return fmt.Errorf("vacuum %s: %w", describeVacuumRequest(req), err) } } return nil } +func describeVacuumRequest(req *master_pb.VacuumVolumeRequest) string { + switch { + case req.VolumeId != 0 && req.Collection != "": + return fmt.Sprintf("volume %d in collection %s", req.VolumeId, req.Collection) + case req.VolumeId != 0: + return fmt.Sprintf("volume %d", req.VolumeId) + case req.Collection != "": + return "collection " + req.Collection + } + return "all volumes" +} + // vidReplica pairs a volume replica's read-only state with the server that // reports it, so the unnamed-collection sweep can reproduce the master's // low-disk-space exception client-side. diff --git a/weed/topology/topology_vacuum.go b/weed/topology/topology_vacuum.go index 83144cea1d..6428e0259e 100644 --- a/weed/topology/topology_vacuum.go +++ b/weed/topology/topology_vacuum.go @@ -2,6 +2,7 @@ package topology import ( "context" + "errors" "io" "sync" "sync/atomic" @@ -255,13 +256,22 @@ func (t *Topology) batchVacuumVolumeCleanup(grpcDialOption grpc.DialOption, vl * } } -func (t *Topology) Vacuum(grpcDialOption grpc.DialOption, garbageThreshold float64, maxParallelVacuumPerServer int, maxParallelVacuumPerRack int, volumeId uint32, collection string, preallocate int64, automatic bool, deleteEmptyAfter time.Duration) { +// ErrVacuumRunning is what Vacuum returns when another vacuum holds the +// topology-wide guard. The request was dropped, not queued: nothing waits for +// the running vacuum, so the caller has to ask again. +var ErrVacuumRunning = errors.New("another vacuum is running, try again later") + +// Vacuum runs one vacuum at a time. A call that finds another one running, +// the automatic vacuum or an earlier request, does nothing and returns +// ErrVacuumRunning, so that a request for a volume cannot look done when it +// never started. +func (t *Topology) Vacuum(grpcDialOption grpc.DialOption, garbageThreshold float64, maxParallelVacuumPerServer int, maxParallelVacuumPerRack int, volumeId uint32, collection string, preallocate int64, automatic bool, deleteEmptyAfter time.Duration) error { // if there is vacuum going on, return immediately swapped := atomic.CompareAndSwapInt64(&t.vacuumLockCounter, 0, 1) if !swapped { glog.V(0).Infof("Vacuum is already running") - return + return ErrVacuumRunning } defer atomic.StoreInt64(&t.vacuumLockCounter, 0) @@ -301,6 +311,7 @@ func (t *Topology) Vacuum(grpcDialOption grpc.DialOption, garbageThreshold float break } } + return nil } func (t *Topology) vacuumOneVolumeLayout(grpcDialOption grpc.DialOption, volumeLayout *VolumeLayout, c *Collection, garbageThreshold float64, maxParallelVacuumPerServer int, maxParallelVacuumPerRack int, preallocate int64, automatic bool, deleteEmptyAfter time.Duration) { diff --git a/weed/topology/topology_vacuum_test.go b/weed/topology/topology_vacuum_test.go index 106011d594..0b909f1954 100644 --- a/weed/topology/topology_vacuum_test.go +++ b/weed/topology/topology_vacuum_test.go @@ -518,6 +518,83 @@ func TestVacuumStalledVolumeServerReleasesGuard(t *testing.T) { } } +// A request that finds another vacuum running is dropped, not queued. It has +// to say so: returning as if it had run leaves the caller believing its volume +// was compacted. +func TestVacuumReportsAnotherVacuumRunning(t *testing.T) { + fake := &fakeVacuumServer{ + checks: map[uint32]*volume_server_pb.VacuumVolumeCheckResponse{ + 1: {GarbageRatio: 0.9}, + 2: {GarbageRatio: 0.9}, + }, + entered: make(chan string, 1), + compactGate: make(chan struct{}), + compactTracker: &compactTracker{}, + } + grpcPort, dialOption := startFakeVolumeServer(t, fake) + + topo := NewTopology("weedfs", sequence.NewMemorySequencer(), 32*1024, 5, false) + dn := topo.GetOrCreateDataCenter("dc1").GetOrCreateRack("rack1"). + GetOrCreateDataNode("127.0.0.1", 8080, grpcPort, "127.0.0.1", "dn", map[string]uint32{"": 10}) + var volumes []storage.VolumeInfo + for vid := 1; vid <= 2; vid++ { + volumes = append(volumes, storage.VolumeInfo{ + Id: needle.VolumeId(vid), + Size: 1 << 20, + Collection: "c", + ModifiedAtSecond: time.Now().Unix(), + Version: needle.GetCurrentVersion(), + ReplicaPlacement: &super_block.ReplicaPlacement{}, + Ttl: needle.EMPTY_TTL, + }) + } + dn.UpdateVolumes(volumes) + for _, v := range volumes { + topo.RegisterVolumeLayout(v, dn) + } + committed := func() []uint32 { + fake.mu.Lock() + defer fake.mu.Unlock() + return append([]uint32(nil), fake.committed...) + } + + // the first request holds the guard for as long as its compaction is gated + first := make(chan error, 1) + go func() { + first <- topo.Vacuum(dialOption, 0.3, 1, 0, 1, "c", 0, false, 0) + }() + select { + case <-fake.entered: + case <-time.After(15 * time.Second): + t.Fatal("the first vacuum never reached the compact phase") + } + + if err := topo.Vacuum(dialOption, 0.3, 1, 0, 2, "c", 0, false, 0); !errors.Is(err, ErrVacuumRunning) { + t.Fatalf("vacuum of volume 2 while volume 1 is compacting: err = %v, want ErrVacuumRunning", err) + } + if got := committed(); len(got) != 0 { + t.Fatalf("committed volumes = %v, want none while the first vacuum is gated", got) + } + + close(fake.compactGate) + select { + case err := <-first: + if err != nil { + t.Fatalf("the first vacuum: %v", err) + } + case <-time.After(15 * time.Second): + t.Fatal("the first vacuum did not finish after the gate opened") + } + + // with the guard free again, the same request runs + if err := topo.Vacuum(dialOption, 0.3, 1, 0, 2, "c", 0, false, 0); err != nil { + t.Fatalf("vacuum of volume 2 after the first one finished: %v", err) + } + if got := committed(); len(got) != 2 || got[0] != 1 || got[1] != 2 { + t.Fatalf("committed volumes = %v, want [1 2]", got) + } +} + // A replica tiered to remote storage is never compacted by a vacuum, not even // on an explicit request with a zero threshold: its .dat is the remote object. // volume.tier.compact is the way to reclaim its garbage.