diff --git a/weed/mount/meta_cache/meta_cache.go b/weed/mount/meta_cache/meta_cache.go index 7e234a786..1b33d8417 100644 --- a/weed/mount/meta_cache/meta_cache.go +++ b/weed/mount/meta_cache/meta_cache.go @@ -497,13 +497,19 @@ func (mc *MetaCache) entryVersionRecordLocked(ctx context.Context, fp util.FullP return int64(util.BytesToUint64(value[:8])), tombstone, unversioned } -// entryVersionBlocksLocked reports whether a write at tsNs is already +// entryVersionBlocksLocked reports whether a change at tsNs is already // reflected at fp. The path's version is its own record or, lacking one, the // listing floors — which cover names a listing saw present and absent alike. // A tombstone fences with no entry present, and a later floor outranks it; a // plain record only counts while its entry exists: records linger after a // bulk folder wipe and must not fence a recreate. -func (mc *MetaCache) entryVersionBlocksLocked(ctx context.Context, fp util.FullPath, tsNs int64) bool { +// +// removal asks a different question from a write. An entry still present at +// exactly tsNs has the write at that version reflected but not its removal -- +// a rename stamps the source and takes the name away at the same version -- +// so only a strictly newer record fences one out. A tombstone is the removal +// already reflected, and fences at tsNs like any other write. +func (mc *MetaCache) entryVersionBlocksLocked(ctx context.Context, fp util.FullPath, tsNs int64, removal bool) bool { recordTsNs, tombstone, unversioned := mc.entryVersionRecordLocked(ctx, fp) if unversioned { // Local content no log position describes: fence nothing, so any @@ -513,7 +519,11 @@ func (mc *MetaCache) entryVersionBlocksLocked(ctx context.Context, fp util.FullP if !tombstone && !mc.entryExistsLocked(ctx, fp) { recordTsNs = 0 } - return mc.entryVersionFloorLocked(fp, recordTsNs) >= tsNs + floorTsNs := mc.entryVersionFloorLocked(fp, recordTsNs) + if removal && !tombstone { + return floorTsNs > tsNs + } + return floorTsNs >= tsNs } // entryExistsLocked reports whether fp has a live entry, applying the same TTL @@ -951,10 +961,10 @@ func (mc *MetaCache) applyMetadataResponseLocked(ctx context.Context, resp *file // already reflected in it, and applying it would roll the entry back while // the version keeps the newer claim. Each half is gated independently. if resp.TsNs != 0 { - if oldPath != "" && mc.entryVersionBlocksLocked(ctx, oldPath, resp.TsNs) { + if oldPath != "" && mc.entryVersionBlocksLocked(ctx, oldPath, resp.TsNs, true) { oldPath = "" } - if newEntry != nil && mc.entryVersionBlocksLocked(ctx, newEntry.FullPath, resp.TsNs) { + if newEntry != nil && mc.entryVersionBlocksLocked(ctx, newEntry.FullPath, resp.TsNs, false) { newEntry = nil } } diff --git a/weed/mount/meta_cache/meta_cache_apply_test.go b/weed/mount/meta_cache/meta_cache_apply_test.go index 2f363815a..13f688270 100644 --- a/weed/mount/meta_cache/meta_cache_apply_test.go +++ b/weed/mount/meta_cache/meta_cache_apply_test.go @@ -868,7 +868,7 @@ func TestExpiredEntryIsJudgedByDirectoryFloor(t *testing.T) { // is the accurate answer, and an event above the floor must apply. mc.setEntryVersionLocked(context.Background(), util.FullPath("/dir/file"), 5000) mc.dirVersionFloors[util.FullPath("/dir")] = 3000 - blocks := mc.entryVersionBlocksLocked(context.Background(), util.FullPath("/dir/file"), 4000) + blocks := mc.entryVersionBlocksLocked(context.Background(), util.FullPath("/dir/file"), 4000, false) mc.Unlock() if blocks { diff --git a/weed/mount/meta_cache/meta_cache_rename_gate_test.go b/weed/mount/meta_cache/meta_cache_rename_gate_test.go new file mode 100644 index 000000000..e665983b0 --- /dev/null +++ b/weed/mount/meta_cache/meta_cache_rename_gate_test.go @@ -0,0 +1,95 @@ +package meta_cache + +import ( + "context" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/util" +) + +// A rename stamps its source and takes the name away at the same log position, +// so the removal arrives at exactly the version the source already records. +// Read as a write that is already reflected, it was dropped and the source +// stayed visible next to the destination. +func TestRenameRemovesSourceAtItsOwnVersion(t *testing.T) { + const version = 1787761708173717200 + + mc, _, _, _ := newTestMetaCache(t, map[util.FullPath]bool{"/": true, "/dir": true}) + defer mc.Shutdown() + ctx := context.Background() + + write := func(name string, tsNs int64) *filer_pb.SubscribeMetadataResponse { + return &filer_pb.SubscribeMetadataResponse{ + Directory: "/dir", + EventNotification: &filer_pb.EventNotification{ + NewEntry: &filer_pb.Entry{ + Name: name, + Attributes: &filer_pb.FuseAttributes{Crtime: 1, Mtime: 1, FileMode: 0100644, FileSize: 3}, + }, + }, + TsNs: tsNs, + } + } + for _, event := range []*filer_pb.SubscribeMetadataResponse{write("src", version), write("dst", version-1000)} { + if err := mc.ApplyMetadataResponseOwned(ctx, event, LocalMetadataResponseApplyOptions); err != nil { + t.Fatalf("seed %s: %v", event.EventNotification.NewEntry.Name, err) + } + } + + rename := &filer_pb.SubscribeMetadataResponse{ + Directory: "/dir", + EventNotification: &filer_pb.EventNotification{ + OldEntry: &filer_pb.Entry{Name: "src"}, + NewEntry: &filer_pb.Entry{ + Name: "dst", + Attributes: &filer_pb.FuseAttributes{Crtime: 1, Mtime: 1, FileMode: 0100644, FileSize: 3}, + }, + NewParentPath: "/dir", + }, + TsNs: version, + } + if err := mc.ApplyMetadataResponseOwned(ctx, rename, LocalMetadataResponseApplyOptions); err != nil { + t.Fatalf("rename: %v", err) + } + + if entry, _, err := mc.FindEntry(ctx, "/dir/src"); err == nil && entry != nil { + t.Error("the renamed-away source is still cached") + } + if entry, _, err := mc.FindEntry(ctx, "/dir/dst"); err != nil || entry == nil { + t.Errorf("the rename destination is missing: %v", err) + } +} + +// A removal older than the entry's record is still stale and must not apply: +// the record describes a write the removal never saw. +func TestRemovalOlderThanTheRecordStillFences(t *testing.T) { + mc, _, _, _ := newTestMetaCache(t, map[util.FullPath]bool{"/": true, "/dir": true}) + defer mc.Shutdown() + ctx := context.Background() + + if err := mc.ApplyMetadataResponseOwned(ctx, &filer_pb.SubscribeMetadataResponse{ + Directory: "/dir", + EventNotification: &filer_pb.EventNotification{ + NewEntry: &filer_pb.Entry{ + Name: "file", + Attributes: &filer_pb.FuseAttributes{Crtime: 1, Mtime: 1, FileMode: 0100644, FileSize: 3}, + }, + }, + TsNs: 2000, + }, LocalMetadataResponseApplyOptions); err != nil { + t.Fatalf("seed: %v", err) + } + + if err := mc.ApplyMetadataResponseOwned(ctx, &filer_pb.SubscribeMetadataResponse{ + Directory: "/dir", + EventNotification: &filer_pb.EventNotification{OldEntry: &filer_pb.Entry{Name: "file"}}, + TsNs: 1000, + }, LocalMetadataResponseApplyOptions); err != nil { + t.Fatalf("stale removal: %v", err) + } + + if entry, _, err := mc.FindEntry(ctx, "/dir/file"); err != nil || entry == nil { + t.Errorf("a removal older than the entry's record deleted it: %v", err) + } +} diff --git a/weed/mount/meta_cache/meta_cache_sections.go b/weed/mount/meta_cache/meta_cache_sections.go index 4dbba4786..a24402781 100644 --- a/weed/mount/meta_cache/meta_cache_sections.go +++ b/weed/mount/meta_cache/meta_cache_sections.go @@ -362,7 +362,7 @@ func (mc *MetaCache) applySectionRefreshNow(ctx context.Context, dirPath util.Fu mc.setEntryVersionLocked(ctx, entry.FullPath, 0) continue } - if mc.entryVersionBlocksLocked(ctx, entry.FullPath, snapshotTsNs) { + if mc.entryVersionBlocksLocked(ctx, entry.FullPath, snapshotTsNs, false) { continue } // An unversioned marker would bypass the section floor, so it cannot @@ -404,7 +404,7 @@ func (mc *MetaCache) applySectionRefreshNow(ctx context.Context, dirPath util.Fu if mc.pinnedChildFn != nil && mc.pinnedChildFn(entry) { continue } - if mc.entryVersionBlocksLocked(ctx, entry.FullPath, snapshotTsNs) { + if mc.entryVersionBlocksLocked(ctx, entry.FullPath, snapshotTsNs, true) { continue } if err := mc.localStore.DeleteEntry(ctx, entry.FullPath); err != nil { diff --git a/weed/mount/meta_cache/meta_cache_sections_test.go b/weed/mount/meta_cache/meta_cache_sections_test.go index 63200bc21..619995106 100644 --- a/weed/mount/meta_cache/meta_cache_sections_test.go +++ b/weed/mount/meta_cache/meta_cache_sections_test.go @@ -927,3 +927,48 @@ func TestRenameAcrossSectionsCountsBoth(t *testing.T) { t.Fatal("the landing side's section should be invalidated") } } + +// A refresh sweeps the names its listing did not return. One recorded at +// exactly the snapshot has its write reflected but not its removal, so reading +// the gate as a write left it cached for good. +func TestSectionRefreshSweepsANameRecordedAtTheSnapshot(t *testing.T) { + mc, _, _, _ := newTestMetaCache(t, map[util.FullPath]bool{"/": true, "/dir": true}) + defer mc.Shutdown() + + buildSectionedDir(t, mc, util.FullPath("/dir"), 1000, []string{"m"}) + + const snapshotTsNs = 5000 + if err := mc.ApplyMetadataResponse(context.Background(), &filer_pb.SubscribeMetadataResponse{ + Directory: "/dir", + EventNotification: &filer_pb.EventNotification{ + NewEntry: &filer_pb.Entry{ + Name: "b-vanish", + Attributes: &filer_pb.FuseAttributes{Crtime: 1, Mtime: 1, FileMode: 0100644, FileSize: 3}, + }, + }, + TsNs: snapshotTsNs, + }, SubscriberMetadataResponseApplyOptions); err != nil { + t.Fatalf("seed at the snapshot version: %v", err) + } + + if err := mc.enqueueAndWait(context.Background(), metadataApplyRequest{ + kind: metadataSectionRefresh, + buildPath: util.FullPath("/dir"), + refresh: §ionRefresh{ + hi: "m", + entries: []*filer.Entry{ + { + FullPath: util.FullPath("/dir/b-keep"), + Attr: filer.Attr{Crtime: time.Unix(1, 0), Mtime: time.Unix(2, 0), Mode: 0100644, FileSize: 7}, + }, + }, + snapshotTsNs: snapshotTsNs, + }, + }); err != nil { + t.Fatalf("refresh: %v", err) + } + + if entry, _, err := mc.FindEntry(context.Background(), util.FullPath("/dir/b-vanish")); err == nil && entry != nil { + t.Error("a name the refresh did not list is still cached") + } +}