fix(mount): retain refused async metadata before releasing handle (#12598)

* fix(mount): retain refused async metadata before releasing handle

* mount: sync the failed-metadata directories with util.FsyncDir

Directory fsync fails on Windows, where the mount also builds, and returns
EINVAL on some filesystems. Either kept the retention loop retrying a record
that could never be saved, so the drain never finished.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* mount: keep failed-metadata records in a per-user directory

The write cache root defaults to the shared temp directory. The first mount
to save a record created failed-metadata there as 0700, so another user's
mount could never save one and its drain retried forever.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* mount: drop the async conflict probe test

TestContentAsyncConflictKeepsRecoveryRecord drives the same refused drain
with the retry backoff stubbed out; the probe slept through the real backoff
(7 s) and its kept-handle branch could not run. Its peer-entry check moves
into the recovery record test.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Chris Lu <chrislusf@users.noreply.github.com>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

weed/mount/content_flush_recovery.go

@@ -0,0 +1,124 @@
+package mount
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"os"
+	"path/filepath"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+	"google.golang.org/protobuf/proto"
+)
+
+// contentPublicationError keeps the exact failed request, before any later
+// subscription event can change the live handle. Unwrap preserves refusal and
+// transport classification.
+type contentPublicationError struct {
+	cause   error
+	request *filer_pb.CreateEntryRequest
+}
+
+func (e *contentPublicationError) Error() string { return e.cause.Error() }
+func (e *contentPublicationError) Unwrap() error { return e.cause }
+
+// retainFailedContent saves the refused request and its original CAS base for
+// operator recovery to a separate path. It never retries against a fresh base.
+// failed-metadata is outside the per-mount cache directories that shutdown and
+// restart remove. Records are private, immutable and never automatically replayed
+// or removed: chunk references may include cipher keys. They do not prevent a
+// volume.fsck run from collecting unreferenced chunks; recover before that sweep.
+func (wfs *WFS) retainFailedContent(fh *FileHandle, dir, name string, cause error) (string, error) {
+	root, err := wfs.failedContentDir()
+	if err != nil {
+		return "", err
+	}
+	if err = os.MkdirAll(root, 0700); err != nil {
+		return "", err
+	}
+	var request *filer_pb.CreateEntryRequest
+	var publication *contentPublicationError
+	if errors.As(cause, &publication) {
+		// Snapshot the request that failed, not a handle a later metadata event may
+		// already have changed after the publication lock was released.
+		request = proto.Clone(publication.request).(*filer_pb.CreateEntryRequest)
+	} else {
+		var entry *filer_pb.Entry
+		fh.UpdateEntry(func(e *filer_pb.Entry) { entry = proto.Clone(e).(*filer_pb.Entry) })
+		entry.Name = name
+		wfs.mapPbIdFromLocalToFiler(entry)
+		request = &filer_pb.CreateEntryRequest{Directory: dir, Entry: entry, Condition: wfs.failedContentCondition(fh), Signatures: []int32{wfs.signature}, SkipCheckParentDirectory: true}
+	}
+	record := struct {
+		Path            string
+		Error           string
+		ClientSignature int32
+		Request         *filer_pb.CreateEntryRequest
+	}{
+		Path: string(util.FullPath(request.Directory).Child(request.Entry.Name)), Error: cause.Error(), ClientSignature: wfs.signature, Request: request,
+	}
+	raw, err := json.Marshal(record)
+	if err != nil {
+		return "", err
+	}
+	file, err := os.CreateTemp(root, "pending-*.tmp")
+	if err != nil {
+		return "", err
+	}
+	temporary := file.Name()
+	path := temporary + ".json"
+	saved := false
+	defer func() {
+		file.Close()
+		if !saved {
+			os.Remove(temporary)
+			os.Remove(path)
+		}
+	}()
+	if _, err = file.Write(append(raw, '\n')); err != nil {
+		return "", err
+	}
+	if err = file.Sync(); err != nil {
+		return "", err
+	}
+	if err = file.Close(); err != nil {
+		return "", err
+	}
+	if err = os.Rename(temporary, path); err != nil {
+		return "", err
+	}
+	if err = util.FsyncDir(root); err != nil {
+		return "", err
+	}
+	// Persist creation of failed-metadata itself when this is its first record.
+	if err = util.FsyncDir(filepath.Dir(root)); err != nil {
+		return "", err
+	}
+	saved = true
+	return path, nil
+}
+
+// failedContentDir is per user: the write cache root defaults to the shared
+// temp directory, where another user's private directory would refuse every save.
+func (wfs *WFS) failedContentDir() (string, error) {
+	if wfs.option.CacheDirForWrite == "" {
+		return "", fmt.Errorf("write cache root unavailable for failed metadata retention")
+	}
+	name := "failed-metadata"
+	if uid := os.Getuid(); uid >= 0 {
+		name = fmt.Sprintf("%s-%d", name, uid)
+	}
+	return filepath.Abs(filepath.Join(wfs.option.CacheDirForWrite, name))
+}
+
+// Used only when failure happened before a publication request existed.
+func (wfs *WFS) failedContentCondition(fh *FileHandle) *filer_pb.WriteCondition {
+	clause := &filer_pb.WriteCondition_Clause{Kind: filer_pb.WriteCondition_IF_NOT_EXISTS}
+	if base := fh.baseEntry.Load(); base != nil {
+		expected := proto.Clone(base).(*filer_pb.Entry)
+		wfs.mapPbIdFromLocalToFiler(expected)
+		clause = &filer_pb.WriteCondition_Clause{Kind: filer_pb.WriteCondition_IF_ENTRY_EQUAL, ExpectedEntry: expected}
+	}
+	return &filer_pb.WriteCondition{Clauses: []*filer_pb.WriteCondition_Clause{clause}}
+}

