From d7ae08ecad36c17c05c74c297b8d519c3f4b09b2 Mon Sep 17 00:00:00 2001 From: Ping Qiu Date: Thu, 12 Nov 2026 16:15:08 -0800 Subject: [PATCH] fix(mount): refuse stale whole-entry content flushes (#12594) * fix(mount): condition content flushes on the observed entry * mount: stop retrying a flush the filer refused on its condition The filer reports a failed CreateEntry condition as FilerError_PRECONDITION_FAILED in the response, which surfaces as filer_pb.ErrPreconditionFailed rather than a FailedPrecondition status. The flush retry loop treated it as transient and replayed the refused flush three more times over 7 s before returning EIO. The test filer now answers like the real one. Co-Authored-By: Claude Opus 5.5 * mount: take the handle base from the entry the filer stored After an acknowledged flush the handle base was the requested entry, but the filer stores it with a fresh commit_ts_ns (and fills crtime, inode, atime), so the next IF_ENTRY_EQUAL flush on the same handle never matched: write, fsync, write, fsync returned EIO, as did close after a periodic metadata flush and the first flush after an eager create. Use the stored entry from the acknowledgement's metadata event. Co-Authored-By: Claude Opus 5.5 * mount: condition the periodic metadata flush like the close flush The periodic metadata flush of an open, dirty handle sent the same whole entry through an unconditional CreateEntry, so a file held open past the flush interval still overwrote a peer's acknowledged append, and the close flush then matched the overwritten state. Both paths now publish through one helper that carries the handle's write condition. Co-Authored-By: Claude Opus 5.5 * mount: republish a refused flush once when the stored entry is unchanged A handle's base can differ from the stored entry without any peer write: a rename while the file is open restamps the entry under its new name, and with -map.uid a filer id that is also a mapped local id does not map back to itself. Each made every flush of the handle fail with EIO. On a miss, read the stored entry and, if it still holds the observed state apart from name, commit stamp, id encoding and derived size, retry once with that entry as the condition. Co-Authored-By: Claude Opus 5.5 * mount: take the filer's entry once a write open holds the DLM lock The lock orders this writer after the previous holder, but the holder's flush may not have reached the local cache yet, so the handle's base named the entry from before the wait and the conditional flush refused. Co-Authored-By: Claude Opus 5.5 * mount: share the open-existing path of Create Co-Authored-By: Claude Opus 5.5 * mount: open the file a previous DLM holder created during a deferred create A deferred create never asks the filer, so a create that waited for the lock went on to publish over a file the previous holder had just made, and its flush expecting no entry was refused. Co-Authored-By: Claude Opus 5.5 * master: send the lock ring snapshot to mounts A mount running -dlm is a lock client like the S3 gateway, and needs the ring on connect to reach a file's owner filer before the next membership change is broadcast. Co-Authored-By: Claude Opus 5.5 * mount: follow the masters' lock ring under -dlm The lock client only knew the seed filer, so every lock took the forward hop and the mount had no way to tell which filer owns a file. Subscribe to the masters named by the filer, like the S3 gateway, and wait briefly for the first ring before serving. Co-Authored-By: Claude Opus 5.5 * mount: read the DLM-ordered entry from the file's owner filer The filer checks a flush's condition on the owner of the path's write route key, and another filer may not have applied the previous holder's flush yet. Look the entry up on that owner once the ring is known. Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Chris Lu Co-authored-by: Claude Opus 5.5 --- weed/cluster/cluster.go | 1 + weed/mount/content_flush_condition_test.go | 248 +++++++++++++++++++++ weed/mount/error_classifier.go | 7 + weed/mount/error_classifier_test.go | 3 + weed/mount/weedfs.go | 4 + weed/mount/weedfs_file_mkrm.go | 91 ++++---- weed/mount/weedfs_file_sync.go | 80 ++++++- weed/mount/weedfs_filehandle.go | 36 +++ weed/mount/weedfs_lock_ring.go | 75 +++++++ weed/mount/weedfs_lock_ring_test.go | 44 ++++ weed/mount/weedfs_metadata_flush.go | 4 +- weed/server/master_grpc_server.go | 8 +- weed/server/master_grpc_server_test.go | 15 ++ 13 files changed, 561 insertions(+), 55 deletions(-) create mode 100644 weed/mount/content_flush_condition_test.go create mode 100644 weed/mount/weedfs_lock_ring.go create mode 100644 weed/mount/weedfs_lock_ring_test.go diff --git a/weed/cluster/cluster.go b/weed/cluster/cluster.go index 7436474dce..2c47d3a2c4 100644 --- a/weed/cluster/cluster.go +++ b/weed/cluster/cluster.go @@ -13,6 +13,7 @@ const ( VolumeServerType = "volumeServer" FilerType = "filer" BrokerType = "broker" + MountType = "mount" S3Type = "s3" VectorType = "vector" ) diff --git a/weed/mount/content_flush_condition_test.go b/weed/mount/content_flush_condition_test.go new file mode 100644 index 0000000000..4eb719a253 --- /dev/null +++ b/weed/mount/content_flush_condition_test.go @@ -0,0 +1,248 @@ +//go:build !freebsd + +package mount + +import ( + "context" + "sync" + "testing" + + "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" + "google.golang.org/protobuf/proto" +) + +type contentFlushConditionFiler struct { + filer_pb.UnimplementedSeaweedFilerServer + mu sync.Mutex + entry *filer_pb.Entry + calls int +} + +func (s *contentFlushConditionFiler) 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() { + matches := false + switch c.Kind { + case filer_pb.WriteCondition_IF_ENTRY_EQUAL: + matches = proto.Equal(c.ExpectedEntry, s.entry) + case filer_pb.WriteCondition_IF_NOT_EXISTS: + matches = s.entry == nil + } + if !matches { + return &filer_pb.CreateEntryResponse{Error: "precondition failed", ErrorCode: filer_pb.FilerError_PRECONDITION_FAILED}, nil + } + } + // Like the filer, stamp the stored version and announce what was stored. + old := s.entry + s.entry = proto.Clone(r.Entry).(*filer_pb.Entry) + s.entry.CommitTsNs = 1000 + int64(s.calls) + return &filer_pb.CreateEntryResponse{LogTsNs: 2000, MetadataEvent: &filer_pb.SubscribeMetadataResponse{ + Directory: r.Directory, + TsNs: s.entry.CommitTsNs, + EventNotification: &filer_pb.EventNotification{ + OldEntry: old, + NewEntry: proto.Clone(s.entry).(*filer_pb.Entry), + }, + }}, nil +} +func (s *contentFlushConditionFiler) LookupDirectoryEntry(ctx context.Context, r *filer_pb.LookupDirectoryEntryRequest) (*filer_pb.LookupDirectoryEntryResponse, error) { + s.mu.Lock() + defer s.mu.Unlock() + if s.entry == nil { + return nil, status.Error(codes.NotFound, filer_pb.ErrNotFound.Error()) + } + return &filer_pb.LookupDirectoryEntryResponse{Entry: proto.Clone(s.entry).(*filer_pb.Entry)}, nil +} + +func flushConditionEntry(ids ...uint64) *filer_pb.Entry { + e := &filer_pb.Entry{Name: "file", Attributes: &filer_pb.FuseAttributes{Inode: 42, FileMode: 0644, Uid: 2000, Gid: 20, FileSize: 3}} + for i, id := range ids { + e.Chunks = append(e.Chunks, &filer_pb.FileChunk{Fid: &filer_pb.FileId{VolumeId: 1, FileKey: id, Cookie: 1}, Offset: int64(i), Size: 1, ModifiedTsNs: int64(id)}) + } + return e +} + +// Exercise the actual mount content-publication caller. A peer's successful +// append or replacement is authoritative; refusing our stale flush must retain +// our dirty entry, instead of refreshing a guard and replaying stale content. +func TestContentFlushPreservesPeerChunks(t *testing.T) { + for _, name := range []string{"single-writer", "peer-append", "replaced", "deleted", "deferred-create", "deferred-create-race"} { + t.Run(name, func(t *testing.T) { + wfs := newInvalidateTestWFS(t) + observed := flushConditionEntry(1) + live := flushConditionEntry(1, 3) + live.Chunks[1].Offset = 2 + current := flushConditionEntry(1) + wantError := false + switch name { + case "peer-append": + current = flushConditionEntry(1, 2) + wantError = true + case "replaced": + current = flushConditionEntry(4) + current.Attributes.Inode = 99 + wantError = true + case "deleted": + current = nil + wantError = true + case "deferred-create": + observed = nil + current = nil + case "deferred-create-race": + observed = nil + current = flushConditionEntry(4) + current.Attributes.Inode = 99 + wantError = true + } + if observed != nil { + wfs.mapPbIdFromFilerToLocal(observed) + } + wfs.mapPbIdFromFilerToLocal(live) + fake := &contentFlushConditionFiler{entry: current} + startFakeFiler(t, wfs, fake) + fh := newFileHandle(wfs, 1, 42, live) + t.Cleanup(func() { fh.dirtyPages.Destroy() }) + fh.baseEntry.Store(observed) + fh.dirtyMetadata = true + fh.RememberPath(util.FullPath("/dir/file")) + before := proto.Clone(current) + code := wfs.doFlush(context.Background(), fh, 1000, 20, false) + fake.mu.Lock() + defer fake.mu.Unlock() + if wantError { + if code == fuse.OK { + t.Fatal("stale content flush acknowledged after a peer changed the entry") + } + if !proto.Equal(fake.entry, before) { + t.Fatalf("peer entry changed: %v", fake.entry) + } + if !fh.dirtyMetadata { + t.Fatal("refused flush lost dirty metadata") + } + if len(fh.entry.Entry.Chunks) != 2 || fh.entry.Entry.Chunks[1].GetFileIdString() != live.Chunks[1].GetFileIdString() { + t.Fatal("refused flush lost pending content") + } + } else { + if code != fuse.OK { + t.Fatal(code) + } + if fh.dirtyMetadata { + t.Fatal("successful flush remains dirty") + } + if len(fake.entry.Chunks) != 2 || fake.entry.Chunks[1].Offset != 2 || fake.entry.Chunks[1].Size != 1 { + t.Fatalf("normal flush lost content: %v", fake.entry) + } + } + if fake.calls != 1 { + t.Fatalf("unexpected replay: %d calls", fake.calls) + } + }) + } +} + +// Each acknowledged flush must leave the handle a base the next flush on the +// same handle can match: the stored entry, with the filer's own commit stamp. +func TestContentFlushRepeatsOnOneHandle(t *testing.T) { + wfs := newInvalidateTestWFS(t) + stored := flushConditionEntry(1) + stored.CommitTsNs = 7 + fake := &contentFlushConditionFiler{entry: proto.Clone(stored).(*filer_pb.Entry)} + startFakeFiler(t, wfs, fake) + observed := proto.Clone(stored).(*filer_pb.Entry) + wfs.mapPbIdFromFilerToLocal(observed) + fh := newFileHandle(wfs, 1, 42, proto.Clone(observed).(*filer_pb.Entry)) + t.Cleanup(func() { fh.dirtyPages.Destroy() }) + fh.RememberPath(util.FullPath("/dir/file")) + for i := uint64(2); i <= 3; i++ { + fh.UpdateEntry(func(e *filer_pb.Entry) { + e.Chunks = append(e.Chunks, &filer_pb.FileChunk{Fid: &filer_pb.FileId{VolumeId: 1, FileKey: i, Cookie: 1}, Offset: int64(i - 1), Size: 1, ModifiedTsNs: int64(i)}) + e.Attributes.FileSize = i + }) + fh.dirtyMetadata = true + if code := wfs.doFlush(context.Background(), fh, 1000, 20, false); code != fuse.OK { + t.Fatalf("flush %d: %v", i-1, code) + } + } + fake.mu.Lock() + defer fake.mu.Unlock() + if len(fake.entry.Chunks) != 3 || fake.calls != 2 { + t.Fatalf("stored %d chunks after %d calls", len(fake.entry.Chunks), fake.calls) + } +} + +// The periodic metadata flush publishes the same whole entry, so it must not +// overwrite a peer's append either. +func TestPeriodicMetadataFlushPreservesPeerChunks(t *testing.T) { + wfs := newInvalidateTestWFS(t) + current := flushConditionEntry(1, 2) + fake := &contentFlushConditionFiler{entry: proto.Clone(current).(*filer_pb.Entry)} + startFakeFiler(t, wfs, fake) + observed := flushConditionEntry(1) + wfs.mapPbIdFromFilerToLocal(observed) + live := flushConditionEntry(1, 3) + live.Chunks[1].Offset = 2 + wfs.mapPbIdFromFilerToLocal(live) + fh := newFileHandle(wfs, 1, 42, live) + t.Cleanup(func() { fh.dirtyPages.Destroy() }) + fh.baseEntry.Store(observed) + fh.dirtyMetadata = true + fh.RememberPath(util.FullPath("/dir/file")) + if err := wfs.flushFileMetadata(fh); err == nil { + t.Fatal("periodic flush overwrote a peer's append") + } + fake.mu.Lock() + defer fake.mu.Unlock() + if !proto.Equal(fake.entry, current) || fake.calls != 1 { + t.Fatalf("peer entry changed after %d calls: %v", fake.calls, fake.entry) + } + if !fh.dirtyMetadata { + t.Fatal("refused periodic flush lost dirty metadata") + } +} + +// A miss on state no peer changed is republished once against the stored +// entry: a rename while open, or a filer uid the local id mapping cannot +// reproduce (newInvalidateTestWFS maps local 1000 to filer 2000). +func TestContentFlushRepublishesUnchangedEntry(t *testing.T) { + for _, name := range []string{"renamed", "colliding-uid"} { + t.Run(name, func(t *testing.T) { + wfs := newInvalidateTestWFS(t) + stored := flushConditionEntry(1) + stored.CommitTsNs = 7 + observed := proto.Clone(stored).(*filer_pb.Entry) + path := util.FullPath("/dir/file") + switch name { + case "renamed": + wfs.mapPbIdFromFilerToLocal(observed) + stored.Name, stored.CommitTsNs = "moved", 8 + path = "/dir/moved" + case "colliding-uid": + stored.Attributes.Uid, observed.Attributes.Uid = 1000, 1000 + } + fake := &contentFlushConditionFiler{entry: proto.Clone(stored).(*filer_pb.Entry)} + startFakeFiler(t, wfs, fake) + live := proto.Clone(observed).(*filer_pb.Entry) + live.Name = stored.Name + live.Chunks = append(live.Chunks, &filer_pb.FileChunk{Fid: &filer_pb.FileId{VolumeId: 1, FileKey: 3, Cookie: 1}, Offset: 1, Size: 1, ModifiedTsNs: 3}) + fh := newFileHandle(wfs, 1, 42, live) + t.Cleanup(func() { fh.dirtyPages.Destroy() }) + fh.baseEntry.Store(observed) + fh.dirtyMetadata = true + fh.RememberPath(path) + if code := wfs.doFlush(context.Background(), fh, 1000, 20, false); code != fuse.OK { + t.Fatal(code) + } + fake.mu.Lock() + defer fake.mu.Unlock() + if len(fake.entry.Chunks) != 2 || fake.calls != 2 { + t.Fatalf("stored %d chunks after %d calls", len(fake.entry.Chunks), fake.calls) + } + }) + } +} diff --git a/weed/mount/error_classifier.go b/weed/mount/error_classifier.go index 22e664d321..be2bb24552 100644 --- a/weed/mount/error_classifier.go +++ b/weed/mount/error_classifier.go @@ -1,12 +1,15 @@ package mount import ( + "errors" "strings" "syscall" "github.com/seaweedfs/go-fuse/v2/fuse" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" ) func grpcErrorToFuseStatus(err error) fuse.Status { @@ -81,6 +84,10 @@ func isRetryableFilerError(err error) bool { if err == nil { return false } + // CreateEntry reports a failed write condition in its response, not as a status. + if errors.Is(err, filer_pb.ErrPreconditionFailed) { + return false + } if s, ok := status.FromError(err); ok { switch s.Code() { case codes.NotFound, diff --git a/weed/mount/error_classifier_test.go b/weed/mount/error_classifier_test.go index 04cb44cabb..7099265ed0 100644 --- a/weed/mount/error_classifier_test.go +++ b/weed/mount/error_classifier_test.go @@ -11,6 +11,8 @@ import ( "github.com/seaweedfs/go-fuse/v2/fuse" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" ) // Regression: wfs_save.go used to wrap the gRPC error with fmt.Errorf(... %v ...) @@ -97,6 +99,7 @@ func TestIsRetryableFilerError(t *testing.T) { {"permission_denied", status.Error(codes.PermissionDenied, "no access"), false}, {"unauthenticated", status.Error(codes.Unauthenticated, "missing creds"), false}, {"failed_precondition", status.Error(codes.FailedPrecondition, "not empty"), false}, + {"create_condition_failed", fmt.Errorf("fh flush create /f: CreateEntry /f: %w", filer_pb.ErrPreconditionFailed), false}, {"plain_error_retries", errors.New("random network glitch"), true}, {"wrapped_canceled_still_retries", fmt.Errorf("ctx: %w", status.Error(codes.Canceled, "closing")), true}, {"wrapped_not_found_still_skipped", fmt.Errorf("ctx: %w", status.Error(codes.NotFound, "gone")), false}, diff --git a/weed/mount/weedfs.go b/weed/mount/weedfs.go index 2bfe531f86..47564574c8 100644 --- a/weed/mount/weedfs.go +++ b/weed/mount/weedfs.go @@ -469,6 +469,10 @@ func (wfs *WFS) StartBackgroundTasks() error { glog.V(0).Infof("writebackCache enabled: async flush on close() for improved small file performance") } + if wfs.lockClient != nil { + wfs.followLockRing() + } + follower, err := wfs.subscribeFilerConfEvents() if err != nil { return err diff --git a/weed/mount/weedfs_file_mkrm.go b/weed/mount/weedfs_file_mkrm.go index 213a97eea8..78ad8d819d 100644 --- a/weed/mount/weedfs_file_mkrm.go +++ b/weed/mount/weedfs_file_mkrm.go @@ -45,31 +45,7 @@ func (wfs *WFS) Create(cancel <-chan struct{}, in *fuse.CreateIn, name string, o defer cancelFunc() newEntry, _, code := wfs.maybeLoadEntry(ctx, entryFullPath) if code == fuse.OK { - if newEntry == nil || newEntry.Attributes == nil { - return fuse.EIO - } - if in.Flags&syscall.O_EXCL != 0 { - glog.V(0).Infof("Create O_EXCL %s: already exists (uid=%d gid=%d mode=%o)", - entryFullPath, newEntry.Attributes.Uid, newEntry.Attributes.Gid, newEntry.Attributes.FileMode) - return fuse.Status(syscall.EEXIST) - } - inode = wfs.inodeToPath.Lookup(entryFullPath, newEntry.Attributes.Crtime, false, len(newEntry.HardLinkId) > 0, newEntry.Attributes.Inode, true) - fileHandle, openFh, status := wfs.AcquireHandle(ctx, inode, in.Flags, in.Uid, in.Gid) - if status != fuse.OK { - return status - } - if in.Flags&syscall.O_TRUNC != 0 && in.Flags&fuse.O_ANYWRITE != 0 { - if code = wfs.truncateEntry(entryFullPath, newEntry); code != fuse.OK { - wfs.ReleaseHandle(openFh) - return code - } - newEntry = fileHandle.GetEntry().GetEntry() - } - - wfs.outputPbEntry(&out.EntryOut, inode, newEntry) - out.Fh = uint64(openFh) - out.OpenFlags = 0 - return fuse.OK + return wfs.openExistingForCreate(ctx, in, entryFullPath, newEntry, out) } if code != fuse.ENOENT { return code @@ -86,6 +62,14 @@ func (wfs *WFS) Create(cancel <-chan struct{}, in *fuse.CreateIn, name string, o if dlmLock == nil { return fuse.Status(syscall.EAGAIN) } + // A deferred create does not ask the filer, so find a file the + // previous lock holder created while this create waited. + if !wfs.option.EagerFilerCreate { + if peerEntry, _, err := wfs.lookupFilerEntry(ctx, entryFullPath); err == nil { + dlmLock.Stop() + return wfs.openExistingForCreate(ctx, in, entryFullPath, peerEntry, out) + } + } } inode, newEntry, code = wfs.createRegularFile(ctx, dirFullPath, name, in.Mode, in.Uid, in.Gid, 0, !wfs.option.EagerFilerCreate, true) @@ -99,25 +83,7 @@ func (wfs *WFS) Create(cancel <-chan struct{}, in *fuse.CreateIn, name string, o if code != fuse.OK { return code } - if newEntry == nil || newEntry.Attributes == nil { - return fuse.EIO - } - inode = wfs.inodeToPath.Lookup(entryFullPath, newEntry.Attributes.Crtime, false, len(newEntry.HardLinkId) > 0, newEntry.Attributes.Inode, true) - fileHandle, openFh, status := wfs.AcquireHandle(ctx, inode, in.Flags, in.Uid, in.Gid) - if status != fuse.OK { - return status - } - if in.Flags&syscall.O_TRUNC != 0 && in.Flags&fuse.O_ANYWRITE != 0 { - if code = wfs.truncateEntry(entryFullPath, newEntry); code != fuse.OK { - wfs.ReleaseHandle(openFh) - return code - } - newEntry = fileHandle.GetEntry().GetEntry() - } - wfs.outputPbEntry(&out.EntryOut, inode, newEntry) - out.Fh = uint64(openFh) - out.OpenFlags = 0 - return fuse.OK + return wfs.openExistingForCreate(ctx, in, entryFullPath, newEntry, out) } else if code != fuse.OK { if dlmLock != nil { dlmLock.Stop() @@ -156,6 +122,12 @@ func (wfs *WFS) Create(cancel <-chan struct{}, in *fuse.CreateIn, name string, o // file is closed without any writes. An eager create has already // persisted the entry, so its handle starts clean. fileHandle.dirtyMetadata = !wfs.option.EagerFilerCreate + if !wfs.option.EagerFilerCreate { + // This is a local placeholder, not an acknowledged filer base. + fileHandle.baseEntry.Store(nil) + } else { + fileHandle.setAuthoritativeBase(proto.Clone(newEntry).(*filer_pb.Entry)) + } // Create bypasses AcquireHandle, so attach the lock acquired above. // A surviving handle may already hold one; ours is redundant then. @@ -174,6 +146,35 @@ func (wfs *WFS) Create(cancel <-chan struct{}, in *fuse.CreateIn, name string, o return fuse.OK } +// openExistingForCreate answers a create whose name already exists: an +// exclusive create fails, any other opens the entry, truncating it if asked. +func (wfs *WFS) openExistingForCreate(ctx context.Context, in *fuse.CreateIn, entryFullPath util.FullPath, entry *filer_pb.Entry, out *fuse.CreateOut) fuse.Status { + if entry == nil || entry.Attributes == nil { + return fuse.EIO + } + if in.Flags&syscall.O_EXCL != 0 { + glog.V(0).Infof("Create O_EXCL %s: already exists (uid=%d gid=%d mode=%o)", + entryFullPath, entry.Attributes.Uid, entry.Attributes.Gid, entry.Attributes.FileMode) + return fuse.Status(syscall.EEXIST) + } + inode := wfs.inodeToPath.Lookup(entryFullPath, entry.Attributes.Crtime, false, len(entry.HardLinkId) > 0, entry.Attributes.Inode, true) + fileHandle, openFh, status := wfs.AcquireHandle(ctx, inode, in.Flags, in.Uid, in.Gid) + if status != fuse.OK { + return status + } + if in.Flags&syscall.O_TRUNC != 0 && in.Flags&fuse.O_ANYWRITE != 0 { + if code := wfs.truncateEntry(entryFullPath, entry); code != fuse.OK { + wfs.ReleaseHandle(openFh) + return code + } + entry = fileHandle.GetEntry().GetEntry() + } + wfs.outputPbEntry(&out.EntryOut, inode, entry) + out.Fh = uint64(openFh) + out.OpenFlags = 0 + return fuse.OK +} + /** Create a file node * * This is called for creation of all non-directory, non-symlink @@ -473,7 +474,7 @@ func (wfs *WFS) createRegularFile(ctx context.Context, dirFullPath util.FullPath return 0, nil, grpcErrorToFuseStatus(err) } - return inode, newEntry, fuse.OK + return inode, wfs.ackedBase(resp, newEntry), fuse.OK } // markHandleDeleted flags the inode's open handle so its flushes stop writing diff --git a/weed/mount/weedfs_file_sync.go b/weed/mount/weedfs_file_sync.go index de47440abe..a7d5c8eea5 100644 --- a/weed/mount/weedfs_file_sync.go +++ b/weed/mount/weedfs_file_sync.go @@ -3,6 +3,7 @@ package mount import ( "bytes" "context" + "errors" "fmt" "io" "syscall" @@ -180,9 +181,9 @@ func (wfs *WFS) doFlush(ctx context.Context, fh *FileHandle, uid, gid uint32, al return fuse.Status(syscall.ENOSPC) } - if err := retryMetadataFlush(ctx, func() error { + if err := retryMetadataFlushIf(ctx, func() error { return wfs.flushMetadataToFiler(ctx, fh, dir, name, uid, gid) - }, func(nextAttempt, totalAttempts int, backoff time.Duration, err error) { + }, isRetryableFilerError, func(nextAttempt, totalAttempts int, backoff time.Duration, err error) { glog.Warningf("%v fh %d flush: retrying metadata flush (attempt %d/%d) after %v: %v", fileFullPath, fh.fh, nextAttempt, totalAttempts, backoff, err) }); err != nil { @@ -275,7 +276,7 @@ func (wfs *WFS) flushMetadataToFiler(ctx context.Context, fh *FileHandle, dir, n baseSnapshot := proto.Clone(requestEntry).(*filer_pb.Entry) wfs.mapPbIdFromLocalToFiler(request.Entry) - resp, err := wfs.streamCreateEntry(ctx, request) + resp, err := wfs.createContentEntry(ctx, fh, request) if err != nil { glog.Errorf("fh flush create %s: %v", fileFullPath, err) return fmt.Errorf("fh flush create %s: %w", fileFullPath, err) @@ -291,7 +292,7 @@ func (wfs *WFS) flushMetadataToFiler(ctx context.Context, fh *FileHandle, dir, n // The filer acknowledged this state at the event's log position (or, for // a no-op create, at the response's log position); older queued // subscription events must not roll the handle back. - fh.setAuthoritativeBase(baseSnapshot) + fh.setAuthoritativeBase(wfs.ackedBase(resp, baseSnapshot)) fh.advanceEntryVersion(ackVersionTsNs(resp), resp.GetLogSignature()) if applyErr := wfs.applyLocalMetadataEvent(context.Background(), event); applyErr != nil { glog.Warningf("flush %s: best-effort metadata apply failed: %v", fileFullPath, applyErr) @@ -305,6 +306,77 @@ func (wfs *WFS) flushMetadataToFiler(ctx context.Context, fh *FileHandle, dir, n return err } +// createContentEntry publishes a handle's whole entry only over the filer +// state the handle observed. Never refresh the guard and replay a stale whole +// entry after a conflict: that would discard a peer's acknowledged chunks. +// Deferred creates expect no entry; existing handles compare the full base, +// including inode identity. The unary call keeps a refused condition apart +// from an I/O failure; the mutation stream reduces it to EIO. +// +// A miss is retried once, against the stored entry, only when that entry is +// still the observed state: renamed, restamped, or encoded with filer ids the +// local mapping cannot reproduce. +func (wfs *WFS) createContentEntry(ctx context.Context, fh *FileHandle, request *filer_pb.CreateEntryRequest) (resp *filer_pb.CreateEntryResponse, err error) { + base := fh.baseEntry.Load() + guard := &filer_pb.WriteCondition_Clause{Kind: filer_pb.WriteCondition_IF_NOT_EXISTS} + if base != nil { + expected := proto.Clone(base).(*filer_pb.Entry) + wfs.mapPbIdFromLocalToFiler(expected) + guard = &filer_pb.WriteCondition_Clause{Kind: filer_pb.WriteCondition_IF_ENTRY_EQUAL, ExpectedEntry: expected} + } + for attempt := 0; attempt < 2; attempt++ { + request.Condition = &filer_pb.WriteCondition{Clauses: []*filer_pb.WriteCondition_Clause{guard}} + var stored *filer_pb.Entry + err = wfs.withFilerClient(ctx, false, wfs.filerRPCWait(), func(ctx context.Context, client filer_pb.SeaweedFilerClient) error { + var callErr error + resp, callErr = filer_pb.CreateEntryWithResponse(ctx, client, request) + if base == nil || attempt > 0 || !errors.Is(callErr, filer_pb.ErrPreconditionFailed) { + return callErr + } + if found, lookupErr := filer_pb.LookupEntry(ctx, client, &filer_pb.LookupDirectoryEntryRequest{Directory: request.Directory, Name: request.Entry.Name}); lookupErr == nil { + stored = found.Entry + } + return callErr + }) + if stored == nil || !wfs.sameObservedEntry(stored, base) { + return resp, err + } + guard = &filer_pb.WriteCondition_Clause{Kind: filer_pb.WriteCondition_IF_ENTRY_EQUAL, ExpectedEntry: stored} + } + return resp, err +} + +// sameObservedEntry reports whether the filer's stored entry still holds the +// state the handle observed, ignoring what no peer write is carried by: the +// name, the commit stamp, the id encoding and the derived file size. +func (wfs *WFS) sameObservedEntry(stored, base *filer_pb.Entry) bool { + current := proto.Clone(stored).(*filer_pb.Entry) + wfs.mapPbIdFromFilerToLocal(current) + observed := proto.Clone(base).(*filer_pb.Entry) + for _, e := range []*filer_pb.Entry{current, observed} { + e.Name, e.CommitTsNs = "", 0 + if e.Attributes != nil { + e.Attributes.FileSize = filer.FileSize(e) + } + filer_pb.BeforeEntrySerialization(e.Chunks) + } + return proto.Equal(current, observed) +} + +// ackedBase is the base an acknowledged create leaves: the entry the filer +// stored, in local ids. The filer stamps commit_ts_ns, crtime, inode and atime +// and may rewrite chunks, so the requested entry would never match it again. +// A filer that sends no event leaves the requested entry. +func (wfs *WFS) ackedBase(resp *filer_pb.CreateEntryResponse, requested *filer_pb.Entry) *filer_pb.Entry { + stored := resp.GetMetadataEvent().GetEventNotification().GetNewEntry() + if stored == nil { + return requested + } + base := proto.Clone(stored).(*filer_pb.Entry) + wfs.mapPbIdFromFilerToLocal(base) + return base +} + // shouldMergeChunks reports whether the non-manifest chunks are bloated // enough to justify re-reading and re-uploading the file. The condition // is: sum of compacted chunk sizes > 2 * logical file size. diff --git a/weed/mount/weedfs_filehandle.go b/weed/mount/weedfs_filehandle.go index e401d20756..16220d26c4 100644 --- a/weed/mount/weedfs_filehandle.go +++ b/weed/mount/weedfs_filehandle.go @@ -105,10 +105,46 @@ func (wfs *WFS) AcquireHandle(ctx context.Context, inode uint64, flags, uid, gid return nil, 0, fuse.Status(syscall.EAGAIN) } glog.V(1).Infof("DLM lock acquired for %s", path) + // The previous holder's flush may not have reached the local cache, + // and the lock orders this writer after it: take the filer's entry. + if entry, version, err := wfs.lookupFilerEntry(ctx, path); err == nil { + fileHandle.installAckedEntry(entry, version.tsNs, version.signature) + } } return fileHandle, openFh, fuse.OK } +// lookupFilerEntry reads path from the filer, bypassing the local cache, in +// local ids and with the log position the answer reflects. It asks the filer +// that checks path's write conditions, since another filer may lag it. +func (wfs *WFS) lookupFilerEntry(ctx context.Context, path util.FullPath) (*filer_pb.Entry, entryVersion, error) { + dir, name := path.DirAndName() + var entry *filer_pb.Entry + var version entryVersion + lookup := func(ctx context.Context, client filer_pb.SeaweedFilerClient) error { + resp, err := filer_pb.LookupEntry(ctx, client, &filer_pb.LookupDirectoryEntryRequest{Directory: dir, Name: name}) + if err != nil { + return err + } + entry, version = resp.Entry, entryVersion{tsNs: resp.LogTsNs, signature: resp.LogSignature} + return nil + } + var err error + if owner := wfs.entryWriteOwner(path); owner != "" { + err = wfs.callFiler(ctx, false, wfs.filerRPCWait(), owner.ToGrpcAddress(), lookup) + } else { + err = wfs.withFilerClient(ctx, false, wfs.filerRPCWait(), lookup) + } + if err != nil { + return nil, entryVersion{}, err + } + if entry.Attributes == nil { + entry.Attributes = &filer_pb.FuseAttributes{} + } + wfs.mapPbIdFromFilerToLocal(entry) + return entry, version, nil +} + // ReleaseHandle is called from FUSE Release. For handles with a pending // async flush, the map removal and the pendingAsyncFlush registration are // done under a single lock hold so that a concurrent AcquireHandle cannot diff --git a/weed/mount/weedfs_lock_ring.go b/weed/mount/weedfs_lock_ring.go new file mode 100644 index 0000000000..13c08b3489 --- /dev/null +++ b/weed/mount/weedfs_lock_ring.go @@ -0,0 +1,75 @@ +package mount + +import ( + "context" + "strings" + "time" + + "github.com/seaweedfs/seaweedfs/weed/cluster" + "github.com/seaweedfs/seaweedfs/weed/glog" + "github.com/seaweedfs/seaweedfs/weed/pb" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" + "github.com/seaweedfs/seaweedfs/weed/util" + "github.com/seaweedfs/seaweedfs/weed/wdclient" +) + +// lockRingWait bounds how long startup waits for the first lock ring; until +// one arrives, locks start at the seed filer and lookups use any filer. +const lockRingWait = 10 * time.Second + +// followLockRing feeds the masters' lock ring into lockClient, so locks and +// owner reads go straight to the filer that owns a file. +func (wfs *WFS) followLockRing() { + var masters []string + var filerGroup string + err := wfs.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + resp, err := client.GetFilerConfiguration(context.Background(), &filer_pb.GetFilerConfigurationRequest{}) + if err != nil { + return err + } + masters, filerGroup = resp.Masters, resp.FilerGroup + return nil + }) + if err != nil || len(masters) == 0 { + glog.Warningf("dlm: cannot follow the lock ring, masters %v: %v", masters, err) + return + } + + masterClient := wdclient.NewMasterClient(wfs.option.GrpcDialOption, filerGroup, cluster.MountType, "", wfs.option.DataCenter, "", + *pb.ServerAddresses(strings.Join(masters, ",")).ToServiceDiscovery()) + masterClient.SetOnLockRingUpdateFn(func(update *master_pb.LockRingUpdate) { + servers := make([]pb.ServerAddress, 0, len(update.Servers)) + for _, s := range update.Servers { + servers = append(servers, pb.ServerAddress(s)) + } + wfs.lockClient.SetRing(servers, update.Version) + }) + masterClient.SetOnMasterChangeFn(func(previous, current pb.ServerAddress) { + wfs.lockClient.ResetRing() + }) + go masterClient.KeepConnectedToMaster(context.Background()) + + root := wfs.option.FilerMountRootPath + for deadline := time.Now().Add(lockRingWait); wfs.lockClient.PrimaryForKey(root) == ""; time.Sleep(100 * time.Millisecond) { + if time.Now().After(deadline) { + glog.Warningf("dlm: no lock ring from masters %v after %v", masters, lockRingWait) + return + } + } +} + +// entryWriteOwner is the filer that serializes path's writes per this mount's +// ring view, matching the filer's routing: the prior owner while a rebalance +// cools off, else the primary. "" before a ring arrives. +func (wfs *WFS) entryWriteOwner(path util.FullPath) pb.ServerAddress { + if wfs.lockClient == nil { + return "" + } + key := s3_constants.ObjectWriteRouteKeyPrefix + string(path) + if prior := wfs.lockClient.PriorOwnerForKey(key); prior != "" { + return prior + } + return wfs.lockClient.PrimaryForKey(key) +} diff --git a/weed/mount/weedfs_lock_ring_test.go b/weed/mount/weedfs_lock_ring_test.go new file mode 100644 index 0000000000..1e8e993f2d --- /dev/null +++ b/weed/mount/weedfs_lock_ring_test.go @@ -0,0 +1,44 @@ +package mount + +import ( + "fmt" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/cluster" + "github.com/seaweedfs/seaweedfs/weed/cluster/lock_manager" + "github.com/seaweedfs/seaweedfs/weed/pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" + "github.com/seaweedfs/seaweedfs/weed/util" +) + +func TestEntryWriteOwnerMatchesFilerRouting(t *testing.T) { + wfs := &WFS{} + path := util.FullPath("/dir/file") + if got := wfs.entryWriteOwner(path); got != "" { + t.Fatalf("no lock client: owner %q, want none", got) + } + + wfs.lockClient = cluster.NewLockClient(nil, "seed:8888") + if got := wfs.entryWriteOwner(path); got != "" { + t.Fatalf("no ring yet: owner %q, want none", got) + } + + servers := []pb.ServerAddress{"filer1:8888", "filer2:8888", "filer3:8888"} + wfs.lockClient.SetRing(servers, 1) + filerRing := lock_manager.NewLockRing(time.Hour) + filerRing.SetSnapshot(servers, 1) + + owners := map[pb.ServerAddress]bool{} + for i := 0; i < 64; i++ { + p := util.FullPath(fmt.Sprintf("/dir/file%d", i)) + want := filerRing.WriteOwner(s3_constants.ObjectWriteRouteKeyPrefix + string(p)) + if got := wfs.entryWriteOwner(p); got != want { + t.Fatalf("%s: mount routes to %q, filer to %q", p, got, want) + } + owners[want] = true + } + if len(owners) < 2 { + t.Fatalf("all paths hashed to %v; the test does not tell routing keys apart", owners) + } +} diff --git a/weed/mount/weedfs_metadata_flush.go b/weed/mount/weedfs_metadata_flush.go index bce2e9e8d9..fdcafa0ef5 100644 --- a/weed/mount/weedfs_metadata_flush.go +++ b/weed/mount/weedfs_metadata_flush.go @@ -179,7 +179,7 @@ func (wfs *WFS) flushFileMetadata(fh *FileHandle) error { baseSnapshot := proto.Clone(requestEntry).(*filer_pb.Entry) wfs.mapPbIdFromLocalToFiler(request.Entry) - resp, err := wfs.streamCreateEntry(context.Background(), request) + resp, err := wfs.createContentEntry(context.Background(), fh, request) if err != nil { return err } @@ -191,7 +191,7 @@ func (wfs *WFS) flushFileMetadata(fh *FileHandle) error { event.TsNs = ackVersionTsNs(resp) } } - fh.setAuthoritativeBase(baseSnapshot) + fh.setAuthoritativeBase(wfs.ackedBase(resp, baseSnapshot)) fh.advanceEntryVersion(ackVersionTsNs(resp), resp.GetLogSignature()) if applyErr := wfs.applyLocalMetadataEvent(context.Background(), event); applyErr != nil { glog.Warningf("flushFileMetadata %s: best-effort metadata apply failed: %v", fileFullPath, applyErr) diff --git a/weed/server/master_grpc_server.go b/weed/server/master_grpc_server.go index 7d7b97818c..e046eba54c 100644 --- a/weed/server/master_grpc_server.go +++ b/weed/server/master_grpc_server.go @@ -667,10 +667,10 @@ func (ms *MasterServer) initialLockRingUpdate(clientType string, filerGroup stri if ms.LockRingManager == nil { return nil } - // Filers are ring members; S3 gateways are lock clients that need the same - // view to dial a key's primary directly. Both get the initial snapshot; - // later membership changes already broadcast to every connected client. - if clientType != cluster.FilerType && clientType != cluster.S3Type { + // Filers are ring members; S3 gateways and mounts are lock clients that + // need the same view to dial a key's primary directly. All get the initial + // snapshot; later membership changes already broadcast to every client. + if clientType != cluster.FilerType && clientType != cluster.S3Type && clientType != cluster.MountType { return nil } diff --git a/weed/server/master_grpc_server_test.go b/weed/server/master_grpc_server_test.go index 787d1f1913..1a83f07136 100644 --- a/weed/server/master_grpc_server_test.go +++ b/weed/server/master_grpc_server_test.go @@ -148,6 +148,21 @@ func TestInitialLockRingUpdateSkipsNonFilers(t *testing.T) { assert.Nil(t, ms.initialLockRingUpdate(cluster.BrokerType, "group-a")) } +func TestInitialLockRingUpdateReachesLockClients(t *testing.T) { + ms := &MasterServer{ + LockRingManager: cluster.NewLockRingManager(nil), + } + + ms.LockRingManager.AddServer("group-a", "filer1:8888") + ms.LockRingManager.FlushPending("group-a") + + for _, clientType := range []string{cluster.S3Type, cluster.MountType} { + resp := ms.initialLockRingUpdate(clientType, "group-a") + require.NotNil(t, resp, clientType) + assert.Equal(t, []string{"filer1:8888"}, resp.LockRingUpdate.Servers, clientType) + } +} + // TestReconnectedClientSurvivesOldHandlerCleanup covers a client reconnecting // KeepConnected under the same name before the old handler has exited: the old // handler's deferred cleanup must not close the channel the reconnected stream