diff --git a/weed/mount/filehandle_read.go b/weed/mount/filehandle_read.go index 88eb379c5..ad7e45a4e 100644 --- a/weed/mount/filehandle_read.go +++ b/weed/mount/filehandle_read.go @@ -190,7 +190,9 @@ func (fh *FileHandle) downloadRemoteEntry(entry *LockedEntry) error { glog.V(4).Infof("download entry: %v", request) // Barrier through this closure's client: on failover WithFilerClient // retries against another filer while the current-filer index still - // points at the failed one. + // points at the failed one. Only a fallback for old filers — the + // response's own log timestamp is causal with the returned entry, + // while a pre-RPC ping cannot cover events committed during the RPC. baselineTsNs := fh.wfs.filerBarrierTsNsWith(client) resp, err := client.CacheRemoteObjectToLocalCluster(context.Background(), request) if err != nil { @@ -198,6 +200,9 @@ func (fh *FileHandle) downloadRemoteEntry(entry *LockedEntry) error { } fh.SetEntry(resp.Entry) + if resp.GetLogTsNs() > baselineTsNs { + baselineTsNs = resp.GetLogTsNs() + } fh.noteFilerAck(baselineTsNs, resp.GetMetadataEvent()) // Async: a sync apply deadlocks against the apply loop's invalidate, which needs this read's file-handle lock. diff --git a/weed/mount/weedfs_filehandle.go b/weed/mount/weedfs_filehandle.go index d6a551f02..6b84d28c1 100644 --- a/weed/mount/weedfs_filehandle.go +++ b/weed/mount/weedfs_filehandle.go @@ -29,6 +29,21 @@ func (wfs *WFS) AcquireHandle(inode uint64, flags, uid, gid uint32) (fileHandle var path util.FullPath var existingFh *FileHandle path, existingFh, entry, status = wfs.maybeReadEntry(inode) + // Stable-read: an event applied while the lookup was in flight may + // already be reflected in the returned entry yet sit above the + // pre-lookup cursor, so its queued invalidation would replace fresher + // state. Re-read until the cursor is stable across the lookup. Bounded: + // on sustained churn the last pre-lookup value stands, which can only + // under-fence — a too-low fence lets an event re-apply, never blocks a + // newer one. + for attempt := 0; attempt < 3 && existingFh == nil && status == fuse.OK; attempt++ { + currentTsNs := wfs.latestKnownFilerTsNs() + if currentTsNs == baselineTsNs { + break + } + baselineTsNs = currentTsNs + 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 diff --git a/weed/mount/weedfs_invalidate_open_handle_test.go b/weed/mount/weedfs_invalidate_open_handle_test.go index cea0a7317..502f3030d 100644 --- a/weed/mount/weedfs_invalidate_open_handle_test.go +++ b/weed/mount/weedfs_invalidate_open_handle_test.go @@ -4,6 +4,7 @@ import ( "context" "net" "path/filepath" + "sync/atomic" "testing" "time" @@ -345,9 +346,13 @@ func TestQueuedEventOlderThanFlushedStateIsIgnored(t *testing.T) { type fakeFilerServer struct { filer_pb.UnimplementedSeaweedFilerServer - lookupSize uint64 - pingTsNs int64 - cacheSize uint64 + lookupSize uint64 + pingTsNs int64 + cacheSize uint64 + cacheLogTsNs int64 + lookupCalls atomic.Int32 + lookupStarted chan struct{} // closed when the first lookup arrives + lookupGate chan struct{} // first lookup waits here when non-nil } func (s *fakeFilerServer) CacheRemoteObjectToLocalCluster(ctx context.Context, req *filer_pb.CacheRemoteObjectToLocalClusterRequest) (*filer_pb.CacheRemoteObjectToLocalClusterResponse, error) { @@ -357,10 +362,15 @@ func (s *fakeFilerServer) CacheRemoteObjectToLocalCluster(ctx context.Context, r Name: req.Name, Attributes: &filer_pb.FuseAttributes{FileSize: s.cacheSize, FileMode: 0100644}, }, + LogTsNs: s.cacheLogTsNs, }, nil } func (s *fakeFilerServer) LookupDirectoryEntry(ctx context.Context, req *filer_pb.LookupDirectoryEntryRequest) (*filer_pb.LookupDirectoryEntryResponse, error) { + if s.lookupGate != nil && s.lookupCalls.Add(1) == 1 { + close(s.lookupStarted) + <-s.lookupGate + } return &filer_pb.LookupDirectoryEntryResponse{ Entry: &filer_pb.Entry{ Name: req.Name, @@ -783,3 +793,132 @@ func TestRemoteCacheBarrierFollowsFailover(t *testing.T) { t.Fatalf("open handle file size = %d, want 200 (barrier must come from the failover filer)", size) } } + +// An event applied while the open's lookup is in flight can already be +// reflected in the returned entry while sitting above the pre-lookup cursor. +// The open re-reads until the cursor is stable across the lookup so the fence +// covers such events. +func TestEventDuringOpenLookupIsFenced(t *testing.T) { + wfs := newInvalidateTestWFS(t) + fake := &fakeFilerServer{ + lookupSize: 200, + lookupStarted: make(chan struct{}), + lookupGate: make(chan struct{}), + } + startFakeFiler(t, wfs, fake) + + // Stall the invalidation worker on an unrelated handle's lock. + blockerInode := wfs.inodeToPath.Lookup(util.FullPath("/other/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) + blockerReleased := false + releaseBlocker := func() { + if !blockerReleased { + blockerReleased = true + wfs.fhLockTable.ReleaseLock(blockerFh.fh, blockerLock) + } + } + defer releaseBlocker() + blockerEvent := &filer_pb.SubscribeMetadataResponse{ + Directory: "/other", + 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: "/other", + }, + } + if err := wfs.metaCache.ApplyMetadataResponse(context.Background(), blockerEvent, meta_cache.SubscriberMetadataResponseApplyOptions); err != nil { + t.Fatalf("apply blocker event: %v", err) + } + + inode := wfs.inodeToPath.Lookup(util.FullPath("/dir/file"), time.Now().Unix(), false, false, 0, false) + type openResult struct { + fh *FileHandle + status fuse.Status + } + opened := make(chan openResult, 1) + go func() { + fh, status := wfs.AcquireHandle(inode, 0, 0, 0) + opened <- openResult{fh, status} + }() + + // While the open's lookup is blocked in the filer, an event lands and its + // invalidation is queued behind the blocker. + <-fake.lookupStarted + during := &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(), during, meta_cache.SubscriberMetadataResponseApplyOptions); err != nil { + t.Fatalf("apply mid-lookup event: %v", err) + } + close(fake.lookupGate) + + result := <-opened + if result.status != fuse.OK { + t.Fatalf("AcquireHandle status = %v, want OK", result.status) + } + + releaseBlocker() + wfs.metaCache.WaitForEntryInvalidations() + + if size := result.fh.GetEntry().GetEntry().Attributes.FileSize; size != 200 { + t.Fatalf("open handle file size = %d, want 200 (mid-lookup event must be fenced)", size) + } +} + +// A pre-RPC ping cannot cover an event committed during the RPC itself. The +// cache response carries a log timestamp stamped before the filer read the +// entry, causally fencing everything the returned entry reflects. +func TestCacheResponseLogTsFencesEventsCommittedDuringRPC(t *testing.T) { + wfs := newInvalidateTestWFS(t) + // The ping (100) predates the event (1500); only the response's log + // timestamp (2000) can cover it. + startFakeFiler(t, wfs, &fakeFilerServer{pingTsNs: 100, cacheSize: 200, cacheLogTsNs: 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}, + }) + + if err := fh.downloadRemoteEntry(fh.GetEntry()); err != nil { + t.Fatalf("downloadRemoteEntry: %v", err) + } + + 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 (response log ts must fence the mid-RPC event)", size) + } +} diff --git a/weed/pb/filer.proto b/weed/pb/filer.proto index d350b117a..480884d8b 100644 --- a/weed/pb/filer.proto +++ b/weed/pb/filer.proto @@ -763,6 +763,9 @@ message CacheRemoteObjectToLocalClusterRequest { message CacheRemoteObjectToLocalClusterResponse { Entry entry = 1; SubscribeMetadataResponse metadata_event = 2; + // filer log position stamped before the entry read: every event at or + // below it is reflected in the returned entry + int64 log_ts_ns = 3; } ///////////////////////// diff --git a/weed/pb/filer_pb/filer.pb.go b/weed/pb/filer_pb/filer.pb.go index 059dde3f3..4d8b339d7 100644 --- a/weed/pb/filer_pb/filer.pb.go +++ b/weed/pb/filer_pb/filer.pb.go @@ -5262,6 +5262,9 @@ type CacheRemoteObjectToLocalClusterResponse struct { state protoimpl.MessageState `protogen:"open.v1"` Entry *Entry `protobuf:"bytes,1,opt,name=entry,proto3" json:"entry,omitempty"` MetadataEvent *SubscribeMetadataResponse `protobuf:"bytes,2,opt,name=metadata_event,json=metadataEvent,proto3" json:"metadata_event,omitempty"` + // filer log position stamped before the entry read: every event at or + // below it is reflected in the returned entry + LogTsNs int64 `protobuf:"varint,3,opt,name=log_ts_ns,json=logTsNs,proto3" json:"log_ts_ns,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -5310,6 +5313,13 @@ func (x *CacheRemoteObjectToLocalClusterResponse) GetMetadataEvent() *SubscribeM return nil } +func (x *CacheRemoteObjectToLocalClusterResponse) GetLogTsNs() int64 { + if x != nil { + return x.LogTsNs + } + return 0 +} + // /////////////////////// // distributed lock management // /////////////////////// @@ -7333,10 +7343,11 @@ const file_filer_proto_rawDesc = "" + "\tdirectory\x18\x01 \x01(\tR\tdirectory\x12\x12\n" + "\x04name\x18\x02 \x01(\tR\x04name\x12+\n" + "\x11chunk_concurrency\x18\x03 \x01(\x05R\x10chunkConcurrency\x121\n" + - "\x14download_concurrency\x18\x04 \x01(\x05R\x13downloadConcurrency\"\x9c\x01\n" + + "\x14download_concurrency\x18\x04 \x01(\x05R\x13downloadConcurrency\"\xb8\x01\n" + "'CacheRemoteObjectToLocalClusterResponse\x12%\n" + "\x05entry\x18\x01 \x01(\v2\x0f.filer_pb.EntryR\x05entry\x12J\n" + - "\x0emetadata_event\x18\x02 \x01(\v2#.filer_pb.SubscribeMetadataResponseR\rmetadataEvent\"\x9b\x01\n" + + "\x0emetadata_event\x18\x02 \x01(\v2#.filer_pb.SubscribeMetadataResponseR\rmetadataEvent\x12\x1a\n" + + "\tlog_ts_ns\x18\x03 \x01(\x03R\alogTsNs\"\x9b\x01\n" + "\vLockRequest\x12\x12\n" + "\x04name\x18\x01 \x01(\tR\x04name\x12&\n" + "\x0fseconds_to_lock\x18\x02 \x01(\x03R\rsecondsToLock\x12\x1f\n" + diff --git a/weed/server/filer_grpc_server_remote.go b/weed/server/filer_grpc_server_remote.go index e7fd0b79d..d4ad767d3 100644 --- a/weed/server/filer_grpc_server_remote.go +++ b/weed/server/filer_grpc_server_remote.go @@ -64,6 +64,11 @@ func (fs *FilerServer) CacheRemoteObjectToLocalCluster(ctx context.Context, req // doCacheRemoteObjectToLocalCluster performs the actual caching operation. // This is called from singleflight, so only one instance runs per object. func (fs *FilerServer) doCacheRemoteObjectToLocalCluster(ctx context.Context, req *filer_pb.CacheRemoteObjectToLocalClusterRequest) (*filer_pb.CacheRemoteObjectToLocalClusterResponse, error) { + // Log position fence, stamped before the entry read: metadata events are + // logged after their store write and carry this clock, so every event at + // or below this timestamp is reflected in the entry returned below. + logTsNs := time.Now().UnixNano() + // find the entry first to check if already cached entry, err := fs.filer.FindEntry(ctx, util.JoinPath(req.Directory, req.Name)) if err == filer_pb.ErrNotFound { @@ -73,7 +78,7 @@ func (fs *FilerServer) doCacheRemoteObjectToLocalCluster(ctx context.Context, re return nil, fmt.Errorf("find entry %s/%s: %v", req.Directory, req.Name, err) } - resp := &filer_pb.CacheRemoteObjectToLocalClusterResponse{} + resp := &filer_pb.CacheRemoteObjectToLocalClusterResponse{LogTsNs: logTsNs} // Early return if not a remote-only object or already cached if entry.Remote == nil || entry.Remote.RemoteSize == 0 {