weed/mount/content_flush_recovery_test.go

@@ -0,0 +1,240 @@
+//go:build !freebsd
+
+package mount
+
+import (
+	"context"
+	"encoding/json"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+	"google.golang.org/protobuf/proto"
+)
+
+// Match the actual CreateEntry wire contract: the server stamps CommitTsNs,
+// returns its committed entry in MetadataEvent, and refuses in-band.
+type committedContentFiler struct{ contentFlushConditionFiler }
+
+func (s *committedContentFiler) CreateEntry(ctx context.Context, r *filer_pb.CreateEntryRequest) (*filer_pb.CreateEntryResponse, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.calls++
+	for _, c := range r.GetCondition().GetClauses() {
+		match := c.Kind == filer_pb.WriteCondition_IF_NOT_EXISTS && s.entry == nil || c.Kind == filer_pb.WriteCondition_IF_ENTRY_EQUAL && proto.Equal(c.ExpectedEntry, s.entry)
+		if !match {
+			return &filer_pb.CreateEntryResponse{Error: "precondition failed", ErrorCode: filer_pb.FilerError_PRECONDITION_FAILED}, nil
+		}
+	}
+	s.entry = proto.Clone(r.Entry).(*filer_pb.Entry)
+	s.entry.CommitTsNs = int64(2000 + s.calls)
+	return &filer_pb.CreateEntryResponse{LogTsNs: s.entry.CommitTsNs, MetadataEvent: &filer_pb.SubscribeMetadataResponse{Directory: r.Directory, TsNs: s.entry.CommitTsNs, EventNotification: &filer_pb.EventNotification{NewEntry: proto.Clone(s.entry).(*filer_pb.Entry)}}}, nil
+}
+func committedContentHandle(t *testing.T, peer bool) (*WFS, *FileHandle, *committedContentFiler) {
+	t.Helper()
+	w := newInvalidateTestWFS(t)
+	base := flushConditionEntry(1)
+	base.CommitTsNs = 100
+	current := proto.Clone(base).(*filer_pb.Entry)
+	if peer {
+		current.Chunks = append(current.Chunks, flushConditionEntry(1, 2).Chunks[1])
+		current.CommitTsNs = 101
+	}
+	s := &committedContentFiler{contentFlushConditionFiler: contentFlushConditionFiler{entry: current}}
+	startFakeFiler(t, w, s)
+	w.mapPbIdFromFilerToLocal(base)
+	fh := newFileHandle(w, 1, 42, base)
+	fh.RememberPath(util.FullPath("/dir/file"))
+	fh.dirtyMetadata = true
+	t.Cleanup(fh.dirtyPages.Destroy)
+	return w, fh, s
+}
+func TestContentAsyncSuccessReleasesHandle(t *testing.T) {
+	w, fh, s := committedContentHandle(t, false)
+	w.option.CacheDirForWrite = t.TempDir()
+	w.inodeToPath.Lookup("/dir/file", 0, false, false, fh.inode, true)
+	w.fhMap.inode2fh[fh.inode] = fh
+	done := make(chan struct{})
+	w.pendingAsyncFlush = map[uint64]chan struct{}{fh.inode: done}
+	w.asyncFlushWg.Add(1)
+	w.processAsyncFlushItem(&asyncFlushItem{fh: fh, done: done})
+	select {
+	case <-done:
+	default:
+		t.Fatal("successful drain did not complete")
+	}
+	if _, kept := w.fhMap.FindFileHandle(fh.inode); kept {
+		t.Fatal("successful drain retained handle")
+	}
+	files, err := globFailedContent(t, w)
+	if err != nil || len(files) != 0 || s.calls != 1 {
+		t.Fatalf("successful drain: recovery=%v error=%v calls=%d", files, err, s.calls)
+	}
+}
+
+func TestContentAsyncConflictKeepsRecoveryRecord(t *testing.T) {
+	w, fh, s := committedContentHandle(t, true)
+	w.option.CacheDirForWrite = t.TempDir()
+	pending := flushConditionEntry(1, 3).Chunks[1]
+	pending.Offset = 2
+	fh.UpdateEntry(func(e *filer_pb.Entry) { e.Chunks = append(e.Chunks, pending) })
+	w.inodeToPath.Lookup("/dir/file", 0, false, false, fh.inode, true)
+	w.fhMap.inode2fh[fh.inode] = fh
+	fh.asyncFlushPending = true
+	done := make(chan struct{})
+	w.pendingAsyncFlush = map[uint64]chan struct{}{fh.inode: done}
+	w.asyncFlushWg.Add(1)
+	oldSleep := metadataFlushSleep
+	metadataFlushSleep = func(context.Context, time.Duration) {}
+	defer func() { metadataFlushSleep = oldSleep }()
+	peer := proto.Clone(s.entry).(*filer_pb.Entry)
+	w.processAsyncFlushItem(&asyncFlushItem{fh: fh, done: done})
+	select {
+	case <-done:
+	default:
+		t.Fatal("async completion not signalled")
+	}
+	if !proto.Equal(peer, s.entry) {
+		t.Fatal("refused async flush changed the peer entry")
+	}
+	files, err := globFailedContent(t, w)
+	if err != nil || len(files) != 1 {
+		t.Fatalf("failed close discarded pending refs: recovery files %v, error %v", files, err)
+	}
+	raw, err := os.ReadFile(files[0])
+	if err != nil {
+		t.Fatal(err)
+	}
+	var r struct {
+		Path    string
+		Error   string
+		Request *filer_pb.CreateEntryRequest
+	}
+	if err = json.Unmarshal(raw, &r); err != nil {
+		t.Fatal(err)
+	}
+	if r.Path != "/dir/file" || r.Error == "" || r.Request == nil {
+		t.Fatalf("missing recovery context: %s", raw)
+	}
+	expected := flushConditionEntry(1)
+	expected.CommitTsNs = 100
+	if !proto.Equal(r.Request.Condition.Clauses[0].ExpectedEntry, expected) {
+		t.Fatal("recovery lost observed CAS base")
+	}
+	if len(r.Request.Entry.Chunks) != 2 || !proto.Equal(r.Request.Entry.Chunks[1], pending) {
+		t.Fatal("recovery lost pending chunks")
+	}
+	info, err := os.Stat(files[0])
+	if err != nil || info.Mode().Perm() != 0600 {
+		t.Fatalf("private recovery record: %v %v", info, err)
+	}
+	if s.calls != metadataFlushRetries+1 {
+		t.Fatalf("publication attempts = %d, want existing retry bound %d", s.calls, metadataFlushRetries+1)
+	}
+}
+
+func TestContentRecoveryUsesFailedRequestNotLaterHandle(t *testing.T) {
+	w, fh, _ := committedContentHandle(t, true)
+	w.option.CacheDirForWrite = t.TempDir()
+	err := w.flushMetadataToFiler(context.Background(), fh, "/dir", "file", 1000, 20)
+	if err == nil {
+		t.Fatal("expected refusal")
+	}
+	fh.UpdateEntry(func(e *filer_pb.Entry) { e.Chunks = nil; e.CommitTsNs = 9000 })
+	path, saveErr := w.retainFailedContent(fh, "/dir", "file", err)
+	if saveErr != nil {
+		t.Fatal(saveErr)
+	}
+	raw, readErr := os.ReadFile(path)
+	if readErr != nil {
+		t.Fatal(readErr)
+	}
+	var r struct{ Request *filer_pb.CreateEntryRequest }
+	if readErr = json.Unmarshal(raw, &r); readErr != nil {
+		t.Fatal(readErr)
+	}
+	if len(r.Request.Entry.Chunks) != 1 || r.Request.Entry.CommitTsNs != 100 {
+		t.Fatal("recovery used a later handle, not the failed publication")
+	}
+}
+
+func TestContentAsyncRetentionWriteFailureKeepsDrainPending(t *testing.T) {
+	w, fh, s := committedContentHandle(t, true)
+	cache := filepath.Join(t.TempDir(), "cache")
+	if err := os.WriteFile(cache, []byte("blocked"), 0600); err != nil {
+		t.Fatal(err)
+	}
+	w.option.CacheDirForWrite = cache
+	w.inodeToPath.Lookup("/dir/file", 0, false, false, fh.inode, true)
+	w.fhMap.inode2fh[fh.inode] = fh
+	done := make(chan struct{})
+	w.pendingAsyncFlush = map[uint64]chan struct{}{fh.inode: done}
+	w.asyncFlushWg.Add(1)
+	oldSleep := metadataFlushSleep
+	defer func() { metadataFlushSleep = oldSleep }()
+	failures := 0
+	metadataFlushSleep = func(context.Context, time.Duration) {
+		if s.calls < metadataFlushRetries+1 {
+			return // Existing publication retry backoff, before retention.
+		}
+		failures++
+		if failures > 1 {
+			t.Fatal("retention did not recover after writable cache restored")
+		}
+		select {
+		case <-done:
+			t.Fatal("failed retention signalled completion")
+		default:
+		}
+		if _, ok := w.fhMap.FindFileHandle(fh.inode); !ok {
+			t.Fatal("failed retention removed pending handle")
+		}
+		if err := os.Remove(cache); err != nil {
+			t.Fatal(err)
+		}
+		if err := os.Mkdir(cache, 0700); err != nil {
+			t.Fatal(err)
+		}
+	}
+	w.processAsyncFlushItem(&asyncFlushItem{fh: fh, done: done})
+	files, err := globFailedContent(t, w)
+	if failures != 1 || err != nil || len(files) != 1 {
+		t.Fatalf("expected one retention failure then saved record: %d %v %v", failures, files, err)
+	}
+	if s.calls != metadataFlushRetries+1 {
+		t.Fatalf("retention retry added publication attempts: %d", s.calls)
+	}
+}
+
+func globFailedContent(t *testing.T, w *WFS) ([]string, error) {
+	t.Helper()
+	root, err := w.failedContentDir()
+	if err != nil {
+		return nil, err
+	}
+	return filepath.Glob(filepath.Join(root, "*.json"))
+}
+
+func TestContentRecoverySkipsAnotherUsersDirectory(t *testing.T) {
+	if os.Getuid() <= 0 {
+		t.Skip("needs a non-root uid to model another user's private directory")
+	}
+	w, fh, _ := committedContentHandle(t, true)
+	w.option.CacheDirForWrite = t.TempDir()
+	if err := os.Mkdir(filepath.Join(w.option.CacheDirForWrite, "failed-metadata"), 0500); err != nil {
+		t.Fatal(err)
+	}
+	err := w.flushMetadataToFiler(context.Background(), fh, "/dir", "file", 1000, 20)
+	if err == nil {
+		t.Fatal("expected refusal")
+	}
+	if _, saveErr := w.retainFailedContent(fh, "/dir", "file", err); saveErr != nil {
+		t.Fatalf("retention blocked by a directory this user cannot write: %v", saveErr)
+	}
+	if files, globErr := globFailedContent(t, w); globErr != nil || len(files) != 1 {
+		t.Fatalf("recovery records %v, error %v", files, globErr)
+	}
+}

