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()