Files
seaweedfs/weed/mount/meta_cache/meta_cache_apply_test.go
T
Chris LuandDevin 9e41500a3a mount: rebuild dropped directory listings and stale sections on lookup (#12195)
* mount: rebuild dropped directory listings on lookup

Invalidation, idle eviction, and a kernel Forget all clear a directory's
cached-listing flag, but only an expired listing was rebuilt on lookup — a
dropped one read through per entry until a ReadDir happened to run. Track
directories that once held a listing in wasCachedDirs, which survives those
drops, and let ShouldRebuildExpiredDir rebuild them on the next lookup.
Never-cached and read-through directories keep reading through.

* mount: re-validate a stale directory section on lookup

A burst of remote changes invalidates only the directory section it lands
in, but nothing re-validates that section until a ReadDir runs — so a
lookup-heavy workload reads through per entry indefinitely. EnsureNameFresh
re-lists just the section covering the looked-up name, after which lookups
in it are served from the cache again.

* mount: wait for pending flush before a stale-section refresh on lookup

A section refresh lists the range from the filer; without the wait a
pending async flush of the looked-up entry could land after the listing
and leave the refreshed section holding pre-flush metadata — the same
hazard the rebuild path already waits out.

* mount: bound the dropped-listing rebuild history

wasCachedDirs grew one full path per directory ever cached, removed only
by the read-through mark, so deleted and forgotten directories
accumulated forever. Remove the record when its name is deleted, move it
on rename so rebuild eligibility travels with the directory, and cap the
history with FIFO eviction — past the bound an evicted directory reads
through per entry, the same end state as a never-cached one.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* mount: drop oversized stale sections and back off failed refreshes

A section that outgrew one refresh failed EnsureNameFresh with
ErrRefreshRangeTooLarge on every lookup, repeating the multi-page
listing before each per-entry read, and other refresh failures retried
with no cooldown. Drop the cached listing on ErrRefreshRangeTooLarge so
a later lookup rebuilds it re-tiled, as ReadDir does, and stamp the
rebuild attempt on other failures so retries back off under the same
cooldown the rebuild gate uses.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* mount: version hard-link sibling syncs at the committed log position

* mount: tag dropped-listing queue slots so stale ones cannot evict

RemovePath, MovePath, and MarkDirectoryReadThrough delete the rebuild
record but leave the FIFO slot behind, so a path removed and re-cached
holds two queue entries. When the stale slot reaches the front it evicts
the newer record, and the directory reads through per lookup until the
next listing. Tag each record with an insertion generation and only
evict on a match, keeping the newer record for its own slot's turn.

* mount: refresh hard-link siblings from the filer, not the cached copy

Stamping a cached sibling with the mutation's version claimed that
position for content that may predate it — an in-flight event with
newer chunks or attributes then hits the version gate and the sibling
serves stale data until a refresh. Fetch each cached sibling's entry
from the filer after the commit and record the higher of the fetched
and committed positions, falling back to the in-place counter patch
only when the fetch itself fails.

* mount: back off failed section refreshes per section, not per directory

A failed refresh of one stale section marked a directory-wide attempt,
so every other stale section in that directory waited out the cooldown
with per-entry filer reads. Key the attempt timestamp by the section's
lower bound so unrelated ranges still refresh on their own schedule.

* mount: fence hard-linked entries by the group's newest write position

Every path of a hard-link group reads one shared content blob, but the
version gate only tracks per-path records — a sibling absent from
inodeToPath got no version from the sibling sync, so a stale section
refresh or event could roll the shared blob back for all of them.
Record a monotonic version floor per hard-link id wherever versioned
content lands, and check it in the section-refresh and event gates.

* mount: keep hard-link names landing when the group floor fences content

The per-group version floor rejected a delayed event's newEntry while
its oldPath still applied — a rename could drop the source without
adding the destination, and a section refresh could skip then mark a
live link fresh. Rebase the group's current cached content onto the
new path so the name lands without rolling the shared blob back, only
skip refresh entries already cached, and gate the sibling refresh
write under the same lock so a racing event or delete wins.

* mount: rebase hard-link entries from the shared blob when no sibling is cached

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* mount: park group content on uncached hard-link names a stale refresh installs

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* mount: stale the destination section when a fenced hard-link name cannot land

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* mount: bound the sibling hard-link refresh under one request deadline

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* mount: replace a reused-name occupant the hard-link floor would skip

* mount: claim no version for a sibling counter patched without a fetch

* mount: vacate a reused-name occupant without decrementing its group

A sibling write can fold the occupant's filer-side unlink into the shared
blob counter while the stale entry is still cached, so replacing it through
the plain insert may decrement a group that no longer counts the name and
drop a blob live siblings still read. Vacate the path like a rename first;
the counter then only ever drifts high, which leaks a blob rather than
losing shared content.

* mount: keep a fetch-failed sibling's version when patching its counter

The fallback marked the entry unversioned, which fences nothing — a
replayed delete older than the cached content could remove a live link.
Claim the version the entry already holds instead: newer in-flight events
still land and correct it, older replays stay fenced.

---------

Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
2026-10-29 00:55:47 -07:00

932 lines
30 KiB
Go

package meta_cache
import (
"context"
"os"
"path/filepath"
"sync"
"testing"
"time"
"github.com/seaweedfs/seaweedfs/weed/filer"
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
"github.com/seaweedfs/seaweedfs/weed/util"
)
func TestApplyMetadataResponseAppliesEventsInOrder(t *testing.T) {
mc, _, notifications, invalidations := newTestMetaCache(t, map[util.FullPath]bool{
"/": true,
"/dir": true,
})
defer mc.Shutdown()
createResp := &filer_pb.SubscribeMetadataResponse{
Directory: "/dir",
EventNotification: &filer_pb.EventNotification{
NewEntry: &filer_pb.Entry{
Name: "file.txt",
Attributes: &filer_pb.FuseAttributes{
Crtime: 1,
Mtime: 1,
FileMode: 0100644,
FileSize: 11,
},
},
},
}
updateResp := &filer_pb.SubscribeMetadataResponse{
Directory: "/dir",
EventNotification: &filer_pb.EventNotification{
OldEntry: &filer_pb.Entry{
Name: "file.txt",
},
NewEntry: &filer_pb.Entry{
Name: "file.txt",
Attributes: &filer_pb.FuseAttributes{
Crtime: 1,
Mtime: 2,
FileMode: 0100644,
FileSize: 29,
},
},
NewParentPath: "/dir",
},
}
deleteResp := &filer_pb.SubscribeMetadataResponse{
Directory: "/dir",
EventNotification: &filer_pb.EventNotification{
OldEntry: &filer_pb.Entry{
Name: "file.txt",
},
},
}
if err := mc.ApplyMetadataResponse(context.Background(), createResp, SubscriberMetadataResponseApplyOptions); err != nil {
t.Fatalf("apply create: %v", err)
}
entry, _, err := mc.FindEntry(context.Background(), util.FullPath("/dir/file.txt"))
if err != nil {
t.Fatalf("find created entry: %v", err)
}
if entry.FileSize != 11 {
t.Fatalf("created file size = %d, want 11", entry.FileSize)
}
if err := mc.ApplyMetadataResponse(context.Background(), updateResp, SubscriberMetadataResponseApplyOptions); err != nil {
t.Fatalf("apply update: %v", err)
}
entry, _, err = mc.FindEntry(context.Background(), util.FullPath("/dir/file.txt"))
if err != nil {
t.Fatalf("find updated entry: %v", err)
}
if entry.FileSize != 29 {
t.Fatalf("updated file size = %d, want 29", entry.FileSize)
}
if err := mc.ApplyMetadataResponse(context.Background(), deleteResp, SubscriberMetadataResponseApplyOptions); err != nil {
t.Fatalf("apply delete: %v", err)
}
entry, _, err = mc.FindEntry(context.Background(), util.FullPath("/dir/file.txt"))
if err != filer_pb.ErrNotFound {
t.Fatalf("find deleted entry error = %v, want %v", err, filer_pb.ErrNotFound)
}
if entry != nil {
t.Fatalf("deleted entry still cached: %+v", entry)
}
if got := countPath(notifications.paths(), util.FullPath("/dir")); got != 3 {
t.Fatalf("directory notifications for /dir = %d, want 3", got)
}
mc.WaitForEntryInvalidations()
if got := countPath(invalidations.paths(), util.FullPath("/dir/file.txt")); got != 3 {
t.Fatalf("invalidations for /dir/file.txt = %d, want 3 (create + update + delete)", got)
}
}
func TestApplyMetadataResponseRenamesAcrossCachedDirectories(t *testing.T) {
mc, _, notifications, invalidations := newTestMetaCache(t, map[util.FullPath]bool{
"/": true,
"/src": true,
"/dst": true,
})
defer mc.Shutdown()
if err := mc.InsertEntry(context.Background(), &filer.Entry{
FullPath: "/src/file.tmp",
Attr: filer.Attr{
Crtime: time.Unix(1, 0),
Mtime: time.Unix(1, 0),
Mode: 0100644,
FileSize: 7,
},
}, 0); err != nil {
t.Fatalf("insert source entry: %v", err)
}
renameResp := &filer_pb.SubscribeMetadataResponse{
Directory: "/src",
EventNotification: &filer_pb.EventNotification{
OldEntry: &filer_pb.Entry{
Name: "file.tmp",
},
NewEntry: &filer_pb.Entry{
Name: "file.txt",
Attributes: &filer_pb.FuseAttributes{
Crtime: 1,
Mtime: 2,
FileMode: 0100644,
FileSize: 41,
},
},
NewParentPath: "/dst",
},
}
if err := mc.ApplyMetadataResponse(context.Background(), renameResp, SubscriberMetadataResponseApplyOptions); err != nil {
t.Fatalf("apply rename: %v", err)
}
oldEntry, _, err := mc.FindEntry(context.Background(), util.FullPath("/src/file.tmp"))
if err != filer_pb.ErrNotFound {
t.Fatalf("find old path error = %v, want %v", err, filer_pb.ErrNotFound)
}
if oldEntry != nil {
t.Fatalf("old path still cached: %+v", oldEntry)
}
newEntry, _, err := mc.FindEntry(context.Background(), util.FullPath("/dst/file.txt"))
if err != nil {
t.Fatalf("find new path: %v", err)
}
if newEntry.FileSize != 41 {
t.Fatalf("renamed file size = %d, want 41", newEntry.FileSize)
}
if got := countPath(notifications.paths(), util.FullPath("/src")); got != 1 {
t.Fatalf("directory notifications for /src = %d, want 1", got)
}
if got := countPath(notifications.paths(), util.FullPath("/dst")); got != 1 {
t.Fatalf("directory notifications for /dst = %d, want 1", got)
}
mc.WaitForEntryInvalidations()
if got := countPath(invalidations.paths(), util.FullPath("/src/file.tmp")); got != 1 {
t.Fatalf("invalidations for /src/file.tmp = %d, want 1", got)
}
if got := countPath(invalidations.paths(), util.FullPath("/dst/file.txt")); got != 1 {
t.Fatalf("invalidations for /dst/file.txt = %d, want 1", got)
}
}
func TestApplyMetadataResponseLocalOptionsSkipInvalidations(t *testing.T) {
mc, _, notifications, invalidations := newTestMetaCache(t, map[util.FullPath]bool{
"/": true,
"/dir": true,
})
defer mc.Shutdown()
if err := mc.InsertEntry(context.Background(), &filer.Entry{
FullPath: "/dir/file.txt",
Attr: filer.Attr{
Crtime: time.Unix(1, 0),
Mtime: time.Unix(1, 0),
Mode: 0100644,
FileSize: 7,
},
}, 0); err != nil {
t.Fatalf("insert source entry: %v", err)
}
updateResp := &filer_pb.SubscribeMetadataResponse{
Directory: "/dir",
EventNotification: &filer_pb.EventNotification{
OldEntry: &filer_pb.Entry{
Name: "file.txt",
},
NewEntry: &filer_pb.Entry{
Name: "file.txt",
Attributes: &filer_pb.FuseAttributes{
Crtime: 1,
Mtime: 2,
FileMode: 0100644,
FileSize: 17,
},
},
NewParentPath: "/dir",
},
}
if err := mc.ApplyMetadataResponse(context.Background(), updateResp, LocalMetadataResponseApplyOptions); err != nil {
t.Fatalf("apply local update: %v", err)
}
entry, _, err := mc.FindEntry(context.Background(), util.FullPath("/dir/file.txt"))
if err != nil {
t.Fatalf("find updated entry: %v", err)
}
if entry.FileSize != 17 {
t.Fatalf("updated file size = %d, want 17", entry.FileSize)
}
if got := countPath(notifications.paths(), util.FullPath("/dir")); got != 1 {
t.Fatalf("directory notifications for /dir = %d, want 1", got)
}
mc.WaitForEntryInvalidations()
if got := len(invalidations.paths()); got != 0 {
t.Fatalf("invalidations = %d, want 0", got)
}
}
func TestApplyMetadataResponseDeduplicatesRepeatedFilerEvent(t *testing.T) {
mc, _, notifications, invalidations := newTestMetaCache(t, map[util.FullPath]bool{
"/": true,
"/dir": true,
})
defer mc.Shutdown()
if err := mc.InsertEntry(context.Background(), &filer.Entry{
FullPath: "/dir/file.txt",
Attr: filer.Attr{
Crtime: time.Unix(1, 0),
Mtime: time.Unix(1, 0),
Mode: 0100644,
FileSize: 5,
},
}, 0); err != nil {
t.Fatalf("insert source entry: %v", err)
}
updateResp := &filer_pb.SubscribeMetadataResponse{
Directory: "/dir",
EventNotification: &filer_pb.EventNotification{
OldEntry: &filer_pb.Entry{
Name: "file.txt",
},
NewEntry: &filer_pb.Entry{
Name: "file.txt",
Attributes: &filer_pb.FuseAttributes{
Crtime: 1,
Mtime: 2,
FileMode: 0100644,
FileSize: 15,
},
},
NewParentPath: "/dir",
Signatures: []int32{7},
},
TsNs: 99,
}
if err := mc.ApplyMetadataResponse(context.Background(), updateResp, SubscriberMetadataResponseApplyOptions); err != nil {
t.Fatalf("first apply: %v", err)
}
if err := mc.ApplyMetadataResponse(context.Background(), updateResp, SubscriberMetadataResponseApplyOptions); err != nil {
t.Fatalf("second apply: %v", err)
}
entry, _, err := mc.FindEntry(context.Background(), util.FullPath("/dir/file.txt"))
if err != nil {
t.Fatalf("find updated entry: %v", err)
}
if entry.FileSize != 15 {
t.Fatalf("updated file size = %d, want 15", entry.FileSize)
}
if got := countPath(notifications.paths(), util.FullPath("/dir")); got != 1 {
t.Fatalf("directory notifications for /dir = %d, want 1", got)
}
mc.WaitForEntryInvalidations()
if got := countPath(invalidations.paths(), util.FullPath("/dir/file.txt")); got != 1 {
t.Fatalf("invalidations for /dir/file.txt = %d, want 1", got)
}
}
func TestApplyMetadataResponseSkipsHiddenSystemEntryWhenDisabled(t *testing.T) {
mc, _, _, _ := newTestMetaCache(t, map[util.FullPath]bool{
"/": true,
})
defer mc.Shutdown()
createResp := &filer_pb.SubscribeMetadataResponse{
Directory: "/",
EventNotification: &filer_pb.EventNotification{
NewEntry: &filer_pb.Entry{
Name: "topics",
Attributes: &filer_pb.FuseAttributes{
Crtime: 1,
Mtime: 1,
FileMode: uint32(os.ModeDir | 0o755),
},
IsDirectory: true,
},
},
}
if err := mc.ApplyMetadataResponse(context.Background(), createResp, SubscriberMetadataResponseApplyOptions); err != nil {
t.Fatalf("apply create: %v", err)
}
entry, _, err := mc.FindEntry(context.Background(), util.FullPath("/topics"))
if err != filer_pb.ErrNotFound {
t.Fatalf("find hidden entry error = %v, want %v", err, filer_pb.ErrNotFound)
}
if entry != nil {
t.Fatalf("hidden entry still cached: %+v", entry)
}
}
func TestApplyMetadataResponsePurgesHiddenDestinationPath(t *testing.T) {
mc, _, _, _ := newTestMetaCache(t, map[util.FullPath]bool{
"/": true,
"/src": true,
})
defer mc.Shutdown()
if err := mc.InsertEntry(context.Background(), &filer.Entry{
FullPath: "/topics",
Attr: filer.Attr{
Crtime: time.Unix(1, 0),
Mtime: time.Unix(1, 0),
Mode: os.ModeDir | 0o755,
},
}, 0); err != nil {
t.Fatalf("insert stale hidden dir: %v", err)
}
if err := mc.InsertEntry(context.Background(), &filer.Entry{
FullPath: "/topics/leaked.txt",
Attr: filer.Attr{
Crtime: time.Unix(1, 0),
Mtime: time.Unix(1, 0),
Mode: 0o644,
FileSize: 7,
},
}, 0); err != nil {
t.Fatalf("insert leaked hidden child: %v", err)
}
if err := mc.InsertEntry(context.Background(), &filer.Entry{
FullPath: "/src/visible",
Attr: filer.Attr{
Crtime: time.Unix(1, 0),
Mtime: time.Unix(1, 0),
Mode: os.ModeDir | 0o755,
},
}, 0); err != nil {
t.Fatalf("insert source dir: %v", err)
}
renameResp := &filer_pb.SubscribeMetadataResponse{
Directory: "/src",
EventNotification: &filer_pb.EventNotification{
OldEntry: &filer_pb.Entry{
Name: "visible",
IsDirectory: true,
},
NewEntry: &filer_pb.Entry{
Name: "topics",
Attributes: &filer_pb.FuseAttributes{
Crtime: 2,
Mtime: 2,
FileMode: uint32(os.ModeDir | 0o755),
},
IsDirectory: true,
},
NewParentPath: "/",
},
}
if err := mc.ApplyMetadataResponse(context.Background(), renameResp, SubscriberMetadataResponseApplyOptions); err != nil {
t.Fatalf("apply rename: %v", err)
}
if entry, _, err := mc.FindEntry(context.Background(), util.FullPath("/src/visible")); err != filer_pb.ErrNotFound || entry != nil {
t.Fatalf("source dir after rename = %+v, %v; want nil, %v", entry, err, filer_pb.ErrNotFound)
}
if entry, _, err := mc.FindEntry(context.Background(), util.FullPath("/topics")); err != filer_pb.ErrNotFound || entry != nil {
t.Fatalf("hidden destination after rename = %+v, %v; want nil, %v", entry, err, filer_pb.ErrNotFound)
}
if entry, _, err := mc.FindEntry(context.Background(), util.FullPath("/topics/leaked.txt")); err != filer_pb.ErrNotFound || entry != nil {
t.Fatalf("hidden child after rename = %+v, %v; want nil, %v", entry, err, filer_pb.ErrNotFound)
}
}
// The entry attached to each invalidation is what an open file handle gets
// refreshed with, so it must be the entry now at that path — or nil when the
// path was vacated and the handle should keep its last entry — versioned by
// the event's filer log timestamp.
func TestCollectEntryInvalidationsCarryAuthoritativeEntries(t *testing.T) {
newEntry := &filer_pb.Entry{
Name: "file.txt",
Attributes: &filer_pb.FuseAttributes{FileSize: 42},
}
update := &filer_pb.SubscribeMetadataResponse{
Directory: "/dir",
TsNs: 77,
EventNotification: &filer_pb.EventNotification{
OldEntry: &filer_pb.Entry{Name: "file.txt"},
NewEntry: newEntry,
NewParentPath: "/dir",
},
}
got := collectEntryInvalidations(update)
if len(got) != 1 || got[0].Path != "/dir/file.txt" || got[0].Entry != newEntry || got[0].TsNs != 77 {
t.Fatalf("in-place update invalidations = %+v, want [{/dir/file.txt NewEntry ts 77}]", got)
}
rename := &filer_pb.SubscribeMetadataResponse{
Directory: "/src",
TsNs: 78,
EventNotification: &filer_pb.EventNotification{
OldEntry: &filer_pb.Entry{Name: "file.tmp"},
NewEntry: newEntry,
NewParentPath: "/dst",
},
}
got = collectEntryInvalidations(rename)
if len(got) != 2 || got[0].Path != "/src/file.tmp" || got[0].Entry != nil || got[0].TsNs != 78 ||
got[1].Path != "/dst/file.txt" || got[1].Entry != newEntry || got[1].TsNs != 78 {
t.Fatalf("rename invalidations = %+v, want [{/src/file.tmp nil} {/dst/file.txt NewEntry}]", got)
}
create := &filer_pb.SubscribeMetadataResponse{
Directory: "/dir",
TsNs: 79,
EventNotification: &filer_pb.EventNotification{
NewEntry: newEntry,
},
}
got = collectEntryInvalidations(create)
if len(got) != 1 || got[0].Path != "/dir/file.txt" || got[0].Entry != newEntry || got[0].TsNs != 79 {
t.Fatalf("create invalidations = %+v, want [{/dir/file.txt NewEntry ts 79}]", got)
}
deleteResp := &filer_pb.SubscribeMetadataResponse{
Directory: "/dir",
TsNs: 80,
EventNotification: &filer_pb.EventNotification{
OldEntry: &filer_pb.Entry{Name: "file.txt"},
},
}
got = collectEntryInvalidations(deleteResp)
if len(got) != 1 || got[0].Path != "/dir/file.txt" || got[0].Entry != nil || got[0].TsNs != 80 {
t.Fatalf("delete invalidations = %+v, want [{/dir/file.txt nil ts 80}]", got)
}
}
func newTestMetaCache(t *testing.T, cached map[util.FullPath]bool) (*MetaCache, map[util.FullPath]bool, *recordedPaths, *recordedPaths) {
t.Helper()
mapper, err := NewUidGidMapper("", "")
if err != nil {
t.Fatalf("uid/gid mapper: %v", err)
}
var cachedMu sync.Mutex
notifications := &recordedPaths{}
invalidations := &recordedPaths{}
mc := NewMetaCache(
filepath.Join(t.TempDir(), "meta"),
mapper,
util.FullPath("/"),
false,
func(path util.FullPath) {
cachedMu.Lock()
defer cachedMu.Unlock()
cached[path] = true
},
func(path util.FullPath) bool {
cachedMu.Lock()
defer cachedMu.Unlock()
return cached[path]
},
func(inv EntryInvalidation) {
invalidations.record(inv.Path)
},
func(dir util.FullPath) {
notifications.record(dir)
},
)
return mc, cached, notifications, invalidations
}
type recordedPaths struct {
mu sync.Mutex
items []util.FullPath
}
func (r *recordedPaths) record(path util.FullPath) {
r.mu.Lock()
defer r.mu.Unlock()
r.items = append(r.items, path)
}
func (r *recordedPaths) paths() []util.FullPath {
r.mu.Lock()
defer r.mu.Unlock()
return append([]util.FullPath(nil), r.items...)
}
func countPath(paths []util.FullPath, target util.FullPath) int {
count := 0
for _, path := range paths {
if path == target {
count++
}
}
return count
}
// An unversioned local write replaces the content, so the previous version
// claim no longer describes it and must be cleared — keeping it would fence
// out events correcting the unversioned state.
func TestUnversionedWriteClearsEntryVersion(t *testing.T) {
mc, _, _, _ := newTestMetaCache(t, map[util.FullPath]bool{
"/": true,
"/dir": true,
})
defer mc.Shutdown()
versioned := &filer_pb.SubscribeMetadataResponse{
Directory: "/dir",
TsNs: 1000,
EventNotification: &filer_pb.EventNotification{
NewEntry: &filer_pb.Entry{
Name: "file.txt",
Attributes: &filer_pb.FuseAttributes{FileSize: 11, FileMode: 0100644},
},
},
}
if err := mc.ApplyMetadataResponse(context.Background(), versioned, SubscriberMetadataResponseApplyOptions); err != nil {
t.Fatalf("apply versioned event: %v", err)
}
if _, versionTsNs, err := mc.FindEntry(context.Background(), util.FullPath("/dir/file.txt")); err != nil || versionTsNs != 1000 {
t.Fatalf("versioned entry = ts %d, %v; want 1000", versionTsNs, err)
}
if err := mc.UpdateEntry(context.Background(), &filer.Entry{
FullPath: "/dir/file.txt",
Attr: filer.Attr{
Crtime: time.Unix(1, 0),
Mtime: time.Unix(2, 0),
Mode: 0100644,
FileSize: 22,
},
}, 0); err != nil {
t.Fatalf("local update: %v", err)
}
if _, versionTsNs, err := mc.FindEntry(context.Background(), util.FullPath("/dir/file.txt")); err != nil || versionTsNs != 0 {
t.Fatalf("after unversioned write = ts %d, %v; want 0", versionTsNs, err)
}
mc.WaitForEntryInvalidations()
}
// A rebuild against a pre-upgrade filer returns no snapshot; the reinserted
// entries must not inherit the version records their pre-rebuild
// incarnations left behind, or valid events below the stale claim are
// rejected.
func TestUnversionedRebuildClearsStaleVersions(t *testing.T) {
mc, _, _, _ := newTestMetaCache(t, map[util.FullPath]bool{
"/": true,
"/dir": true,
})
defer mc.Shutdown()
versioned := &filer_pb.SubscribeMetadataResponse{
Directory: "/dir",
TsNs: 3000,
EventNotification: &filer_pb.EventNotification{
NewEntry: &filer_pb.Entry{
Name: "file.txt",
Attributes: &filer_pb.FuseAttributes{FileSize: 11, FileMode: 0100644},
},
},
}
if err := mc.ApplyMetadataResponse(context.Background(), versioned, SubscriberMetadataResponseApplyOptions); err != nil {
t.Fatalf("apply versioned event: %v", err)
}
if err := mc.BeginDirectoryBuild(context.Background(), util.FullPath("/dir")); err != nil {
t.Fatalf("begin build: %v", err)
}
if err := mc.doBatchInsertEntries(context.Background(), []*filer.Entry{{
FullPath: "/dir/file.txt",
Attr: filer.Attr{
Crtime: time.Unix(1, 0),
Mtime: time.Unix(1, 0),
Mode: 0100644,
FileSize: 22,
},
}}); err != nil {
t.Fatalf("batch insert: %v", err)
}
if err := mc.CompleteDirectoryBuild(context.Background(), util.FullPath("/dir"), 0, nil); err != nil {
t.Fatalf("complete unversioned build: %v", err)
}
if _, versionTsNs, err := mc.FindEntry(context.Background(), util.FullPath("/dir/file.txt")); err != nil || versionTsNs != 0 {
t.Fatalf("rebuilt entry version = %d, %v; want 0 (stale claim must not survive an unversioned rebuild)", versionTsNs, err)
}
mc.WaitForEntryInvalidations()
}
// Tombstones are scoped to directories whose cached state the fence
// protects; an uncached parent never serves from the store nor applies the
// resurrecting insert, so a tombstone there would only accumulate.
func TestTombstonesScopedToCachedDirectories(t *testing.T) {
mc, _, _, _ := newTestMetaCache(t, map[util.FullPath]bool{
"/": true,
"/cached": true,
})
defer mc.Shutdown()
deleteEventFor := func(dir, name string, tsNs int64) *filer_pb.SubscribeMetadataResponse {
return &filer_pb.SubscribeMetadataResponse{
Directory: dir,
TsNs: tsNs,
EventNotification: &filer_pb.EventNotification{
OldEntry: &filer_pb.Entry{Name: name},
},
}
}
if err := mc.ApplyMetadataResponse(context.Background(), deleteEventFor("/cached", "file", 2000), SubscriberMetadataResponseApplyOptions); err != nil {
t.Fatalf("apply cached delete: %v", err)
}
if tsNs, tombstone := mc.getEntryVersionRecordLocked(context.Background(), util.FullPath("/cached/file")); tsNs != 2000 || !tombstone {
t.Fatalf("cached-dir delete record = (%d, %v), want tombstone at 2000", tsNs, tombstone)
}
if err := mc.ApplyMetadataResponse(context.Background(), deleteEventFor("/uncached", "file", 2000), SubscriberMetadataResponseApplyOptions); err != nil {
t.Fatalf("apply uncached delete: %v", err)
}
if tsNs, _ := mc.getEntryVersionRecordLocked(context.Background(), util.FullPath("/uncached/file")); tsNs != 0 {
t.Fatalf("uncached-dir delete record = %d, want none", tsNs)
}
mc.WaitForEntryInvalidations()
}
// A completed listing's absence floor supersedes the direct-child tombstones
// at or below its snapshot; newer tombstones survive.
func TestBuildCompletionPrunesSupersededTombstones(t *testing.T) {
mc, _, _, _ := newTestMetaCache(t, map[util.FullPath]bool{
"/": true,
"/dir": true,
})
defer mc.Shutdown()
apply := func(resp *filer_pb.SubscribeMetadataResponse) {
t.Helper()
if err := mc.ApplyMetadataResponse(context.Background(), resp, SubscriberMetadataResponseApplyOptions); err != nil {
t.Fatalf("apply: %v", err)
}
}
createEvent := func(name string, tsNs int64) *filer_pb.SubscribeMetadataResponse {
return &filer_pb.SubscribeMetadataResponse{
Directory: "/dir",
TsNs: tsNs,
EventNotification: &filer_pb.EventNotification{
NewEntry: &filer_pb.Entry{
Name: name,
Attributes: &filer_pb.FuseAttributes{FileSize: 1, FileMode: 0100644},
},
},
}
}
deleteEvent := func(name string, tsNs int64) *filer_pb.SubscribeMetadataResponse {
return &filer_pb.SubscribeMetadataResponse{
Directory: "/dir",
TsNs: tsNs,
EventNotification: &filer_pb.EventNotification{
OldEntry: &filer_pb.Entry{Name: name},
},
}
}
apply(createEvent("a", 900))
apply(deleteEvent("a", 1000))
apply(createEvent("b", 3900))
apply(deleteEvent("b", 4000))
// A deeper descendant's tombstone belongs to its own directory's floor.
mc.Lock()
mc.setEntryTombstoneLocked(context.Background(), util.FullPath("/dir/sub/x"), 1000)
mc.Unlock()
if err := mc.BeginDirectoryBuild(context.Background(), util.FullPath("/dir")); err != nil {
t.Fatalf("begin build: %v", err)
}
if err := mc.CompleteDirectoryBuild(context.Background(), util.FullPath("/dir"), 3000, nil); err != nil {
t.Fatalf("complete build: %v", err)
}
if tsNs, _ := mc.getEntryVersionRecordLocked(context.Background(), util.FullPath("/dir/a")); tsNs != 0 {
t.Fatalf("tombstone at 1000 should be pruned by the floor at 3000, got %d", tsNs)
}
if tsNs, tombstone := mc.getEntryVersionRecordLocked(context.Background(), util.FullPath("/dir/b")); tsNs != 4000 || !tombstone {
t.Fatalf("tombstone at 4000 must survive the floor at 3000, got (%d, %v)", tsNs, tombstone)
}
if tsNs, tombstone := mc.getEntryVersionRecordLocked(context.Background(), util.FullPath("/dir/sub/x")); tsNs != 1000 || !tombstone {
t.Fatalf("descendant tombstone must survive the parent's prune, got (%d, %v)", tsNs, tombstone)
}
mc.WaitForEntryInvalidations()
}
// Evicting a directory clears its children's version records (plain and
// tombstone) so they cannot accumulate for the lifetime of a long-running
// mount.
func TestDirectoryEvictionClearsVersionRecords(t *testing.T) {
mc, _, _, _ := newTestMetaCache(t, map[util.FullPath]bool{
"/": true,
"/dir": true,
})
defer mc.Shutdown()
create := &filer_pb.SubscribeMetadataResponse{
Directory: "/dir",
TsNs: 1000,
EventNotification: &filer_pb.EventNotification{
NewEntry: &filer_pb.Entry{Name: "keep", Attributes: &filer_pb.FuseAttributes{FileSize: 1, FileMode: 0100644}},
},
}
if err := mc.ApplyMetadataResponse(context.Background(), create, SubscriberMetadataResponseApplyOptions); err != nil {
t.Fatalf("apply create: %v", err)
}
del := &filer_pb.SubscribeMetadataResponse{
Directory: "/dir",
TsNs: 1100,
EventNotification: &filer_pb.EventNotification{OldEntry: &filer_pb.Entry{Name: "gone"}},
}
if err := mc.ApplyMetadataResponse(context.Background(), del, SubscriberMetadataResponseApplyOptions); err != nil {
t.Fatalf("apply delete: %v", err)
}
if tsNs, _ := mc.getEntryVersionRecordLocked(context.Background(), util.FullPath("/dir/keep")); tsNs != 1000 {
t.Fatalf("version record for /dir/keep = %d, want 1000 before eviction", tsNs)
}
if tsNs, tombstone := mc.getEntryVersionRecordLocked(context.Background(), util.FullPath("/dir/gone")); tsNs != 1100 || !tombstone {
t.Fatalf("tombstone for /dir/gone = (%d,%v), want (1100,true) before eviction", tsNs, tombstone)
}
if err := mc.DeleteFolderChildren(context.Background(), util.FullPath("/dir")); err != nil {
t.Fatalf("evict directory: %v", err)
}
if tsNs, _ := mc.getEntryVersionRecordLocked(context.Background(), util.FullPath("/dir/keep")); tsNs != 0 {
t.Fatalf("version record for /dir/keep survived eviction: %d", tsNs)
}
if tsNs, _ := mc.getEntryVersionRecordLocked(context.Background(), util.FullPath("/dir/gone")); tsNs != 0 {
t.Fatalf("tombstone for /dir/gone survived eviction: %d", tsNs)
}
mc.WaitForEntryInvalidations()
}
// A completed build versions its children through one directory floor rather
// than a version record per child, and the floor fences exactly as per-child
// records did.
func TestBuildFloorVersionsChildrenWithoutPerChildRecords(t *testing.T) {
mc, _, _, _ := newTestMetaCache(t, map[util.FullPath]bool{
"/": true,
"/dir": true,
})
defer mc.Shutdown()
if err := mc.BeginDirectoryBuild(context.Background(), util.FullPath("/dir")); err != nil {
t.Fatalf("begin build: %v", err)
}
if err := mc.doBatchInsertEntries(context.Background(), []*filer.Entry{{
FullPath: "/dir/file",
Attr: filer.Attr{
Crtime: time.Unix(1, 0),
Mtime: time.Unix(1, 0),
Mode: 0100644,
FileSize: 300,
},
}}); err != nil {
t.Fatalf("batch insert: %v", err)
}
if err := mc.CompleteDirectoryBuild(context.Background(), util.FullPath("/dir"), 2000, nil); err != nil {
t.Fatalf("complete build: %v", err)
}
// No per-child record was written...
if tsNs, _ := mc.getEntryVersionRecordLocked(context.Background(), util.FullPath("/dir/file")); tsNs != 0 {
t.Fatalf("per-child version record = %d, want none (the floor versions the child)", tsNs)
}
// ...yet the child reads back at the listing snapshot.
if _, versionTsNs, err := mc.FindEntry(context.Background(), util.FullPath("/dir/file")); err != nil || versionTsNs != 2000 {
t.Fatalf("child version = %d, %v; want 2000 from the directory floor", versionTsNs, err)
}
// And an event the snapshot already covered is still fenced out.
covered := &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, FileMode: 0100644}},
NewParentPath: "/dir",
},
}
if err := mc.ApplyMetadataResponse(context.Background(), covered, SubscriberMetadataResponseApplyOptions); err != nil {
t.Fatalf("apply covered event: %v", err)
}
entry, _, err := mc.FindEntry(context.Background(), util.FullPath("/dir/file"))
if err != nil || entry.FileSize != 300 {
t.Fatalf("entry = %+v, %v; want size 300 (floor must fence the covered event)", entry, err)
}
mc.WaitForEntryInvalidations()
}
// The presence probe applies the same TTL expiry the read path does, so a
// logically-expired entry is judged by its directory's floor rather than by
// the stale version record it still carries.
func TestExpiredEntryIsJudgedByDirectoryFloor(t *testing.T) {
mc, _, _, _ := newTestMetaCache(t, map[util.FullPath]bool{
"/": true,
"/dir": true,
})
defer mc.Shutdown()
if err := mc.InsertEntry(context.Background(), &filer.Entry{
FullPath: "/dir/file",
Attr: filer.Attr{
Crtime: time.Now().Add(-2 * time.Hour),
Mtime: time.Now().Add(-2 * time.Hour),
Mode: 0100644,
FileSize: 7,
TtlSec: 1,
},
}, 0); err != nil {
t.Fatalf("insert expiring entry: %v", err)
}
mc.Lock()
// A record newer than the directory floor: it only governs while the entry
// it describes still exists. Once the entry has logically expired the floor
// 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, false)
mc.Unlock()
if blocks {
t.Fatal("event at 4000 fenced by an expired entry's stale record at 5000; the floor at 3000 should govern")
}
}
// An unversioned local write leaves content no log position describes, so it
// must not inherit the directory's listing floor — that would fence the very
// events needed to correct it.
func TestUnversionedWriteDoesNotInheritDirectoryFloor(t *testing.T) {
mc, _, _, _ := newTestMetaCache(t, map[util.FullPath]bool{
"/": true,
"/dir": true,
})
defer mc.Shutdown()
if err := mc.BeginDirectoryBuild(context.Background(), util.FullPath("/dir")); err != nil {
t.Fatalf("begin build: %v", err)
}
if err := mc.CompleteDirectoryBuild(context.Background(), util.FullPath("/dir"), 2000, nil); err != nil {
t.Fatalf("complete build: %v", err)
}
// A local write with no log position behind it.
if err := mc.InsertEntry(context.Background(), &filer.Entry{
FullPath: "/dir/file",
Attr: filer.Attr{
Crtime: time.Unix(1, 0),
Mtime: time.Unix(1, 0),
Mode: 0100644,
FileSize: 7,
},
}, 0); err != nil {
t.Fatalf("local insert: %v", err)
}
if _, versionTsNs, err := mc.FindEntry(context.Background(), util.FullPath("/dir/file")); err != nil || versionTsNs != 0 {
t.Fatalf("version = %d, %v; want 0 (unversioned content must not inherit the floor at 2000)", versionTsNs, err)
}
// And an event below the floor can still correct it.
correcting := &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: 42, FileMode: 0100644}},
NewParentPath: "/dir",
},
}
if err := mc.ApplyMetadataResponse(context.Background(), correcting, SubscriberMetadataResponseApplyOptions); err != nil {
t.Fatalf("apply correcting event: %v", err)
}
entry, _, err := mc.FindEntry(context.Background(), util.FullPath("/dir/file"))
if err != nil || entry.FileSize != 42 {
t.Fatalf("entry = %+v, %v; want size 42 (the floor must not fence a correction of unversioned content)", entry, err)
}
mc.WaitForEntryInvalidations()
}