weed/mount/weedfs_async_flush.go

@@ -101,7 +101,21 @@ func (wfs *WFS) completeAsyncFlush(fh *FileHandle) {
 				fileFullPath = resolvedPath
 			}
 
-			wfs.flushMetadataWithRetry(fh, dir, name, fileFullPath)
+			if err := wfs.flushMetadataWithRetry(fh, dir, name, fileFullPath); err != nil {
+				// close already returned. Do not discard the only pending
+				// references until a private recovery record is durable.
+				// Disk failures keep this drain pending; retry retention only,
+				// never replay the refused publication against a new base.
+				for {
+					path, saveErr := wfs.retainFailedContent(fh, dir, name, err)
+					if saveErr == nil {
+						glog.Errorf("completeAsyncFlush %s: publication failed: %v; recovery request retained at %s (no automatic replay)", fileFullPath, err, path)
+						break
+					}
+					glog.Errorf("completeAsyncFlush %s: retaining pending references in memory: %v; publication failure: %v", fileFullPath, saveErr, err)
+					metadataFlushSleep(context.Background(), util.RetryWaitTime)
+				}
+			}
 		}
 	}
 
@@ -124,7 +138,7 @@ func isFilerUnreachable(err error) bool {
 // 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) {
+func (wfs *WFS) flushMetadataWithRetry(fh *FileHandle, dir, name string, fileFullPath util.FullPath) error {
 	attempts := 0
 	err := retryMetadataFlushUntil(context.Background(), func() error {
 		attempts++
@@ -144,6 +158,7 @@ func (wfs *WFS) flushMetadataWithRetry(fh *FileHandle, dir, name string, fileFul
 			"they will appear as orphans in volume.fsck",
 			fileFullPath, attempts, err)
 	}
+	return err
 }
 
 // WaitForAsyncFlush waits for all pending background flush work items to complete.

weed/mount/weedfs_file_sync.go

@@ -306,7 +306,7 @@ func (wfs *WFS) flushMetadataToFiler(ctx context.Context, fh *FileHandle, dir, n
 	resp, merged, err := wfs.createContentEntry(ctx, fh, request, baseSnapshot, false)
 	if err != nil {
 		glog.Errorf("fh flush create %s: %v", fileFullPath, err)
-		return fmt.Errorf("fh flush create %s: %w", fileFullPath, err)
+		return &contentPublicationError{cause: fmt.Errorf("fh flush create %s: %w", fileFullPath, err), request: proto.Clone(request).(*filer_pb.CreateEntryRequest)}
 	}
 	if merged != nil {
 		fh.rebaseEntry(merged, requestEntry)