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 <noreply@anthropic.com>

* 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 <noreply@anthropic.com>

* 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 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

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

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)
+	}
+}

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

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

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)
+		}
+	}
+}

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