From e0f84382840f3fd467d17ef3dadb352a6d5e64d5 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Thu, 12 Nov 2026 08:57:42 -0800 Subject: [PATCH] mount: retry a writeback metadata flush until the filer answers (#12578) * mount: keep the filer status in the flush create error flushMetadataToFiler formatted the CreateEntry error with %v, so callers classifying it saw only the message, which includes the file path. Wrap it so the gRPC status the filer sent stays reachable. Co-Authored-By: Claude Opus 5.5 * mount: retry a writeback metadata flush until the filer answers With -writebackCache, close() returns before the chunk references reach the filer. The deferred flush gave up after four attempts, about a minute once each attempt's own filer retries are counted, then released the handle: a longer filer outage dropped the file even though write() and close() had succeeded, and its uploaded chunks became orphans. Keep retrying for as long as the filer is unreachable, as the writeback chunk uploads already do. A filer that answers with an error still ends the flush after the bounded attempts, and an unlink of the draining file still ends it on the next attempt. Co-Authored-By: Claude Opus 5.5 * mount: retry only an unanswered filer call without a bound The unbounded phase used util.IsTransientError, which falls back to matching the error text. A refusal the filer sends in its response (read-only, name too long, a plain error string) has no gRPC status, and the flush error carries the file path, so a file named like "unavailable-transport.log" kept retrying against a filer that answered. Retry without a bound only when the call failed with a gRPC Unavailable status: the dial, the connection or the mount's call deadline gave out. Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Claude Opus 5.5 --- weed/mount/metadata_flush_retry.go | 20 ++++ weed/mount/metadata_flush_retry_test.go | 120 ++++++++++++++++++++++++ weed/mount/weedfs.go | 2 +- weed/mount/weedfs_async_flush.go | 30 ++++-- weed/mount/weedfs_async_flush_test.go | 103 ++++++++++++++++++++ weed/mount/weedfs_file_sync.go | 2 +- 6 files changed, 269 insertions(+), 8 deletions(-) create mode 100644 weed/mount/weedfs_async_flush_test.go diff --git a/weed/mount/metadata_flush_retry.go b/weed/mount/metadata_flush_retry.go index 8e14cffee8..3092b3542c 100644 --- a/weed/mount/metadata_flush_retry.go +++ b/weed/mount/metadata_flush_retry.go @@ -3,6 +3,8 @@ package mount import ( "context" "time" + + "github.com/seaweedfs/seaweedfs/weed/util" ) const metadataFlushRetries = 3 @@ -21,6 +23,24 @@ func retryMetadataFlush(ctx context.Context, flush func() error, onRetry func(ne return retryMetadataFlushIf(ctx, flush, nil, onRetry) } +// retryMetadataFlushUntil runs retryMetadataFlush, then keeps retrying every +// util.RetryWaitTime for as long as keepRetrying accepts the error. A zero +// totalAttempts in onRetry marks those unbounded attempts. +func retryMetadataFlushUntil(ctx context.Context, flush func() error, keepRetrying func(error) bool, onRetry func(nextAttempt, totalAttempts int, backoff time.Duration, err error)) error { + err := retryMetadataFlush(ctx, flush, onRetry) + for attempt := metadataFlushRetries + 2; err != nil && keepRetrying(err) && ctx.Err() == nil; attempt++ { + if onRetry != nil { + onRetry(attempt, 0, util.RetryWaitTime, err) + } + metadataFlushSleep(ctx, util.RetryWaitTime) + if ctx.Err() != nil { + break + } + err = flush() + } + return err +} + // retryMetadataFlushIf retries flush with exponential backoff, stopping early // when shouldRetry returns false (clearly permanent errors) or when ctx is // cancelled (the FUSE request was interrupted, e.g. the process was killed). diff --git a/weed/mount/metadata_flush_retry_test.go b/weed/mount/metadata_flush_retry_test.go index c72032f249..4f0c6d3755 100644 --- a/weed/mount/metadata_flush_retry_test.go +++ b/weed/mount/metadata_flush_retry_test.go @@ -5,6 +5,8 @@ import ( "errors" "testing" "time" + + "github.com/seaweedfs/seaweedfs/weed/util" ) func TestRetryMetadataFlushEventuallySucceeds(t *testing.T) { @@ -116,3 +118,121 @@ func TestRetryMetadataFlushStopsOnCancel(t *testing.T) { t.Fatalf("sleeps = %v, want none (no backoff after cancel)", sleeps) } } + +func TestRetryMetadataFlushUntilOutlastsOutage(t *testing.T) { + originalSleep := metadataFlushSleep + t.Cleanup(func() { + metadataFlushSleep = originalSleep + }) + + var sleeps []time.Duration + metadataFlushSleep = func(ctx context.Context, d time.Duration) { + sleeps = append(sleeps, d) + } + + const failures = 10 + attempts := 0 + var unbounded []int + err := retryMetadataFlushUntil(context.Background(), func() error { + attempts++ + if attempts <= failures { + return errors.New("filer unreachable") + } + return nil + }, func(error) bool { return true }, func(nextAttempt, totalAttempts int, backoff time.Duration, err error) { + if totalAttempts == 0 { + unbounded = append(unbounded, nextAttempt) + } + }) + if err != nil { + t.Fatalf("retryMetadataFlushUntil returned error: %v", err) + } + if attempts != failures+1 { + t.Fatalf("attempts = %d, want %d", attempts, failures+1) + } + + wantSleeps := []time.Duration{time.Second, 2 * time.Second, 4 * time.Second} + for len(wantSleeps) < failures { + wantSleeps = append(wantSleeps, util.RetryWaitTime) + } + if len(sleeps) != len(wantSleeps) { + t.Fatalf("sleeps = %v, want %v", sleeps, wantSleeps) + } + for i, want := range wantSleeps { + if sleeps[i] != want { + t.Fatalf("sleep[%d] = %v, want %v", i, sleeps[i], want) + } + } + if len(unbounded) != failures-metadataFlushRetries || unbounded[0] != metadataFlushRetries+2 { + t.Fatalf("unbounded retry attempts = %v, want %d of them starting at %d", + unbounded, failures-metadataFlushRetries, metadataFlushRetries+2) + } +} + +func TestRetryMetadataFlushUntilStopsOnRejectedError(t *testing.T) { + originalSleep := metadataFlushSleep + t.Cleanup(func() { + metadataFlushSleep = originalSleep + }) + metadataFlushSleep = func(ctx context.Context, d time.Duration) {} + + transient := errors.New("filer unreachable") + permanent := errors.New("permission denied") + attempts := 0 + err := retryMetadataFlushUntil(context.Background(), func() error { + attempts++ + if attempts < 7 { + return transient + } + return permanent + }, func(err error) bool { return errors.Is(err, transient) }, nil) + if !errors.Is(err, permanent) { + t.Fatalf("retryMetadataFlushUntil error = %v, want %v", err, permanent) + } + if attempts != 7 { + t.Fatalf("attempts = %d, want 7", attempts) + } + + attempts = 0 + err = retryMetadataFlushUntil(context.Background(), func() error { + attempts++ + return permanent + }, func(err error) bool { return errors.Is(err, transient) }, nil) + if !errors.Is(err, permanent) { + t.Fatalf("retryMetadataFlushUntil error = %v, want %v", err, permanent) + } + if attempts != metadataFlushRetries+1 { + t.Fatalf("attempts = %d, want the bounded %d", attempts, metadataFlushRetries+1) + } +} + +func TestRetryMetadataFlushUntilStopsOnCancel(t *testing.T) { + originalSleep := metadataFlushSleep + t.Cleanup(func() { + metadataFlushSleep = originalSleep + }) + + ctx, cancel := context.WithCancel(context.Background()) + sleeps := 0 + metadataFlushSleep = func(ctx context.Context, d time.Duration) { + sleeps++ + if d == util.RetryWaitTime { + cancel() + } + } + + attempts := 0 + err := retryMetadataFlushUntil(ctx, func() error { + attempts++ + return errors.New("filer unreachable") + }, func(error) bool { return true }, nil) + if err == nil { + t.Fatal("retryMetadataFlushUntil returned nil after cancel") + } + if attempts != metadataFlushRetries+1 { + t.Fatalf("attempts = %d, want %d (none after cancel)", attempts, metadataFlushRetries+1) + } + if sleeps != metadataFlushRetries+1 { + t.Fatalf("sleeps = %d, want %d", sleeps, metadataFlushRetries+1) + } +} diff --git a/weed/mount/weedfs.go b/weed/mount/weedfs.go index 1453bdd2b4..2bfe531f86 100644 --- a/weed/mount/weedfs.go +++ b/weed/mount/weedfs.go @@ -317,7 +317,7 @@ func NewSeaweedFileSystem(option *Option) *WFS { // after server.Serve() would never execute. Drain here first. // // Use a timeout to avoid hanging on Ctrl-C if the filer is - // unreachable (metadata retry can take up to 7 seconds). + // unreachable: writeback flushes retry until it answers. // If the timeout expires, skip the write-cache removal so that // still-running goroutines can finish reading swap files. asyncDrained := true diff --git a/weed/mount/weedfs_async_flush.go b/weed/mount/weedfs_async_flush.go index 93031cb034..f75fef044b 100644 --- a/weed/mount/weedfs_async_flush.go +++ b/weed/mount/weedfs_async_flush.go @@ -7,6 +7,7 @@ import ( "github.com/seaweedfs/go-fuse/v2/fuse" "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/util" + "google.golang.org/grpc/codes" ) // asyncFlushItem holds the data needed for a background flush work item. @@ -110,13 +111,30 @@ func (wfs *WFS) completeAsyncFlush(fh *FileHandle) { fh.ReleaseHandle() } -// flushMetadataWithRetry attempts to flush file metadata to the filer, retrying -// with exponential backoff on transient errors. The chunk data is already on the -// volume servers at this point; only the filer metadata reference needs persisting. +// isFilerUnreachable reports a filer call that failed without the filer +// answering it: the dial, the connection or the call deadline gave out. It +// reads only the gRPC status, because a refusal the filer sends in its +// response has none, and its message carries the file path. +func isFilerUnreachable(err error) bool { + st, ok := util.ServerStatus(err) + return ok && st.Code() == codes.Unavailable +} + +// flushMetadataWithRetry persists the chunk references of a closed file. The +// chunk data is already on the volume servers and close() has returned, so +// nothing else holds these references: like the writeback chunk uploads, it +// keeps retrying for as long as the filer is unreachable. func (wfs *WFS) flushMetadataWithRetry(fh *FileHandle, dir, name string, fileFullPath util.FullPath) { - err := retryMetadataFlush(context.Background(), func() error { + attempts := 0 + err := retryMetadataFlushUntil(context.Background(), func() error { + attempts++ return wfs.flushMetadataToFiler(context.Background(), fh, dir, name, fh.asyncFlushUid, fh.asyncFlushGid) - }, func(nextAttempt, totalAttempts int, backoff time.Duration, err error) { + }, isFilerUnreachable, func(nextAttempt, totalAttempts int, backoff time.Duration, err error) { + if totalAttempts == 0 { + glog.Warningf("completeAsyncFlush %s: filer unreachable, retrying metadata flush (attempt %d) after %v: %v", + fileFullPath, nextAttempt, backoff, err) + return + } glog.Warningf("completeAsyncFlush %s: retrying metadata flush (attempt %d/%d) after %v: %v", fileFullPath, nextAttempt, totalAttempts, backoff, err) }) @@ -124,7 +142,7 @@ func (wfs *WFS) flushMetadataWithRetry(fh *FileHandle, dir, name string, fileFul glog.Errorf("completeAsyncFlush %s: metadata flush failed after %d attempts: %v - "+ "chunks are uploaded but NOT referenced in filer metadata; "+ "they will appear as orphans in volume.fsck", - fileFullPath, metadataFlushRetries+1, err) + fileFullPath, attempts, err) } } diff --git a/weed/mount/weedfs_async_flush_test.go b/weed/mount/weedfs_async_flush_test.go new file mode 100644 index 0000000000..c1578b2da9 --- /dev/null +++ b/weed/mount/weedfs_async_flush_test.go @@ -0,0 +1,103 @@ +package mount + +import ( + "context" + "errors" + "fmt" + "sync/atomic" + "syscall" + "testing" + "time" + + "github.com/seaweedfs/go-fuse/v2/fuse" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/util" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +type rejectingCreateFiler struct { + filer_pb.UnimplementedSeaweedFilerServer + rejections int32 + refuse func(req *filer_pb.CreateEntryRequest) (*filer_pb.CreateEntryResponse, error) + calls atomic.Int32 +} + +func (s *rejectingCreateFiler) CreateEntry(ctx context.Context, req *filer_pb.CreateEntryRequest) (*filer_pb.CreateEntryResponse, error) { + if s.calls.Add(1) <= s.rejections { + return s.refuse(req) + } + return &filer_pb.CreateEntryResponse{}, nil +} + +// A filer that answers with a refusal is not an outage, even when the file's +// own path reads like one: the deferred flush gives up after its bounded +// attempts instead of retrying until the refusal stops. +func TestFlushMetadataWithRetryStopsOnFilerRefusal(t *testing.T) { + originalSleep := metadataFlushSleep + t.Cleanup(func() { + metadataFlushSleep = originalSleep + }) + metadataFlushSleep = func(ctx context.Context, d time.Duration) {} + + const name = "unavailable-transport.log" + refusals := map[string]func(req *filer_pb.CreateEntryRequest) (*filer_pb.CreateEntryResponse, error){ + "grpc status": func(req *filer_pb.CreateEntryRequest) (*filer_pb.CreateEntryResponse, error) { + return nil, status.Error(codes.PermissionDenied, "denied") + }, + "response error": func(req *filer_pb.CreateEntryRequest) (*filer_pb.CreateEntryResponse, error) { + return &filer_pb.CreateEntryResponse{Error: "store write refused"}, nil + }, + } + for label, refuse := range refusals { + t.Run(label, func(t *testing.T) { + wfs, _ := newCreateTestWFS(t) + fake := &rejectingCreateFiler{rejections: 20, refuse: refuse} + startFakeFiler(t, wfs, fake) + + out := &fuse.CreateOut{} + if code := wfs.Create(make(chan struct{}), &fuse.CreateIn{ + InHeader: fuse.InHeader{NodeId: 1}, + Flags: syscall.O_WRONLY | syscall.O_CREAT, + Mode: 0o644, + }, name, out); code != fuse.OK { + t.Fatalf("Create status = %v, want OK", code) + } + fh := wfs.GetHandle(FileHandleId(out.Fh)) + if fh == nil { + t.Fatal("Create did not register an open file handle") + } + + wfs.flushMetadataWithRetry(fh, "/", name, util.FullPath("/").Child(name)) + + if got := fake.calls.Load(); got != metadataFlushRetries+1 { + t.Fatalf("CreateEntry calls = %d, want the bounded %d", got, metadataFlushRetries+1) + } + if !fh.dirtyMetadata { + t.Fatal("refused flush cleared dirtyMetadata") + } + }) + } +} + +func TestIsFilerUnreachable(t *testing.T) { + path := "/unavailable/transport.log" + cases := []struct { + name string + err error + want bool + }{ + {"dial refused", fmt.Errorf("fh flush create %s: %w", path, fmt.Errorf("CreateEntry: %w", + status.Error(codes.Unavailable, `connection error: desc = "transport: Error while dialing: dial tcp 127.0.0.1:18888: connect: connection refused"`))), true}, + {"call timed out", fmt.Errorf("fh flush create %s: %w", path, status.Error(codes.Unavailable, "filer 127.0.0.1:18888: call timed out")), true}, + {"read only", fmt.Errorf("fh flush create %s: %w", path, fmt.Errorf("CreateEntry %s: %w", path, filer_pb.ErrReadOnly)), false}, + {"message only", fmt.Errorf("fh flush create %s: %w", path, fmt.Errorf("CreateEntry: %w", errors.New("existing "+path+" is a directory"))), false}, + {"permission denied", fmt.Errorf("fh flush create %s: %w", path, status.Error(codes.PermissionDenied, "denied")), false}, + {"message too large", fmt.Errorf("fh flush create %s: %w", path, status.Error(codes.ResourceExhausted, "grpc: received message larger than max")), false}, + } + for _, c := range cases { + if got := isFilerUnreachable(c.err); got != c.want { + t.Errorf("%s: isFilerUnreachable(%v) = %v, want %v", c.name, c.err, got, c.want) + } + } +} diff --git a/weed/mount/weedfs_file_sync.go b/weed/mount/weedfs_file_sync.go index 8e58e2f23c..de47440abe 100644 --- a/weed/mount/weedfs_file_sync.go +++ b/weed/mount/weedfs_file_sync.go @@ -278,7 +278,7 @@ func (wfs *WFS) flushMetadataToFiler(ctx context.Context, fh *FileHandle, dir, n resp, err := wfs.streamCreateEntry(ctx, request) if err != nil { glog.Errorf("fh flush create %s: %v", fileFullPath, err) - return fmt.Errorf("fh flush create %s: %v", fileFullPath, err) + return fmt.Errorf("fh flush create %s: %w", fileFullPath, err) } event := resp.GetMetadataEvent()