diff --git a/weed/mount/filehandle_read.go b/weed/mount/filehandle_read.go index 1d40d0ef3..b173f39ff 100644 --- a/weed/mount/filehandle_read.go +++ b/weed/mount/filehandle_read.go @@ -188,7 +188,7 @@ func (fh *FileHandle) downloadRemoteEntry(entry *LockedEntry) error { } glog.V(4).Infof("download entry: %v", request) - baselineTsNs := fh.wfs.latestKnownFilerTsNs() + baselineTsNs := fh.wfs.filerBarrierTsNs() resp, err := client.CacheRemoteObjectToLocalCluster(context.Background(), request) if err != nil { return fmt.Errorf("CacheRemoteObjectToLocalCluster file %s: %v", fileFullPath, err) diff --git a/weed/mount/meta_cache/meta_cache.go b/weed/mount/meta_cache/meta_cache.go index 16a91505a..84473f0be 100644 --- a/weed/mount/meta_cache/meta_cache.go +++ b/weed/mount/meta_cache/meta_cache.go @@ -576,7 +576,6 @@ type metadataResponseSideEffects struct { } func (mc *MetaCache) applyMetadataResponseNow(ctx context.Context, resp *filer_pb.SubscribeMetadataResponse, options MetadataResponseApplyOptions) error { - mc.advanceLatestEventTs(resp.TsNs) if mc.shouldSkipDuplicateEvent(resp) { return nil } @@ -610,6 +609,10 @@ func (mc *MetaCache) applyMetadataResponseDirect(ctx context.Context, resp *file if _, err := mc.applyMetadataResponseLocked(ctx, resp, options, allowUncachedInsert); err != nil { return err } + // Advance only after the store write: readers pairing a cursor capture + // with a store read (the open-time handle fence) rely on the cursor never + // leading the store. + mc.advanceLatestEventTs(resp.TsNs) mc.applyMetadataSideEffects(resp, options) return nil } diff --git a/weed/mount/weedfs.go b/weed/mount/weedfs.go index a05e845c2..d53e90e0c 100644 --- a/weed/mount/weedfs.go +++ b/weed/mount/weedfs.go @@ -686,6 +686,27 @@ func (wfs *WFS) latestKnownFilerTsNs() int64 { return wfs.metaCache.LatestEventTsNs() } +// filerBarrierTsNs returns a timestamp at or below the filer's current log +// position. Metadata events are stamped with the filer clock, and a self-ping +// reads that same clock, so unlike latestKnownFilerTsNs this also covers +// events already committed but not yet delivered to the subscription. Call +// before the RPC whose result the barrier fences. Best-effort: one bounded +// attempt, falling back to the newest delivered event — no WithFilerClient +// retry waves for an optimization. +func (wfs *WFS) filerBarrierTsNs() int64 { + baseline := wfs.latestKnownFilerTsNs() + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + _ = pb.WithGrpcClient(ctx, false, wfs.signature, func(conn *grpc.ClientConn) error { + resp, err := filer_pb.NewSeaweedFilerClient(conn).Ping(ctx, &filer_pb.PingRequest{}) + if err == nil && resp.StartTimeNs > baseline { + baseline = resp.StartTimeNs + } + return err + }, wfs.getCurrentFiler().ToGrpcAddress(), false, wfs.option.GrpcDialOption) + return baseline +} + // invalidateOpenFileHandle refreshes an open file handle from a metadata // subscription event. No filer lookup happens here: it can fail transiently, // and since the subscription cursor has already advanced past the event, the diff --git a/weed/mount/weedfs_file_copy_range.go b/weed/mount/weedfs_file_copy_range.go index 1656d0e4f..fb2b29103 100644 --- a/weed/mount/weedfs_file_copy_range.go +++ b/weed/mount/weedfs_file_copy_range.go @@ -216,7 +216,7 @@ func (wfs *WFS) tryServerSideWholeFileCopy(cancel <-chan struct{}, in *fuse.Copy glog.V(1).Infof("CopyFileRange server-side copy %s => %s (%d bytes)", copyRequest.srcPath, copyRequest.dstPath, copyRequest.sourceSize) - baselineTsNs := wfs.latestKnownFilerTsNs() + baselineTsNs := wfs.filerBarrierTsNs() entry, outcome, err := performServerSideWholeFileCopy(cancel, wfs, copyRequest) switch outcome { case serverSideWholeFileCopyCommitted: diff --git a/weed/mount/weedfs_filehandle.go b/weed/mount/weedfs_filehandle.go index 7fe61d1db..d6a551f02 100644 --- a/weed/mount/weedfs_filehandle.go +++ b/weed/mount/weedfs_filehandle.go @@ -17,9 +17,18 @@ func (wfs *WFS) AcquireHandle(inode uint64, flags, uid, gid uint32) (fileHandle // data that was just written asynchronously. wfs.waitForPendingAsyncFlush(inode) + // Fence baseline for a freshly looked-up entry: the lookup below reads + // the local store or the filer, both of which reflect every event applied + // so far, so invalidations already queued at or before this cursor are + // old news for the new handle. Captured before the lookup — never after — + // so an event arriving mid-open cannot inflate the fence past the state + // the lookup actually returned. + baselineTsNs := wfs.latestKnownFilerTsNs() + var entry *filer_pb.Entry var path util.FullPath - path, _, entry, status = wfs.maybeReadEntry(inode) + var existingFh *FileHandle + path, existingFh, entry, status = wfs.maybeReadEntry(inode) if status == fuse.OK { if wormEnforced, _ := wfs.wormEnforcedForEntry(path, entry); wormEnforced && flags&fuse.O_ANYWRITE != 0 { return nil, fuse.EPERM @@ -39,6 +48,11 @@ func (wfs *WFS) AcquireHandle(inode uint64, flags, uid, gid uint32) (fileHandle // need to AcquireFileHandle again to ensure correct handle counter fileHandle = wfs.fhMap.AcquireFileHandle(wfs, inode, entry) fileHandle.RememberPath(path) + // An existing handle's entry did not come from the lookup above, so + // the fence would overstate what it reflects. + if existingFh == nil { + fileHandle.advanceLocalEntryTs(baselineTsNs) + } // Acquire distributed lock for write opens. The lock is held with // auto-renewal until the file handle is released (close). diff --git a/weed/mount/weedfs_invalidate_open_handle_test.go b/weed/mount/weedfs_invalidate_open_handle_test.go index 30199f54c..04b0d82cf 100644 --- a/weed/mount/weedfs_invalidate_open_handle_test.go +++ b/weed/mount/weedfs_invalidate_open_handle_test.go @@ -343,11 +343,26 @@ func TestQueuedEventOlderThanFlushedStateIsIgnored(t *testing.T) { } } -type saveEntryTestServer struct { +type fakeFilerServer struct { filer_pb.UnimplementedSeaweedFilerServer + lookupSize uint64 + pingTsNs int64 } -func (s *saveEntryTestServer) UpdateEntry(ctx context.Context, req *filer_pb.UpdateEntryRequest) (*filer_pb.UpdateEntryResponse, error) { +func (s *fakeFilerServer) LookupDirectoryEntry(ctx context.Context, req *filer_pb.LookupDirectoryEntryRequest) (*filer_pb.LookupDirectoryEntryResponse, error) { + return &filer_pb.LookupDirectoryEntryResponse{ + Entry: &filer_pb.Entry{ + Name: req.Name, + Attributes: &filer_pb.FuseAttributes{FileSize: s.lookupSize, FileMode: 0100644}, + }, + }, nil +} + +func (s *fakeFilerServer) Ping(ctx context.Context, req *filer_pb.PingRequest) (*filer_pb.PingResponse, error) { + return &filer_pb.PingResponse{StartTimeNs: s.pingTsNs}, nil +} + +func (s *fakeFilerServer) UpdateEntry(ctx context.Context, req *filer_pb.UpdateEntryRequest) (*filer_pb.UpdateEntryResponse, error) { return &filer_pb.UpdateEntryResponse{ MetadataEvent: &filer_pb.SubscribeMetadataResponse{ Directory: req.Directory, @@ -371,7 +386,7 @@ func TestSaveEntryKeepsOpenHandleAheadOfOlderEvents(t *testing.T) { } t.Cleanup(func() { _ = listener.Close() }) server := pb.NewGrpcServer() - filer_pb.RegisterSeaweedFilerServer(server, &saveEntryTestServer{}) + filer_pb.RegisterSeaweedFilerServer(server, &fakeFilerServer{}) go server.Serve(listener) t.Cleanup(server.Stop) @@ -490,3 +505,140 @@ func TestNilAckEventFallsBackToLatestSeenTs(t *testing.T) { t.Fatalf("open handle file size = %d, want 200 (queued event at TsNs 1000 predates the known log position 1500)", size) } } + +// startFakeFiler serves fake on a local port and points wfs at it. +func startFakeFiler(t *testing.T, wfs *WFS, fake *fakeFilerServer) { + t.Helper() + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + t.Cleanup(func() { _ = listener.Close() }) + server := pb.NewGrpcServer() + filer_pb.RegisterSeaweedFilerServer(server, fake) + go server.Serve(listener) + t.Cleanup(server.Stop) + wfs.option.FilerAddresses = []pb.ServerAddress{ + pb.NewServerAddressWithGrpcPort("127.0.0.1:1", listener.Addr().(*net.TCPAddr).Port), + } +} + +// A handle opened while an older event sits in the invalidation queue is +// fenced at open time: its entry came from a lookup that reflects every +// event applied so far, so the queued event must not replace it. +func TestQueuedEventDoesNotRollBackHandleOpenedAfterEnqueue(t *testing.T) { + wfs := newInvalidateTestWFS(t) + startFakeFiler(t, wfs, &fakeFilerServer{lookupSize: 200}) + + // Stall the single invalidation worker on an unrelated handle's lock so + // queued events outlive the open below. + blockerInode := wfs.inodeToPath.Lookup(util.FullPath("/dir/blocker"), time.Now().Unix(), false, false, 0, false) + blockerFh := wfs.fhMap.AcquireFileHandle(wfs, blockerInode, &filer_pb.Entry{ + Name: "blocker", + Attributes: &filer_pb.FuseAttributes{FileSize: 1}, + }) + blockerLock := wfs.fhLockTable.AcquireLock("test", blockerFh.fh, util.ExclusiveLock) + blockerEvent := &filer_pb.SubscribeMetadataResponse{ + Directory: "/dir", + TsNs: 500, + EventNotification: &filer_pb.EventNotification{ + OldEntry: &filer_pb.Entry{Name: "blocker"}, + NewEntry: &filer_pb.Entry{ + Name: "blocker", + Attributes: &filer_pb.FuseAttributes{FileSize: 2}, + }, + NewParentPath: "/dir", + }, + } + if err := wfs.metaCache.ApplyMetadataResponse(context.Background(), blockerEvent, meta_cache.SubscriberMetadataResponseApplyOptions); err != nil { + wfs.fhLockTable.ReleaseLock(blockerFh.fh, blockerLock) + t.Fatalf("apply blocker event: %v", err) + } + + // The event for the file predates the open below. + older := &filer_pb.SubscribeMetadataResponse{ + Directory: "/dir", + TsNs: 1000, + EventNotification: &filer_pb.EventNotification{ + OldEntry: &filer_pb.Entry{Name: "file"}, + NewEntry: &filer_pb.Entry{ + Name: "file", + Attributes: &filer_pb.FuseAttributes{FileSize: 100}, + }, + NewParentPath: "/dir", + }, + } + if err := wfs.metaCache.ApplyMetadataResponse(context.Background(), older, meta_cache.SubscriberMetadataResponseApplyOptions); err != nil { + wfs.fhLockTable.ReleaseLock(blockerFh.fh, blockerLock) + t.Fatalf("apply subscriber event: %v", err) + } + + // Open the file now: the lookup reaches the filer, which already serves + // the newer size-200 state. + inode := wfs.inodeToPath.Lookup(util.FullPath("/dir/file"), time.Now().Unix(), false, false, 0, false) + fh, status := wfs.AcquireHandle(inode, 0, 0, 0) + if status != fuse.OK { + wfs.fhLockTable.ReleaseLock(blockerFh.fh, blockerLock) + t.Fatalf("AcquireHandle status = %v, want OK", status) + } + if size := fh.GetEntry().GetEntry().Attributes.FileSize; size != 200 { + wfs.fhLockTable.ReleaseLock(blockerFh.fh, blockerLock) + t.Fatalf("opened handle file size = %d, want 200", size) + } + + wfs.fhLockTable.ReleaseLock(blockerFh.fh, blockerLock) + wfs.metaCache.WaitForEntryInvalidations() + + if size := fh.GetEntry().GetEntry().Attributes.FileSize; size != 200 { + t.Fatalf("open handle file size = %d, want 200 (event queued before the open must not roll it back)", size) + } +} + +// The delivered-event cursor misses events already committed on the filer but +// not yet delivered to the subscription. A pre-RPC filer self-ping reads the +// clock those events are stamped with, so state fetched after the ping fences +// them out. +func TestFilerBarrierCoversUndeliveredEvents(t *testing.T) { + wfs := newInvalidateTestWFS(t) + startFakeFiler(t, wfs, &fakeFilerServer{pingTsNs: 2000}) + + inode := wfs.inodeToPath.Lookup(util.FullPath("/dir/file"), time.Now().Unix(), false, false, 0, false) + fh := wfs.fhMap.AcquireFileHandle(wfs, inode, &filer_pb.Entry{ + Name: "file", + Attributes: &filer_pb.FuseAttributes{FileSize: 88}, + }) + + // The copy/remote-cache pattern: barrier, then the operation's result. + // The event at TsNs 1500 is committed but not yet delivered, so only the + // filer clock (2000) can cover it. + baselineTsNs := wfs.filerBarrierTsNs() + if baselineTsNs != 2000 { + t.Fatalf("filerBarrierTsNs = %d, want 2000 from the filer ping", baselineTsNs) + } + fh.SetEntry(&filer_pb.Entry{ + Name: "file", + Attributes: &filer_pb.FuseAttributes{FileSize: 200}, + }) + fh.advanceLocalEntryTs(baselineTsNs) + + late := &filer_pb.SubscribeMetadataResponse{ + Directory: "/dir", + TsNs: 1500, + EventNotification: &filer_pb.EventNotification{ + OldEntry: &filer_pb.Entry{Name: "file"}, + NewEntry: &filer_pb.Entry{ + Name: "file", + Attributes: &filer_pb.FuseAttributes{FileSize: 100}, + }, + NewParentPath: "/dir", + }, + } + if err := wfs.metaCache.ApplyMetadataResponse(context.Background(), late, meta_cache.SubscriberMetadataResponseApplyOptions); err != nil { + t.Fatalf("apply late event: %v", err) + } + wfs.metaCache.WaitForEntryInvalidations() + + if size := fh.GetEntry().GetEntry().Attributes.FileSize; size != 200 { + t.Fatalf("open handle file size = %d, want 200 (undelivered-at-barrier event must not roll back)", size) + } +}