diff --git a/weed/mount/content_flush_recovery.go b/weed/mount/content_flush_recovery.go new file mode 100644 index 0000000000..40a40b3eae --- /dev/null +++ b/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}} +} diff --git a/weed/mount/content_flush_recovery_test.go b/weed/mount/content_flush_recovery_test.go new file mode 100644 index 0000000000..c593dd8bdb --- /dev/null +++ b/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) + } +} diff --git a/weed/mount/weedfs_async_flush.go b/weed/mount/weedfs_async_flush.go index f75fef044b..383b4abeba 100644 --- a/weed/mount/weedfs_async_flush.go +++ b/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. diff --git a/weed/mount/weedfs_file_sync.go b/weed/mount/weedfs_file_sync.go index a3c2bc96cc..ef905d2095 100644 --- a/weed/mount/weedfs_file_sync.go +++ b/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)