From 81c90cb446480cab38bd93ebf768ce68e92e79b9 Mon Sep 17 00:00:00 2001 From: Konstantin Adzer <47555435+AdzerKI@users.noreply.github.com> Date: Sat, 24 Oct 2026 18:22:29 +0300 Subject: [PATCH] filer.backup: retry a failed checkpoint read instead of replaying from the epoch (#12036) * filer.backup: retry a failed checkpoint read instead of replaying from the epoch Fixes #12035 * test: reset sink.local.directory in backup offset test cleanup The new test sets sink.local.directory on the shared viper but only reset the enabled flag, leaking a key that points at a deleted t.TempDir() to later tests in the package. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test: cover the retried attempt re-reading the checkpoint The fake filer now fails only its first KvGet, and the test runs a second doFilerBackup attempt asserting the checkpoint is re-read and the attempt proceeds past the offset stage. LookupDirectoryEntry is stubbed so the SSE-S3 KEK probe returns a miss instead of retrying 10x2s. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --------- Co-authored-by: user.email Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- weed/command/filer_backup.go | 2 +- weed/command/filer_backup_offset_test.go | 61 +++++++++++++++++++++++- 2 files changed, 61 insertions(+), 2 deletions(-) diff --git a/weed/command/filer_backup.go b/weed/command/filer_backup.go index 060c5fc837..f184dd23dc 100644 --- a/weed/command/filer_backup.go +++ b/weed/command/filer_backup.go @@ -156,7 +156,7 @@ func doFilerBackup(grpcDialOption grpc.DialOption, backupOption *FilerBackupOpti } else { lastOffsetTsNs, err := getOffsetWithFallback(grpcDialOption, sourceFiler, BackupKeyPrefix, sinkId, BackupKeyPrefix, legacySinkId) if err != nil { - glog.V(0).Infof("starting from %v (offset read failed: %v)", startFrom, err) + return 0, fmt.Errorf("read backup offset: %w", err) } else if lastOffsetTsNs > 0 { startFrom = time.Unix(0, lastOffsetTsNs) glog.V(0).Infof("resuming from %v", startFrom) diff --git a/weed/command/filer_backup_offset_test.go b/weed/command/filer_backup_offset_test.go index 7ae18a3c36..e42efdaa2b 100644 --- a/weed/command/filer_backup_offset_test.go +++ b/weed/command/filer_backup_offset_test.go @@ -3,11 +3,14 @@ package command import ( "context" "net" + "strings" "sync" + "sync/atomic" "testing" "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/util" "google.golang.org/grpc" "google.golang.org/grpc/credentials/insecure" ) @@ -34,14 +37,35 @@ func (s *kvFilerServer) KvPut(_ context.Context, req *filer_pb.KvPutRequest) (*f return &filer_pb.KvPutResponse{}, nil } +type failingKvFilerServer struct { + filer_pb.UnimplementedSeaweedFilerServer + calls atomic.Int32 +} + +func (s *failingKvFilerServer) KvGet(context.Context, *filer_pb.KvGetRequest) (*filer_pb.KvGetResponse, error) { + if s.calls.Add(1) == 1 { + return &filer_pb.KvGetResponse{Error: "kv store unavailable"}, nil + } + return &filer_pb.KvGetResponse{}, nil +} + +// a miss reads as ErrNotFound so the retried attempt skips the KEK retry loop +func (s *failingKvFilerServer) LookupDirectoryEntry(context.Context, *filer_pb.LookupDirectoryEntryRequest) (*filer_pb.LookupDirectoryEntryResponse, error) { + return &filer_pb.LookupDirectoryEntryResponse{}, nil +} + func startKvFiler(t *testing.T) (pb.ServerAddress, grpc.DialOption) { + return startFiler(t, &kvFilerServer{kv: map[string][]byte{}}) +} + +func startFiler(t *testing.T, server filer_pb.SeaweedFilerServer) (pb.ServerAddress, grpc.DialOption) { t.Helper() listener, err := net.Listen("tcp", "127.0.0.1:0") if err != nil { t.Fatalf("listen: %v", err) } grpcServer := grpc.NewServer() - filer_pb.RegisterSeaweedFilerServer(grpcServer, &kvFilerServer{kv: map[string][]byte{}}) + filer_pb.RegisterSeaweedFilerServer(grpcServer, server) go func() { _ = grpcServer.Serve(listener) }() t.Cleanup(func() { grpcServer.Stop() }) grpcPort := listener.Addr().(*net.TCPAddr).Port @@ -110,3 +134,38 @@ func TestBackupOffset_DistinctDestinationsDoNotShare(t *testing.T) { t.Fatalf("B reads (%d, %v), want (2000, nil)", got, err) } } + +func TestFilerBackup_OffsetReadFailureFailsAttempt(t *testing.T) { + server := &failingKvFilerServer{} + filerAddr, dial := startFiler(t, server) + + config := util.GetViper() + config.Set("sink.local.enabled", true) + config.Set("sink.local.directory", t.TempDir()) + t.Cleanup(func() { + config.Set("sink.local.enabled", false) + config.Set("sink.local.directory", "") + }) + + options := filerBackupOptions + address := string(filerAddr) + options.filer = &address + + resumeTsNs, err := doFilerBackup(dial, &options, nil, nil, nil, 1, 1, 0) + if err == nil || !strings.Contains(err.Error(), "kv store unavailable") { + t.Fatalf("doFilerBackup error = %v, want the offset read failure", err) + } + if resumeTsNs != 0 { + t.Fatalf("resume position = %d, want 0", resumeTsNs) + } + + // the next attempt re-reads the checkpoint and gets past it once the KV + // recovers, so it must fail later, not on the offset read + _, err = doFilerBackup(dial, &options, nil, nil, nil, 1, 2, resumeTsNs) + if err == nil || strings.Contains(err.Error(), "kv store unavailable") { + t.Fatalf("retried doFilerBackup error = %v, want a later-stage failure", err) + } + if got := server.calls.Load(); got < 2 { + t.Fatalf("checkpoint reads across attempts = %d, want a re-read", got) + } +}