mount: cover buffered build events in the cursor, barrier via the failover client

A buffered build event only reached the cursor through its store write, which never happens on build abort and is skipped for snapshot-covered events at completion — while its invalidation stays queued. An open in that window fenced below the event, so the queued invalidation replaced the newer looked-up state. Advance the cursor when the event is buffered, before its invalidation is observable. This is sound without the store write: the building directory is read-through, so opens there consult the filer, which is at least as new; immediate rename fragments are applied first so their store writes are not outrun. The remote-cache barrier pinged the current-filer index from inside the WithFilerClient callback. During failover the callback retries against another filer while the index still points at the failed one, so the ping failed and the barrier silently degraded to the delivered-event cursor, reopening the undelivered-event rollback. Ping through the callback's own client instead; the server-side copy keeps the current-filer barrier since it posts to that filer over HTTP.
2026-09-20 13:30:46 +02:00 · 2026-07-22 16:11:00 -07:00
parent 239cfe64d3
commit 6a5f1b66b4
4 changed files with 179 additions and 15 deletions
@@ -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)
@@ -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
 }

@@ -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
 }
@@ -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)
+	}
+}