diff --git a/weed/mount/filehandle_read.go b/weed/mount/filehandle_read.go index b173f39ff..88eb379c5 100644 --- a/weed/mount/filehandle_read.go +++ b/weed/mount/filehandle_read.go @@ -188,7 +188,10 @@ func (fh *FileHandle) downloadRemoteEntry(entry *LockedEntry) error { } glog.V(4).Infof("download entry: %v", request) - baselineTsNs := fh.wfs.filerBarrierTsNs() + // Barrier through this closure's client: on failover WithFilerClient + // retries against another filer while the current-filer index still + // points at the failed one. + baselineTsNs := fh.wfs.filerBarrierTsNsWith(client) 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 84473f0be..8a116031b 100644 --- a/weed/mount/meta_cache/meta_cache.go +++ b/weed/mount/meta_cache/meta_cache.go @@ -585,6 +585,20 @@ func (mc *MetaCache) applyMetadataResponseNow(ctx context.Context, resp *filer_p return mc.applyMetadataResponseDirect(ctx, resp, options, false) } + for _, immediateEvent := range immediateEvents { + if err := mc.applyMetadataResponseDirect(ctx, immediateEvent, MetadataResponseApplyOptions{}, false); err != nil { + return err + } + } + // The cursor must cover a buffered event before its invalidation is + // observable, or an open racing the queue fences too low and the event + // later replaces newer state — a buffered event may never reach + // applyMetadataResponseDirect (build abort, snapshot-covered on + // completion). Sound without the store write: the building directory is + // read-through, so opens there consult the filer, which is at least this + // new. Immediate fragments were applied above, so their store writes are + // not outrun either. + mc.advanceLatestEventTs(resp.TsNs) // Apply side effects but skip directory notifications for dirs that are // currently being built. Notifying a building dir can trigger // markDirectoryReadThrough → DeleteFolderChildren, wiping entries that @@ -597,11 +611,6 @@ func (mc *MetaCache) applyMetadataResponseNow(ctx context.Context, resp *filer_p } state.bufferedEvents = append(state.bufferedEvents, events...) } - for _, immediateEvent := range immediateEvents { - if err := mc.applyMetadataResponseDirect(ctx, immediateEvent, MetadataResponseApplyOptions{}, false); err != nil { - return err - } - } return nil } diff --git a/weed/mount/weedfs.go b/weed/mount/weedfs.go index d53e90e0c..4059a639b 100644 --- a/weed/mount/weedfs.go +++ b/weed/mount/weedfs.go @@ -686,23 +686,34 @@ 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 +// filerBarrierTsNsWith returns a timestamp at or below the filer's current +// log position, read through the same client the fenced operation uses — a +// barrier from a different filer would not vouch for the state that filer +// serves. 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. +// attempt, falling back to the newest delivered event. +func (wfs *WFS) filerBarrierTsNsWith(client filer_pb.SeaweedFilerClient) int64 { + baseline := wfs.latestKnownFilerTsNs() + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + if resp, err := client.Ping(ctx, &filer_pb.PingRequest{}); err == nil && resp.StartTimeNs > baseline { + baseline = resp.StartTimeNs + } + return baseline +} + +// filerBarrierTsNs is filerBarrierTsNsWith against the current filer, for +// operations that target it outside a filer client callback (the server-side +// copy posts to the current filer over HTTP). 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 + baseline = wfs.filerBarrierTsNsWith(filer_pb.NewSeaweedFilerClient(conn)) + return nil }, wfs.getCurrentFiler().ToGrpcAddress(), false, wfs.option.GrpcDialOption) return baseline } diff --git a/weed/mount/weedfs_invalidate_open_handle_test.go b/weed/mount/weedfs_invalidate_open_handle_test.go index 04b0d82cf..cea0a7317 100644 --- a/weed/mount/weedfs_invalidate_open_handle_test.go +++ b/weed/mount/weedfs_invalidate_open_handle_test.go @@ -347,6 +347,17 @@ type fakeFilerServer struct { filer_pb.UnimplementedSeaweedFilerServer lookupSize uint64 pingTsNs int64 + cacheSize uint64 +} + +func (s *fakeFilerServer) CacheRemoteObjectToLocalCluster(ctx context.Context, req *filer_pb.CacheRemoteObjectToLocalClusterRequest) (*filer_pb.CacheRemoteObjectToLocalClusterResponse, error) { + // No MetadataEvent: the object was already cached by another client. + return &filer_pb.CacheRemoteObjectToLocalClusterResponse{ + Entry: &filer_pb.Entry{ + Name: req.Name, + Attributes: &filer_pb.FuseAttributes{FileSize: s.cacheSize, FileMode: 0100644}, + }, + }, nil } func (s *fakeFilerServer) LookupDirectoryEntry(ctx context.Context, req *filer_pb.LookupDirectoryEntryRequest) (*filer_pb.LookupDirectoryEntryResponse, error) { @@ -642,3 +653,133 @@ func TestFilerBarrierCoversUndeliveredEvents(t *testing.T) { t.Fatalf("open handle file size = %d, want 200 (undelivered-at-barrier event must not roll back)", size) } } + +// An event buffered for a building directory must be covered by the open-time +// cursor even if it never reaches a store write: aborting the build drops the +// buffered events while their invalidations stay queued, so an open fenced +// below the event would be rolled back. +func TestAbortedBuildEventStillCoveredByOpenFence(t *testing.T) { + wfs := newInvalidateTestWFS(t) + startFakeFiler(t, wfs, &fakeFilerServer{lookupSize: 200}) + + // 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) + 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 { + wfs.fhLockTable.ReleaseLock(blockerFh.fh, blockerLock) + t.Fatalf("apply blocker event: %v", err) + } + + if err := wfs.metaCache.BeginDirectoryBuild(context.Background(), util.FullPath("/dir")); err != nil { + wfs.fhLockTable.ReleaseLock(blockerFh.fh, blockerLock) + t.Fatalf("begin build: %v", err) + } + buffered := &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(), buffered, meta_cache.SubscriberMetadataResponseApplyOptions); err != nil { + wfs.fhLockTable.ReleaseLock(blockerFh.fh, blockerLock) + t.Fatalf("apply buffered event: %v", err) + } + if got := wfs.metaCache.LatestEventTsNs(); got != 1000 { + wfs.fhLockTable.ReleaseLock(blockerFh.fh, blockerLock) + t.Fatalf("LatestEventTsNs = %d, want 1000 (buffered event must advance the cursor)", got) + } + if err := wfs.metaCache.AbortDirectoryBuild(context.Background(), util.FullPath("/dir")); err != nil { + wfs.fhLockTable.ReleaseLock(blockerFh.fh, blockerLock) + t.Fatalf("abort build: %v", err) + } + + // Open after the abort: the lookup reaches the filer, which serves the + // newer size-200 state; the fence must cover the still-queued event. + 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) + } + + 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 (aborted-build event must not roll back the open)", size) + } +} + +// During filer failover, WithFilerClient retries the callback against another +// filer while the current-filer index still points at the failed one. The +// barrier must ping through the callback's client, or it silently degrades to +// the delivered-event cursor and reopens the undelivered-event rollback. +func TestRemoteCacheBarrierFollowsFailover(t *testing.T) { + wfs := newInvalidateTestWFS(t) + fake := &fakeFilerServer{pingTsNs: 2000, cacheSize: 200} + startFakeFiler(t, wfs, fake) + // First filer is unreachable; WithFilerClient fails over to the fake. + live := wfs.option.FilerAddresses[0] + wfs.option.FilerAddresses = []pb.ServerAddress{ + pb.NewServerAddressWithGrpcPort("127.0.0.1:1", 1), + live, + } + + 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) + } + if size := fh.GetEntry().GetEntry().Attributes.FileSize; size != 200 { + t.Fatalf("downloaded file size = %d, want 200", size) + } + + // An event committed before the download (TsNs 1500 < ping 2000) but + // delivered only now must not roll the handle back. + 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 (barrier must come from the failover filer)", size) + } +}