From c86df14f4d2d9a698aede039d3f2818fb17e1ee7 Mon Sep 17 00:00:00 2001 From: Feng Ruohang Date: Sun, 25 Oct 2026 13:22:27 +0800 Subject: [PATCH] test: pin abandoned cleanup races with commit and undo Signed-off-by: Feng Ruohang --- cmd/abandoned-mutation_test.go | 182 +++++++++++++++++++++++++++++++++ 1 file changed, 182 insertions(+) create mode 100644 cmd/abandoned-mutation_test.go diff --git a/cmd/abandoned-mutation_test.go b/cmd/abandoned-mutation_test.go new file mode 100644 index 000000000..2822eaf43 --- /dev/null +++ b/cmd/abandoned-mutation_test.go @@ -0,0 +1,182 @@ +// Copyright (c) 2026 Feng Ruohang +// SPDX-License-Identifier: AGPL-3.0-or-later + +package cmd + +import ( + "context" + "io" + "path/filepath" + "strings" + "sync" + "testing" + "time" +) + +type abandonedCleanupCommitDisk struct { + StorageAPI + entered chan<- struct{} + release <-chan struct{} +} + +func (d abandonedCleanupCommitDisk) RenameData(ctx context.Context, sv, sp string, fi FileInfo, dv, dp string, o RenameOptions) (RenameDataResp, error) { + var success int64 + gate := &scopeStagedContext{ + Context: ctx, meta: filepath.Join(d.Endpoint().Path, dv, dp, fi.DataDir, "part.1"), + entered: d.entered, release: d.release, + tracker: healthDiskCtxValue{lastSuccess: &success}, + } + return d.StorageAPI.RenameData(gate, sv, sp, fi, dv, dp, o) +} + +// A lost namespace lease can admit cleanup while a disk transaction is between +// moving its external shards and replacing xl.meta. Those shards are live work, +// even though the old metadata does not refer to them yet. Cleanup must wait. +func TestMutationAbandonedCleanupWaitsForCommit(t *testing.T) { + z, bucket := mutationPools(t) + const object = "cleanup-during-commit" + body := strings.Repeat("new payload", 200000) + mutationPut(t, z, bucket, object, 0, strings.Repeat("old payload", 200000), time.Now().Add(-time.Minute)) + set := z.serverPools[0].getHashedSet(object) + disks, swap := installListingDiskSwitch(set) + defer swap(disks) + entered, release := make(chan struct{}, len(disks)), make(chan struct{}) + unblock := sync.OnceFunc(func() { close(release) }) + defer unblock() + wrapped := append([]StorageAPI(nil), disks...) + for i, d := range disks { + wrapped[i] = abandonedCleanupCommitDisk{d, entered, release} + } + swap(wrapped) + committed := make(chan error, 1) + go func() { + _, err := z.serverPools[0].PutObject(t.Context(), bucket, object, mustGetPutObjReader(t, strings.NewReader(body), int64(len(body)), "", ""), ObjectOptions{NoLock: true}) + committed <- err + }() + for range disks { + select { + case <-entered: + case err := <-committed: + t.Fatalf("PUT returned before shard barrier: %v", err) + case <-time.After(5 * time.Second): + t.Fatal("PUT did not move all external shards") + } + } + t.Log("All eight new external shard directories exist; old xl.meta is still active") + cleaned := make(chan error, len(disks)) + for _, d := range disks { + go func(d StorageAPI) { cleaned <- d.CleanAbandonedData(context.Background(), bucket, object) }(d) + } + early := 0 + timer := time.NewTimer(150 * time.Millisecond) +wait: + for { + select { + case err := <-cleaned: + early++ + t.Errorf("abandoned-data cleanup overtook unfinished commit: %v", err) + if early == len(disks) { + break wait + } + case <-timer.C: + break wait + } + } + timer.Stop() + unblock() + select { + case err := <-committed: + if err != nil { + t.Fatalf("PUT failed: %v", err) + } + case <-time.After(5 * time.Second): + t.Fatal("PUT did not finish") + } + for i := early; i < len(disks); i++ { + select { + case err := <-cleaned: + if err != nil { + t.Fatal(err) + } + case <-time.After(5 * time.Second): + t.Fatal("cleanup did not finish") + } + } + swap(disks) + r, err := z.GetObjectNInfo(t.Context(), bucket, object, nil, nil, ObjectOptions{}) + if err != nil { + t.Fatalf("acknowledged PUT unreadable after abandoned-data cleanup: %v", err) + } + defer r.Close() + data, err := io.ReadAll(r) + if err != nil || string(data) != body { + t.Fatalf("acknowledged external shards lost: bytes=%d err=%v", len(data), err) + } +} + +// Cleanup can also arrive after a disk has committed but before its delayed +// reply lets the coordinator decide to undo a failed quorum. The previous +// generation's backup is still owned by that unfinished transaction. +func TestMutationAbandonedCleanupPreservesUndoBackup(t *testing.T) { + z, bucket := mutationPools(t) + const object = "cleanup-before-undo" + body := strings.Repeat("last acknowledged", 150000) + base := time.Now().Add(-time.Minute) + mutationPut(t, z, bucket, object, 0, body, base) + set := z.serverPools[0].getHashedSet(object) + disks, swap := installListingDiskSwitch(set) + defer swap(disks) + entered, release := make(chan struct{}, 4), make(chan struct{}) + unblock := sync.OnceFunc(func() { close(release) }) + defer unblock() + wrapped := append([]StorageAPI(nil), disks...) + for i, d := range disks { + wrapped[i] = delayedRenameResult{d, i >= 4, entered, release} + } + swap(wrapped) + committed := make(chan error, 1) + go func() { + next := strings.Repeat("failed update", 180000) + _, err := z.serverPools[0].PutObject(t.Context(), bucket, object, mustGetPutObjReader(t, strings.NewReader(next), int64(len(next)), "", ""), ObjectOptions{NoLock: true, MTime: base.Add(time.Second)}) + committed <- err + }() + for range 4 { + select { + case <-entered: + case err := <-committed: + t.Fatalf("PUT returned before reply barrier: %v", err) + case <-time.After(5 * time.Second): + t.Fatal("disk commit did not reach reply barrier") + } + } + for _, d := range disks[:4] { + if err := d.CleanAbandonedData(t.Context(), bucket, object); err != nil { + t.Fatal(err) + } + } + unblock() + select { + case err := <-committed: + if err == nil { + t.Fatal("four-of-eight commit unexpectedly reached write quorum") + } + case <-time.After(5 * time.Second): + t.Fatal("failed PUT did not finish undo") + } + swap(disks) + for i, d := range disks { + fi, err := d.ReadVersion(t.Context(), "", bucket, object, "", ReadOptions{}) + if err != nil || !fi.ModTime.Equal(base) { + t.Errorf("disk%d lost last acknowledged generation after failed PUT: mtime=%v err=%v", i, fi.ModTime, err) + } + } + r, err := z.GetObjectNInfo(t.Context(), bucket, object, nil, nil, ObjectOptions{}) + if err != nil { + t.Fatalf("last acknowledged value unreadable after cleanup and undo: %v", err) + } + defer r.Close() + data, err := io.ReadAll(r) + if err != nil || string(data) != body { + t.Fatalf("last acknowledged value lost after cleanup and undo: bytes=%d err=%v", len(data), err) + } +}