From 9e41500a3a8070ff067cbff41ac4dab1256bd0ba Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Thu, 29 Oct 2026 00:55:47 -0700 Subject: [PATCH] mount: rebuild dropped directory listings and stale sections on lookup (#12195) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * 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> --- weed/mount/inode_to_path.go | 116 +++++++- weed/mount/meta_cache/meta_cache.go | 114 ++++++- .../mount/meta_cache/meta_cache_apply_test.go | 2 +- weed/mount/meta_cache/meta_cache_sections.go | 65 ++++ .../meta_cache/meta_cache_sections_test.go | 277 ++++++++++++++++++ weed/mount/weedfs.go | 82 ++++-- weed/mount/weedfs_dir_lookup_cache_test.go | 181 ++++++++++++ weed/mount/weedfs_file_mkrm.go | 2 +- weed/mount/weedfs_link.go | 69 ++++- weed/mount/weedfs_link_test.go | 79 ++++- 10 files changed, 931 insertions(+), 56 deletions(-) diff --git a/weed/mount/inode_to_path.go b/weed/mount/inode_to_path.go index 6a837872c4..5ac35bbc42 100644 --- a/weed/mount/inode_to_path.go +++ b/weed/mount/inode_to_path.go @@ -21,6 +21,25 @@ type InodeToPath struct { // dirPaths indexes the same states by path, so a directory lookup never // goes through the map that holds one full path per inode. dirPaths map[util.FullPath]*dirState + // wasCachedDirs remembers directories that once held a fully cached + // listing. Unlike the flags on dirState it survives invalidation, idle + // eviction, and the state drop on kernel Forget, so a lookup can rebuild + // a dropped listing instead of reading through per entry forever. + // MarkDirectoryReadThrough removes oversized directories from it, + // RemovePath drops deleted ones, and wasCachedOrder bounds the retained + // history with FIFO eviction. Records are tagged with an insertion + // generation so a stale queue slot can never evict a path's newer record. + wasCachedDirs map[util.FullPath]uint64 + wasCachedOrder []wasCachedEntry + wasCachedGen uint64 +} + +// wasCachedEntry pairs a queued path with the insertion generation that put +// it there; removals leave the slot behind, and only a matching generation +// may evict the live record. +type wasCachedEntry struct { + path util.FullPath + gen uint64 } // InodeEntry exists per inode the kernel references. Directory cache state is @@ -41,7 +60,11 @@ type dirState struct { lastAccess time.Time lastRefresh time.Time lastRebuildAttempt time.Time - subdirCount int32 // tracked in-memory for POSIX directory nlink + // lastSectionAttempts backs off section refreshes per section (keyed by + // the section's lower bound) so one failing range does not stall the + // rest of the directory. + lastSectionAttempts map[string]time.Time + subdirCount int32 // tracked in-memory for POSIX directory nlink } func (d *dirState) resetCacheState() { @@ -49,6 +72,7 @@ func (d *dirState) resetCacheState() { d.readDirDirect = false d.cachedExpiresTime = time.Time{} d.lastRebuildAttempt = time.Time{} + d.lastSectionAttempts = nil } // appendPaths appends every path the inode is reachable by, primary first. @@ -133,6 +157,7 @@ func NewInodeToPath(root util.FullPath, ttlSec int) *InodeToPath { path2inode: make(map[util.FullPath]uint64), dirStates: make(map[uint64]*dirState), dirPaths: make(map[util.FullPath]*dirState), + wasCachedDirs: make(map[util.FullPath]uint64), cacheMetaTtlSec: time.Second * time.Duration(ttlSec), } t.inode2path[1] = &InodeEntry{ @@ -318,11 +343,36 @@ func (i *InodeToPath) MarkChildrenCached(fullpath util.FullPath) { d.lastAccess = now d.lastRefresh = now d.lastRebuildAttempt = time.Time{} + d.lastSectionAttempts = nil + i.markWasCached(fullpath) if i.cacheMetaTtlSec > 0 { d.cachedExpiresTime = now.Add(i.cacheMetaTtlSec) } } +// maxWasCachedDirs bounds how many dropped listings a lookup may rebuild: +// past it the oldest record is evicted and that directory reads through per +// entry, the same end state as a never-cached one. +const maxWasCachedDirs = 16384 + +// markWasCached records a directory as once-cached, evicting the oldest +// record past maxWasCachedDirs. Callers hold the write lock. +func (i *InodeToPath) markWasCached(fullpath util.FullPath) { + if _, ok := i.wasCachedDirs[fullpath]; ok { + return + } + i.wasCachedGen++ + i.wasCachedDirs[fullpath] = i.wasCachedGen + i.wasCachedOrder = append(i.wasCachedOrder, wasCachedEntry{path: fullpath, gen: i.wasCachedGen}) + for len(i.wasCachedOrder) > maxWasCachedDirs { + oldest := i.wasCachedOrder[0] + i.wasCachedOrder = i.wasCachedOrder[1:] + if gen, ok := i.wasCachedDirs[oldest.path]; ok && gen == oldest.gen { + delete(i.wasCachedDirs, oldest.path) + } + } +} + func (i *InodeToPath) IsChildrenCached(fullpath util.FullPath) bool { i.RLock() defer i.RUnlock() @@ -336,19 +386,26 @@ func (i *InodeToPath) IsChildrenCached(fullpath util.FullPath) bool { return false } -// ShouldRebuildExpiredDir reports whether a directory was fully cached, its -// TTL has elapsed, and enough time has passed since the last failed rebuild -// attempt to retry. Distinct from IsChildrenCached returning false for a -// never-cached, invalidated, evicted, or read-through directory: those leave -// isChildrenCached clear, while a plain TTL expiry keeps it set. +// ShouldRebuildExpiredDir reports whether a lookup should re-list the +// directory into the cache, backing off for cooldown after a failed attempt. +// It covers a fully cached directory whose TTL elapsed — a plain expiry keeps +// isChildrenCached set, unlike invalidation or eviction — and a directory +// whose previously cached listing was dropped by invalidation, idle eviction, +// or a kernel Forget. A never-cached or read-through directory keeps reading +// through: caching one on a single lookup is how oversized listings get +// pinned, so only a directory that proved cacheable earns the rebuild. func (i *InodeToPath) ShouldRebuildExpiredDir(fullpath util.FullPath, cooldown time.Duration) bool { i.RLock() defer i.RUnlock() d := i.dirStateOf(fullpath) - if d == nil || !d.isChildrenCached { + if d == nil || d.readDirDirect { return false } - if d.cachedExpiresTime.IsZero() || time.Now().Before(d.cachedExpiresTime) { + if d.isChildrenCached { + if d.cachedExpiresTime.IsZero() || time.Now().Before(d.cachedExpiresTime) { + return false + } + } else if _, was := i.wasCachedDirs[fullpath]; !was { return false } return d.lastRebuildAttempt.IsZero() || time.Since(d.lastRebuildAttempt) >= cooldown @@ -364,6 +421,37 @@ func (i *InodeToPath) MarkRebuildAttempt(fullpath util.FullPath, now time.Time) } } +// ShouldRefreshStaleSection reports whether a lookup should re-list the +// stale section of a still-cached directory, backing off for cooldown after +// a failed attempt on that same section — a neighbouring range's failure +// does not stall this one. sectionKey is the section's lower bound. +func (i *InodeToPath) ShouldRefreshStaleSection(fullpath util.FullPath, sectionKey string, cooldown time.Duration) bool { + i.RLock() + defer i.RUnlock() + d := i.dirStateOf(fullpath) + if d == nil || d.readDirDirect || !d.isChildrenCached { + return false + } + last, ok := d.lastSectionAttempts[sectionKey] + return !ok || time.Since(last) >= cooldown +} + +// MarkSectionRefreshAttempt records that a refresh of the section keyed by +// sectionKey was attempted at now — the section-level analog of +// MarkRebuildAttempt. +func (i *InodeToPath) MarkSectionRefreshAttempt(fullpath util.FullPath, sectionKey string, now time.Time) { + i.Lock() + defer i.Unlock() + d := i.dirStateOf(fullpath) + if d == nil { + return + } + if d.lastSectionAttempts == nil { + d.lastSectionAttempts = make(map[string]time.Time) + } + d.lastSectionAttempts[sectionKey] = now +} + func (i *InodeToPath) HasInode(inode uint64) bool { if inode == 1 { return true @@ -447,6 +535,7 @@ func (i *InodeToPath) MarkDirectoryReadThrough(fullpath util.FullPath, now time. d.cachedExpiresTime = time.Time{} d.lastAccess = now d.lastRefresh = time.Time{} + delete(i.wasCachedDirs, fullpath) return true } @@ -531,6 +620,9 @@ func (i *InodeToPath) RemovePath(path util.FullPath, onStillReferenced func(inod onStillReferenced(inode) } } + // A deleted name must never justify a listing rebuild; the record is + // keyed by path, so drop it even when the inode was already forgotten. + delete(i.wasCachedDirs, path) } func (i *InodeToPath) removePathFromInode2Path(inode uint64, path util.FullPath) { @@ -560,6 +652,14 @@ func (i *InodeToPath) MovePath(sourcePath, targetPath util.FullPath) (sourceInod delete(i.path2inode, targetPath) i.dropDirPath(targetInode) } + // The rebuild eligibility travels with the directory: the target keeps it + // when the source earned it, and an overwritten target loses its own. + _, sourceWasCached := i.wasCachedDirs[sourcePath] + delete(i.wasCachedDirs, sourcePath) + delete(i.wasCachedDirs, targetPath) + if sourceWasCached { + i.markWasCached(targetPath) + } delete(i.path2inode, sourcePath) i.path2inode[targetPath] = sourceInode if entry, entryFound := i.inode2path[sourceInode]; entryFound { diff --git a/weed/mount/meta_cache/meta_cache.go b/weed/mount/meta_cache/meta_cache.go index 1b33d8417e..d5f89c7294 100644 --- a/weed/mount/meta_cache/meta_cache.go +++ b/weed/mount/meta_cache/meta_cache.go @@ -2,6 +2,7 @@ package meta_cache import ( "context" + "encoding/hex" "errors" "math" "os" @@ -176,6 +177,7 @@ func (mc *MetaCache) doInsertEntry(ctx context.Context, entry *filer.Entry, vers return err } mc.setEntryVersionLocked(ctx, entry.FullPath, versionTsNs) + mc.noteHardLinkVersionLocked(ctx, entry.HardLinkId, versionTsNs) return nil } @@ -228,6 +230,7 @@ func (mc *MetaCache) atomicUpdateEntryFromFilerLocked(ctx context.Context, oldPa return err } mc.setEntryVersionLocked(ctx, newEntry.FullPath, versionTsNs) + mc.noteHardLinkVersionLocked(ctx, newEntry.HardLinkId, versionTsNs) } } return nil @@ -355,13 +358,37 @@ func (mc *MetaCache) PurgeDirectoryChildren(dirPath util.FullPath, resetFn func( }) } -func (mc *MetaCache) UpdateEntry(ctx context.Context, entry *filer.Entry) error { +// UpdateEntry stores a local write against the entry. versionTsNs is the +// filer log position the write restates — a section refresh listed before it +// must not roll the content back — or zero when no position describes it, +// which marks the content unversioned. +func (mc *MetaCache) UpdateEntry(ctx context.Context, entry *filer.Entry, versionTsNs int64) error { mc.Lock() defer mc.Unlock() if err := mc.localStore.UpdateEntry(ctx, entry); err != nil { return err } - mc.markEntryUnversionedLocked(ctx, entry.FullPath) + mc.setEntryVersionLocked(ctx, entry.FullPath, versionTsNs) + mc.noteHardLinkVersionLocked(ctx, entry.HardLinkId, versionTsNs) + return nil +} + +// UpdateEntryChecked is UpdateEntry gated under the cache lock: when the +// path or its hard-link group already reflects a newer change — a +// subscription event or delete that raced the writer's read — the stale +// write is skipped instead of restoring old contents or a removed name. +func (mc *MetaCache) UpdateEntryChecked(ctx context.Context, entry *filer.Entry, versionTsNs int64) error { + mc.Lock() + defer mc.Unlock() + if mc.entryVersionBlocksLocked(ctx, entry.FullPath, versionTsNs, false) || + mc.hardLinkVersionFloorLocked(ctx, entry.HardLinkId) > versionTsNs { + return nil + } + if err := mc.localStore.UpdateEntry(ctx, entry); err != nil { + return err + } + mc.setEntryVersionLocked(ctx, entry.FullPath, versionTsNs) + mc.noteHardLinkVersionLocked(ctx, entry.HardLinkId, versionTsNs) return nil } @@ -429,6 +456,71 @@ func entryVersionChildPrefix(dirPath util.FullPath) []byte { return []byte(entryVersionKeyPrefix + string(dirPath) + "\x00") } +// entryVersionHardLinkPrefix namespaces per-group version records: every +// path of a hard-link group reads the same content blob, so the newest write +// position any sibling produced fences writes to them all — including +// cached siblings inodeToPath no longer tracks. +const entryVersionHardLinkPrefix = "\x00mount.entry.hlver\x00" + +func entryVersionHardLinkKey(hardLinkId []byte) []byte { + return []byte(entryVersionHardLinkPrefix + hex.EncodeToString(hardLinkId)) +} + +// hardLinkVersionFloorLocked returns the newest log position recorded for a +// hard-link group's shared content, or zero when none is known. +func (mc *MetaCache) hardLinkVersionFloorLocked(ctx context.Context, hardLinkId filer.HardLinkId) int64 { + if len(hardLinkId) == 0 { + return 0 + } + value, err := mc.localStore.KvGet(ctx, entryVersionHardLinkKey(hardLinkId)) + if err != nil || len(value) < 8 { + return 0 + } + return int64(util.BytesToUint64(value[:8])) +} + +// noteHardLinkVersionLocked raises a hard-link group's version floor to +// tsNs. It never lowers the floor, so a slow event cannot drag the fence +// backwards, and a write with no position claims nothing. +func (mc *MetaCache) noteHardLinkVersionLocked(ctx context.Context, hardLinkId filer.HardLinkId, tsNs int64) { + if len(hardLinkId) == 0 || tsNs <= mc.hardLinkVersionFloorLocked(ctx, hardLinkId) { + return + } + value := make([]byte, 8) + util.Uint64toBytes(value, uint64(tsNs)) + if err := mc.localStore.KvPut(ctx, entryVersionHardLinkKey(hardLinkId), value); err != nil { + glog.V(1).Infof("set hardlink version %x: %v", []byte(hardLinkId), err) + } +} + +// rebaseHardLinkEntryLocked returns the hard-link group's current cached +// content parked at the event's new path — a rename or delayed write whose +// shared blob is newer than the event must not roll that content back, but +// the name still lands. Returns nil when no cached sibling carries the +// group. +func (mc *MetaCache) rebaseHardLinkEntryLocked(ctx context.Context, newEntry *filer.Entry, oldPath util.FullPath) *filer.Entry { + for _, p := range []util.FullPath{newEntry.FullPath, oldPath} { + if p == "" { + continue + } + if donor, err := mc.localStore.FindEntry(ctx, p); err == nil && donor != nil && string(donor.HardLinkId) == string(newEntry.HardLinkId) { + rebased := *donor + rebased.FullPath = newEntry.FullPath + return &rebased + } + } + // No sibling path is cached: the shared blob itself holds the group's + // current content, so park it under the new name. + if blob, err := mc.localStore.KvGet(ctx, newEntry.HardLinkId); err == nil && len(blob) > 0 { + rebased := *newEntry + if err := rebased.DecodeAttributesAndChunks(blob); err == nil { + rebased.FullPath = newEntry.FullPath + return &rebased + } + } + return nil +} + // setEntryVersionLocked records the log position an entry write reflects. // Zero (an unversioned local write) clears the claim: the new content is not // proven at the old position. @@ -964,8 +1056,22 @@ func (mc *MetaCache) applyMetadataResponseLocked(ctx context.Context, resp *file if oldPath != "" && mc.entryVersionBlocksLocked(ctx, oldPath, resp.TsNs, true) { oldPath = "" } - if newEntry != nil && mc.entryVersionBlocksLocked(ctx, newEntry.FullPath, resp.TsNs, false) { - newEntry = nil + if newEntry != nil { + if mc.entryVersionBlocksLocked(ctx, newEntry.FullPath, resp.TsNs, false) { + newEntry = nil + } else if mc.hardLinkVersionFloorLocked(ctx, newEntry.HardLinkId) > resp.TsNs { + // The group's shared content is newer than this event — the + // stale copy must not overwrite it, but a rename's new name + // still lands: rebase the group's current content onto the + // new path when a cached sibling carries it. + droppedPath := newEntry.FullPath + newEntry = mc.rebaseHardLinkEntryLocked(ctx, newEntry, oldPath) + if newEntry == nil { + // No donor carries the group, so the name cannot land: + // stale its section and let the next lookup refresh it in. + mc.noteSectionChangeLocked(droppedPath, time.Now()) + } + } } } // Only foreign churn counts toward section invalidation: this mount's own diff --git a/weed/mount/meta_cache/meta_cache_apply_test.go b/weed/mount/meta_cache/meta_cache_apply_test.go index 13f688270b..cee1da0e23 100644 --- a/weed/mount/meta_cache/meta_cache_apply_test.go +++ b/weed/mount/meta_cache/meta_cache_apply_test.go @@ -573,7 +573,7 @@ func TestUnversionedWriteClearsEntryVersion(t *testing.T) { Mode: 0100644, FileSize: 22, }, - }); err != nil { + }, 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 { diff --git a/weed/mount/meta_cache/meta_cache_sections.go b/weed/mount/meta_cache/meta_cache_sections.go index a244027811..05781b1505 100644 --- a/weed/mount/meta_cache/meta_cache_sections.go +++ b/weed/mount/meta_cache/meta_cache_sections.go @@ -248,6 +248,42 @@ func EnsureListingFresh(ctx context.Context, mc *MetaCache, client filer_pb.File return nil } +// EnsureNameFresh re-validates only the section covering name. A lookup into +// an invalidated range re-establishes it here rather than every later lookup +// in that range reading through until the next listing. +func EnsureNameFresh(ctx context.Context, mc *MetaCache, client filer_pb.FilerClient, dirPath util.FullPath, name string) error { + r, stale := mc.staleRangeOf(dirPath, name) + if !stale { + return nil + } + ctx, cancel := context.WithTimeout(ctx, sectionRefreshTimeout) + defer cancel() + return mc.refreshSection(ctx, client, dirPath, r.lo, r.hi) +} + +// StaleRangeOf returns the lower bound of the section covering name when it +// needs re-validation — the key a caller may use to scope refresh backoff +// per section. +func StaleRangeOf(mc *MetaCache, dirPath util.FullPath, name string) (string, bool) { + r, stale := mc.staleRangeOf(dirPath, name) + return r.lo, stale +} + +func (mc *MetaCache) staleRangeOf(dirPath util.FullPath, name string) (nameRange, bool) { + mc.RLock() + defer mc.RUnlock() + sl := mc.dirSections[dirPath] + if sl == nil { + return nameRange{}, false + } + idx := sl.sectionOf(name) + if s := sl.sections[idx]; !s.stale || s.unverifiable { + return nameRange{}, false + } + lo, hi := sl.sectionRange(idx) + return nameRange{lo: lo, hi: hi}, true +} + func (mc *MetaCache) refreshSection(ctx context.Context, client filer_pb.FilerClient, dirPath util.FullPath, lo, hi string) error { _, err, _ := mc.visitGroup.Do(string(dirPath)+"\x00section\x00"+lo, func() (interface{}, error) { if !mc.rangeStale(dirPath, lo) { @@ -365,6 +401,25 @@ func (mc *MetaCache) applySectionRefreshNow(ctx context.Context, dirPath util.Fu if mc.entryVersionBlocksLocked(ctx, entry.FullPath, snapshotTsNs, false) { continue } + // A hard-linked entry shares one content blob with its siblings: a + // snapshot older than the group's newest write would roll that + // content back for every path reading through this entry. A cached + // member of the same group is skipped outright; any other live entry + // at the path is a stale occupant of a reused name, so the listed + // name still lands — carrying the group's parked content rather + // than the snapshot's. + staleOccupant := false + if mc.hardLinkVersionFloorLocked(ctx, entry.HardLinkId) > snapshotTsNs { + if existing, findErr := mc.localStore.FindEntry(ctx, entry.FullPath); findErr == nil && existing != nil { + if !isTtlExpired(existing) && string(existing.HardLinkId) == string(entry.HardLinkId) { + continue + } + staleOccupant = len(existing.HardLinkId) != 0 && string(existing.HardLinkId) != string(entry.HardLinkId) + } + if rebased := mc.rebaseHardLinkEntryLocked(ctx, entry, ""); rebased != nil { + entry = rebased + } + } // An unversioned marker would bypass the section floor, so it cannot // outlive the snapshot write that replaces its content — but pinned // local-only state stays authoritative and is not replaced at all. @@ -374,10 +429,20 @@ func (mc *MetaCache) applySectionRefreshNow(ctx context.Context, dirPath util.Fu continue } } + if staleOccupant { + // The name already left the occupant's group on the filer, and a + // later write to a sibling can have folded that unlink into the + // shared blob's counter — vacate the path like a rename so the + // insert does not decrement a group that may no longer count it. + if err := mc.localStore.DeleteEntry(context.WithValue(ctx, "OP", "MV"), entry.FullPath); err != nil { + return err + } + } // no per-entry version: the section floor set below covers the range if err := mc.localStore.InsertEntry(ctx, entry); err != nil { return err } + mc.noteHardLinkVersionLocked(ctx, entry.HardLinkId, snapshotTsNs) if unversioned { // only once the write landed: if it fails, the old content keeps // the marker, and with it the right to be corrected by any event diff --git a/weed/mount/meta_cache/meta_cache_sections_test.go b/weed/mount/meta_cache/meta_cache_sections_test.go index 6199951064..11d869a3ff 100644 --- a/weed/mount/meta_cache/meta_cache_sections_test.go +++ b/weed/mount/meta_cache/meta_cache_sections_test.go @@ -972,3 +972,280 @@ func TestSectionRefreshSweepsANameRecordedAtTheSnapshot(t *testing.T) { t.Error("a name the refresh did not list is still cached") } } + +// TestSectionRefreshParksHardLinkBlobContent checks that a stale snapshot +// install of an uncached hard-link sibling keeps the name but parks the +// group's shared blob content, rather than rolling every sibling back to +// the snapshot's content. +func TestSectionRefreshParksHardLinkBlobContent(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"}) + + hardLinkId := filer.HardLinkId([]byte{1, 2, 3, 4}) + const writeTsNs = 6000 + if err := mc.ApplyMetadataResponse(context.Background(), &filer_pb.SubscribeMetadataResponse{ + Directory: "/dir", + EventNotification: &filer_pb.EventNotification{ + NewEntry: &filer_pb.Entry{ + Name: "a", + HardLinkId: hardLinkId, + HardLinkCounter: 2, + Attributes: &filer_pb.FuseAttributes{Crtime: 1, Mtime: 9, FileMode: 0100644, FileSize: 99}, + }, + }, + TsNs: writeTsNs, + }, SubscriberMetadataResponseApplyOptions); err != nil { + t.Fatalf("seed sibling at %d: %v", writeTsNs, err) + } + + // The refresh's snapshot predates the write: sibling "b" is live on the + // filer but its listed content is older than the group's shared blob. + const snapshotTsNs = 5000 + 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"), + Attr: filer.Attr{Crtime: time.Unix(1, 0), Mtime: time.Unix(2, 0), Mode: 0100644, FileSize: 3}, + HardLinkId: hardLinkId, + HardLinkCounter: 2, + }, + }, + snapshotTsNs: snapshotTsNs, + }, + }); err != nil { + t.Fatalf("refresh: %v", err) + } + + b, _, err := mc.FindEntry(context.Background(), util.FullPath("/dir/b")) + if err != nil || b == nil { + t.Fatalf("the live link is missing after the refresh: %v", err) + } + if b.FileSize != 99 { + t.Errorf("sibling installed the snapshot's stale content: FileSize=%d, want 99", b.FileSize) + } + a, _, err := mc.FindEntry(context.Background(), util.FullPath("/dir/a")) + if err != nil || a == nil { + t.Fatalf("seeded sibling missing: %v", err) + } + if a.FileSize != 99 { + t.Errorf("the shared blob was rolled back: FileSize=%d, want 99", a.FileSize) + } +} + +// TestSectionRefreshReplacesReusedNameOccupant checks that a cached entry +// sharing only the listed name — a path since recreated as a hard-link +// member — is replaced by the group's parked content rather than kept, +// which would serve the stale occupant under a fresh section. +func TestSectionRefreshReplacesReusedNameOccupant(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"}) + + if err := mc.InsertEntry(context.Background(), &filer.Entry{ + FullPath: util.FullPath("/dir/b"), + Attr: filer.Attr{Crtime: time.Unix(1, 0), Mtime: time.Unix(1, 0), Mode: 0100644, FileSize: 7}, + }, 4000); err != nil { + t.Fatalf("seed occupant: %v", err) + } + + hardLinkId := filer.HardLinkId([]byte{1, 2, 3, 4}) + const writeTsNs = 6000 + if err := mc.ApplyMetadataResponse(context.Background(), &filer_pb.SubscribeMetadataResponse{ + Directory: "/dir", + EventNotification: &filer_pb.EventNotification{ + NewEntry: &filer_pb.Entry{ + Name: "a", + HardLinkId: hardLinkId, + HardLinkCounter: 2, + Attributes: &filer_pb.FuseAttributes{Crtime: 1, Mtime: 9, FileMode: 0100644, FileSize: 99}, + }, + }, + TsNs: writeTsNs, + }, SubscriberMetadataResponseApplyOptions); err != nil { + t.Fatalf("seed sibling at %d: %v", writeTsNs, err) + } + + const snapshotTsNs = 5000 + 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"), + Attr: filer.Attr{Crtime: time.Unix(1, 0), Mtime: time.Unix(2, 0), Mode: 0100644, FileSize: 3}, + HardLinkId: hardLinkId, + HardLinkCounter: 2, + }, + }, + snapshotTsNs: snapshotTsNs, + }, + }); err != nil { + t.Fatalf("refresh: %v", err) + } + + b, _, err := mc.FindEntry(context.Background(), util.FullPath("/dir/b")) + if err != nil || b == nil { + t.Fatalf("the live link is missing after the refresh: %v", err) + } + if string(b.HardLinkId) != string(hardLinkId) { + t.Errorf("the stale occupant was kept: HardLinkId=%x, want %x", b.HardLinkId, hardLinkId) + } + if b.FileSize != 99 { + t.Errorf("reused name did not get the parked content: FileSize=%d, want 99", b.FileSize) + } +} + +// TestSectionRefreshKeepsOccupantGroupCounter covers a reused name whose +// cached occupant's shared blob already reflects the filer-side unlink: a +// sibling write folded the counter down while the occupant's entry stayed +// cached. Replacing that occupant must not decrement the group again, or +// the counter reaches zero and the shared blob live siblings read is lost. +func TestSectionRefreshKeepsOccupantGroupCounter(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"}) + + oldGroup := filer.HardLinkId([]byte{5, 6, 7, 8}) + for _, name := range []string{"b", "q"} { + if err := mc.InsertEntry(context.Background(), &filer.Entry{ + FullPath: util.FullPath("/dir/" + name), + Attr: filer.Attr{Crtime: time.Unix(1, 0), Mtime: time.Unix(1, 0), Mode: 0100644, FileSize: 11}, + HardLinkId: oldGroup, + HardLinkCounter: 2, + }, 2000); err != nil { + t.Fatalf("seed %s: %v", name, err) + } + } + // A post-unlink write to the live sibling folds the occupant's removal + // into the shared counter while the occupant's entry is still cached. + if err := mc.InsertEntry(context.Background(), &filer.Entry{ + FullPath: util.FullPath("/dir/q"), + Attr: filer.Attr{Crtime: time.Unix(1, 0), Mtime: time.Unix(3, 0), Mode: 0100644, FileSize: 22}, + HardLinkId: oldGroup, + HardLinkCounter: 1, + }, 3000); err != nil { + t.Fatalf("reseed q: %v", err) + } + + hardLinkId := filer.HardLinkId([]byte{1, 2, 3, 4}) + const writeTsNs = 6000 + if err := mc.ApplyMetadataResponse(context.Background(), &filer_pb.SubscribeMetadataResponse{ + Directory: "/dir", + EventNotification: &filer_pb.EventNotification{ + NewEntry: &filer_pb.Entry{ + Name: "a", + HardLinkId: hardLinkId, + HardLinkCounter: 2, + Attributes: &filer_pb.FuseAttributes{Crtime: 1, Mtime: 9, FileMode: 0100644, FileSize: 99}, + }, + }, + TsNs: writeTsNs, + }, SubscriberMetadataResponseApplyOptions); err != nil { + t.Fatalf("seed sibling at %d: %v", writeTsNs, err) + } + + const snapshotTsNs = 5000 + 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"), + Attr: filer.Attr{Crtime: time.Unix(1, 0), Mtime: time.Unix(2, 0), Mode: 0100644, FileSize: 3}, + HardLinkId: hardLinkId, + HardLinkCounter: 2, + }, + }, + snapshotTsNs: snapshotTsNs, + }, + }); err != nil { + t.Fatalf("refresh: %v", err) + } + + b, _, err := mc.FindEntry(context.Background(), util.FullPath("/dir/b")) + if err != nil || b == nil { + t.Fatalf("the live link is missing after the refresh: %v", err) + } + if string(b.HardLinkId) != string(hardLinkId) { + t.Errorf("the stale occupant was kept: HardLinkId=%x, want %x", b.HardLinkId, hardLinkId) + } + + blob, err := mc.localStore.KvGet(context.Background(), oldGroup) + if err != nil { + t.Fatalf("the live sibling's shared blob was deleted: %v", err) + } + shared := &filer.Entry{} + if err := shared.DecodeAttributesAndChunks(blob); err != nil { + t.Fatalf("decode shared blob: %v", err) + } + if shared.HardLinkCounter != 1 { + t.Errorf("the occupant's old group was decremented again: counter=%d, want 1", shared.HardLinkCounter) + } + q, _, err := mc.FindEntry(context.Background(), util.FullPath("/dir/q")) + if err != nil || q == nil { + t.Fatalf("the live sibling is unreadable after the refresh: %v", err) + } + if q.FileSize != 22 { + t.Errorf("the live sibling's content was rolled back: FileSize=%d, want 22", q.FileSize) + } +} + +// TestEnsureNameFreshRefreshesOnlyItsSection checks that a name-scoped +// refresh re-lists just the section holding that name and leaves other stale +// sections untouched. +func TestEnsureNameFreshRefreshesOnlyItsSection(t *testing.T) { + mc, _, _, _ := newTestMetaCache(t, map[util.FullPath]bool{ + "/": true, + "/dir": true, + }) + defer mc.Shutdown() + + server := §ionFilerServer{snapshot: 5000} + server.names = append(server.names, "a-000", "b-000", "m", "z-000") + client := startSectionFilerServer(t, server) + + buildSectionedDir(t, mc, util.FullPath("/dir"), 1000, []string{"m"}) + applyChurn(t, mc, "/dir", "a-", sectionHotThreshold, 2000, SubscriberMetadataResponseApplyOptions) + applyChurn(t, mc, "/dir", "z-", sectionHotThreshold, 3000, SubscriberMetadataResponseApplyOptions) + if mc.IsNameFresh(util.FullPath("/dir/a-000")) || mc.IsNameFresh(util.FullPath("/dir/z-000")) { + t.Fatal("both sections should be stale before the refresh") + } + + if err := EnsureNameFresh(context.Background(), mc, client, util.FullPath("/dir"), "a-000"); err != nil { + t.Fatalf("refresh: %v", err) + } + if !mc.IsNameFresh(util.FullPath("/dir/a-000")) { + t.Fatal("the section holding the name should be fresh after the refresh") + } + if mc.IsNameFresh(util.FullPath("/dir/z-000")) { + t.Fatal("the other stale section must not be refreshed by a name outside it") + } + + requests := server.listRequests() + if len(requests) != 1 { + t.Fatalf("name refresh made %d list requests, want 1", len(requests)) + } + if requests[0].StartFromFileName != "" || requests[0].InclusiveStartFrom { + t.Fatalf("refresh should list the section from its start: StartFromFileName=%q inclusive=%v", + requests[0].StartFromFileName, requests[0].InclusiveStartFrom) + } + + if err := EnsureNameFresh(context.Background(), mc, client, util.FullPath("/dir"), "z-000"); err != nil { + t.Fatalf("second refresh: %v", err) + } + if !mc.IsNameFresh(util.FullPath("/dir/z-000")) { + t.Fatal("the second section should be fresh after its own refresh") + } +} diff --git a/weed/mount/weedfs.go b/weed/mount/weedfs.go index 1385017eed..1453bdd2b4 100644 --- a/weed/mount/weedfs.go +++ b/weed/mount/weedfs.go @@ -612,13 +612,15 @@ const expiredDirRebuildCooldown = 30 * time.Second // It also returns the log position the entry reflects: the entry's stored // version, the lookup response's, or zero if unknown. func (wfs *WFS) lookupEntry(ctx context.Context, fullpath util.FullPath) (*filer.Entry, entryVersion, fuse.Status) { - dir, _ := fullpath.DirAndName() + dir, name := fullpath.DirAndName() dirPath := util.FullPath(dir) // The kernel can serve a directory listing from its page cache past // cacheMetaTtlSec, so ReadDir never runs and EnsureVisited is not called. - // Rebuild the expired directory once here so the cache hit below serves - // metadata lookups instead of issuing one LookupEntry RPC per entry. + // Invalidation, idle eviction, and a kernel Forget also drop a cached + // listing without scheduling a rebuild. Rebuild here — once — so the + // cache hit below serves metadata lookups instead of issuing one + // LookupEntry RPC per entry. if !wfs.metaCache.IsDirectoryCached(dirPath) && wfs.inodeToPath.ShouldRebuildExpiredDir(dirPath, expiredDirRebuildCooldown) { // The rebuild lists the parent from the filer; let any pending async // flush of the target entry land first so the rebuilt cache does not @@ -639,30 +641,60 @@ func (wfs *WFS) lookupEntry(ctx context.Context, fullpath util.FullPath) (*filer } } - if wfs.metaCache.IsDirectoryCached(dirPath) && wfs.metaCache.IsNameFresh(fullpath) { - cachedEntry, cachedVersionTsNs, cacheErr := wfs.metaCache.FindEntry(ctx, fullpath) - if cacheErr != nil && cacheErr != filer_pb.ErrNotFound { - glog.Errorf("lookupEntry: cache lookup for %s failed: %v", fullpath, cacheErr) - return nil, entryVersion{}, fuse.EIO + if wfs.metaCache.IsDirectoryCached(dirPath) { + if !wfs.metaCache.IsNameFresh(fullpath) { + // A burst of remote changes invalidated this name's section, and a + // workload that never reads the directory would read through per + // entry until one did. Re-validate just that section here so the + // lookups after this one are served from the cache again. As with + // the rebuild above, let a pending flush land first so the section + // refresh cannot capture pre-flush metadata. A failing refresh + // backs off under the same cooldown the rebuild gate uses rather + // than re-listing the range on every lookup. + if sectionLo, stale := meta_cache.StaleRangeOf(wfs.metaCache, dirPath, name); stale && wfs.inodeToPath.ShouldRefreshStaleSection(dirPath, sectionLo, expiredDirRebuildCooldown) { + if inode, found := wfs.inodeToPath.GetInode(fullpath); found { + if !wfs.waitForPendingAsyncFlush(ctx, inode) { + return nil, entryVersion{}, fuse.EINTR + } + } + if err := meta_cache.EnsureNameFresh(ctx, wfs.metaCache, wfs, dirPath, name); err != nil { + if errors.Is(err, meta_cache.ErrRefreshRangeTooLarge) { + // The section outgrew one refresh: drop the cached + // listing so a later lookup rebuilds it re-tiled, the + // same way ReadDir falls back to read-through. + wfs.purgeDirectoryCache(dirPath) + } else { + wfs.inodeToPath.MarkSectionRefreshAttempt(dirPath, sectionLo, time.Now()) + glog.V(2).Infof("lookupEntry: refresh stale section of %s: %v", dirPath, err) + } + } + } } - if cachedEntry != nil { - glog.V(4).Infof("lookupEntry cache hit %s", fullpath) - // Store versions come from applied events, not an RPC fence. - return cachedEntry, entryVersion{tsNs: cachedVersionTsNs}, fuse.OK - } - // Re-check: the directory may have been evicted from cache between - // our IsDirectoryCached check and FindEntry. - // If it's no longer cached, fall through to the filer lookup below. - if wfs.metaCache.IsDirectoryCached(dirPath) && wfs.metaCache.IsNameFresh(fullpath) { - // Authoritative ENOENT only if inodeToPath also has no record. - // If the kernel still tracks this inode, the three layers - // disagree; trust the filer over the local cache (the - // filer-ErrNotFound branch below logs the confirmed drift). - if _, inodeFound := wfs.inodeToPath.GetInode(fullpath); !inodeFound { - glog.V(4).Infof("lookupEntry cache miss (dir cached) %s", fullpath) - return nil, entryVersion{}, fuse.ENOENT + if wfs.metaCache.IsNameFresh(fullpath) { + cachedEntry, cachedVersionTsNs, cacheErr := wfs.metaCache.FindEntry(ctx, fullpath) + if cacheErr != nil && cacheErr != filer_pb.ErrNotFound { + glog.Errorf("lookupEntry: cache lookup for %s failed: %v", fullpath, cacheErr) + return nil, entryVersion{}, fuse.EIO + } + if cachedEntry != nil { + glog.V(4).Infof("lookupEntry cache hit %s", fullpath) + // Store versions come from applied events, not an RPC fence. + return cachedEntry, entryVersion{tsNs: cachedVersionTsNs}, fuse.OK + } + // Re-check: the directory may have been evicted from cache between + // our IsDirectoryCached check and FindEntry. + // If it's no longer cached, fall through to the filer lookup below. + if wfs.metaCache.IsDirectoryCached(dirPath) && wfs.metaCache.IsNameFresh(fullpath) { + // Authoritative ENOENT only if inodeToPath also has no record. + // If the kernel still tracks this inode, the three layers + // disagree; trust the filer over the local cache (the + // filer-ErrNotFound branch below logs the confirmed drift). + if _, inodeFound := wfs.inodeToPath.GetInode(fullpath); !inodeFound { + glog.V(4).Infof("lookupEntry cache miss (dir cached) %s", fullpath) + return nil, entryVersion{}, fuse.ENOENT + } + glog.V(2).Infof("lookupEntry: %s missing from cache while parent %s is cached; inode tracked, consulting filer", fullpath, dirPath) } - glog.V(2).Infof("lookupEntry: %s missing from cache while parent %s is cached; inode tracked, consulting filer", fullpath, dirPath) } } diff --git a/weed/mount/weedfs_dir_lookup_cache_test.go b/weed/mount/weedfs_dir_lookup_cache_test.go index f1f3635cda..296884b73c 100644 --- a/weed/mount/weedfs_dir_lookup_cache_test.go +++ b/weed/mount/weedfs_dir_lookup_cache_test.go @@ -331,3 +331,184 @@ func advanceRebuildAttempt(t *testing.T, itp *InodeToPath, dir util.FullPath, d } state.lastRebuildAttempt = time.Now().Add(-d) } + +// TestLookupRebuildsInvalidatedDirectoryCache covers a dropped listing rather +// than an expired one: invalidation, idle eviction, or the purge on kernel +// Forget all clear the directory's cached flag, and without a rebuild path a +// lookup-only workload (e.g. git object reads) reads through per entry until +// a ReadDir happened to run. +func TestLookupRebuildsInvalidatedDirectoryCache(t *testing.T) { + wfs := newLookupCacheTestWFS(t, 60) + dir := util.FullPath("/dir") + wfs.inodeToPath.Lookup(dir, time.Now().Unix(), true, false, 0, true) + + fake := &lookupCacheTestFiler{ + entries: []*filer_pb.Entry{ + {Name: "a", Attributes: &filer_pb.FuseAttributes{FileSize: 1, FileMode: 0100644}}, + {Name: "b", Attributes: &filer_pb.FuseAttributes{FileSize: 2, FileMode: 0100644}}, + {Name: "c", Attributes: &filer_pb.FuseAttributes{FileSize: 3, FileMode: 0100644}}, + }, + snapshotTsNs: 5000, + } + startFakeFiler(t, wfs, fake) + + if err := meta_cache.EnsureVisited(context.Background(), wfs.metaCache, wfs, dir, 0); err != nil { + t.Fatalf("EnsureVisited: %v", err) + } + wfs.inodeToPath.InvalidateChildrenCache(dir) + if wfs.inodeToPath.IsChildrenCached(dir) { + t.Fatal("directory cache should be dropped") + } + + listCallsBefore := fake.listCalls.Load() + lookupCallsBefore := fake.lookupCalls.Load() + + for _, name := range []string{"a", "b", "c"} { + entry, _, status := wfs.lookupEntry(context.Background(), dir.Child(name)) + if status != fuse.OK { + t.Fatalf("lookupEntry %s: %v", name, status) + } + if entry == nil || entry.Name() != name { + t.Fatalf("lookupEntry %s: got %v", name, entry) + } + } + + if got := fake.listCalls.Load() - listCallsBefore; got != 1 { + t.Errorf("ListEntries RPCs = %d, want 1 (dropped dir cache should be rebuilt once)", got) + } + if got := fake.lookupCalls.Load() - lookupCallsBefore; got != 0 { + t.Errorf("LookupDirectoryEntry RPCs = %d, want 0", got) + } +} + +// TestLookupNeverCachedDirReadsThrough guards the other side: a directory +// that never held a cached listing must not be rebuilt on a single lookup, +// or one stat could pin an arbitrarily large listing. +func TestLookupNeverCachedDirReadsThrough(t *testing.T) { + wfs := newLookupCacheTestWFS(t, 60) + dir := util.FullPath("/dir") + wfs.inodeToPath.Lookup(dir, time.Now().Unix(), true, false, 0, true) + + fake := &lookupCacheTestFiler{snapshotTsNs: 5000} + startFakeFiler(t, wfs, fake) + + if _, _, status := wfs.lookupEntry(context.Background(), dir.Child("a")); status != fuse.OK { + t.Fatalf("lookupEntry a: %v", status) + } + if got := fake.listCalls.Load(); got != 0 { + t.Errorf("ListEntries RPCs = %d, want 0 (never-cached dir stays read-through)", got) + } + if got := fake.lookupCalls.Load(); got != 1 { + t.Errorf("LookupDirectoryEntry RPCs = %d, want 1", got) + } +} + +// TestForgetKeepsDirRebuildable exercises the kernel Forget path: the dropped +// dirState must not lose the record that the directory once held a listing, +// or a re-looked-up directory would read through forever. +func TestForgetKeepsDirRebuildable(t *testing.T) { + itp := NewInodeToPath("/", 60) + dir := util.FullPath("/dir") + + inode := itp.Lookup(dir, time.Now().Unix(), true, false, 0, true) + itp.MarkChildrenCached(dir) + itp.Forget(inode, 1, nil, func(util.FullPath) {}) + if itp.dirStateOf(dir) != nil { + t.Fatal("Forget should drop the dirState") + } + + if itp.ShouldRebuildExpiredDir(dir, expiredDirRebuildCooldown) { + t.Fatal("a forgotten directory has no state to mark cached; lookup must re-track it first") + } + + itp.Lookup(dir, time.Now().Unix(), true, false, 0, true) + if !itp.ShouldRebuildExpiredDir(dir, expiredDirRebuildCooldown) { + t.Fatal("re-looked-up directory that once held a listing should rebuild") + } +} + +// TestReusedPathKeepsRebuildRecord covers a directory deleted and re-cached +// while its first queue slot is still pending: eviction reaching the stale +// slot must not delete the newer record, or the directory reads through per +// lookup until the next listing. +func TestReusedPathKeepsRebuildRecord(t *testing.T) { + itp := NewInodeToPath("/", 60) + dir := util.FullPath("/hot") + now := time.Now().Unix() + + itp.Lookup(dir, now, true, false, 0, true) + itp.MarkChildrenCached(dir) + itp.RemovePath(dir, nil) + + inode := itp.Lookup(dir, now+1, true, false, 0, true) + itp.MarkChildrenCached(dir) + + for n := 0; n < maxWasCachedDirs-1; n++ { + itp.markWasCached(util.FullPath(fmt.Sprintf("/filler%d", n))) + } + + itp.Forget(inode, 1, nil, nil) + itp.Lookup(dir, now+2, true, false, 0, true) + if !itp.ShouldRebuildExpiredDir(dir, expiredDirRebuildCooldown) { + t.Fatal("re-cached directory lost its rebuild record to a stale queue entry") + } +} + +// TestLookupRefreshesStaleSection covers a directory still cached but with an +// invalidated section: without a refresh on lookup, every lookup in that +// range reads through until a listing re-validates it. +func TestLookupRefreshesStaleSection(t *testing.T) { + wfs := newLookupCacheTestWFS(t, 60) + dir := util.FullPath("/dir") + wfs.inodeToPath.Lookup(dir, time.Now().Unix(), true, false, 0, true) + + fake := &lookupCacheTestFiler{ + entries: []*filer_pb.Entry{ + {Name: "a", Attributes: &filer_pb.FuseAttributes{FileSize: 1, FileMode: 0100644}}, + {Name: "b", Attributes: &filer_pb.FuseAttributes{FileSize: 2, FileMode: 0100644}}, + {Name: "c", Attributes: &filer_pb.FuseAttributes{FileSize: 3, FileMode: 0100644}}, + }, + snapshotTsNs: 5000, + } + startFakeFiler(t, wfs, fake) + + if err := meta_cache.EnsureVisited(context.Background(), wfs.metaCache, wfs, dir, 0); err != nil { + t.Fatalf("EnsureVisited: %v", err) + } + + // A burst of remote changes in the only section invalidates it. + for i := 0; i < 64; i++ { + resp := &filer_pb.SubscribeMetadataResponse{ + Directory: string(dir), + TsNs: 6000 + int64(i), + EventNotification: &filer_pb.EventNotification{ + NewEntry: &filer_pb.Entry{ + Name: fmt.Sprintf("x-%03d", i), + Attributes: &filer_pb.FuseAttributes{Crtime: 1, Mtime: 1, FileMode: 0100644, FileSize: 1}, + }, + }, + } + if err := wfs.metaCache.ApplyMetadataResponse(context.Background(), resp, meta_cache.SubscriberMetadataResponseApplyOptions); err != nil { + t.Fatalf("apply churn %d: %v", i, err) + } + } + if wfs.metaCache.IsNameFresh(dir.Child("a")) { + t.Fatal("section should be stale after the change burst") + } + + listCallsBefore := fake.listCalls.Load() + lookupCallsBefore := fake.lookupCalls.Load() + + for _, name := range []string{"a", "b", "c"} { + if entry, _, status := wfs.lookupEntry(context.Background(), dir.Child(name)); status != fuse.OK || entry == nil { + t.Fatalf("lookupEntry %s: status=%v entry=%v", name, status, entry) + } + } + + if got := fake.lookupCalls.Load() - lookupCallsBefore; got != 0 { + t.Errorf("LookupDirectoryEntry RPCs = %d, want 0 (stale section should be refreshed, not read per-entry)", got) + } + if got := fake.listCalls.Load() - listCallsBefore; got != 1 { + t.Errorf("ListEntries RPCs = %d, want 1 (one section refresh)", got) + } +} diff --git a/weed/mount/weedfs_file_mkrm.go b/weed/mount/weedfs_file_mkrm.go index f982f9a2fe..cd54eec281 100644 --- a/weed/mount/weedfs_file_mkrm.go +++ b/weed/mount/weedfs_file_mkrm.go @@ -338,7 +338,7 @@ func (wfs *WFS) Unlink(cancel <-chan struct{}, header *fuse.InHeader, name strin now := time.Now() decremented.Attributes.Ctime = now.Unix() decremented.Attributes.CtimeNs = int32(now.Nanosecond()) - wfs.syncHardLinkSiblings(sharedInode, decremented, entryFullPath) + wfs.syncHardLinkSiblings(ctx, sharedInode, decremented, event.GetTsNs(), entryFullPath) } return fuse.OK diff --git a/weed/mount/weedfs_link.go b/weed/mount/weedfs_link.go index 1797627ca4..c13dfcf9bf 100644 --- a/weed/mount/weedfs_link.go +++ b/weed/mount/weedfs_link.go @@ -144,12 +144,16 @@ func (wfs *WFS) Link(cancel <-chan struct{}, in *fuse.LinkIn, name string, out * updateEvent := updateResp.GetMetadataEvent() if updateEvent == nil { updateEvent = metadataUpdateEvent(oldParentPath, updateOldEntryRequest.Entry) + if updateEvent != nil { + updateEvent.TsNs = ackVersionTsNs(updateResp) + } } if applyErr := wfs.applyLocalMetadataEvent(context.Background(), updateEvent); applyErr != nil { glog.Warningf("link %s: best-effort metadata apply failed: %v", oldEntryPath, applyErr) wfs.inodeToPath.InvalidateChildrenCache(util.FullPath(oldParentPath)) } } + var linkVersionTsNs int64 if err == nil { var createResp *filer_pb.CreateEntryResponse createResp, err = wfs.streamCreateEntry(context.Background(), request) @@ -166,9 +170,13 @@ func (wfs *WFS) Link(cancel <-chan struct{}, in *fuse.LinkIn, name string, out * glog.Warningf("link rollback %s: %v", oldEntryPath, rollbackErr) } } else { + linkVersionTsNs = ackVersionTsNs(createResp) createEvent := createResp.GetMetadataEvent() if createEvent == nil { createEvent = metadataCreateEvent(string(newParentPath), request.Entry) + if createEvent != nil { + createEvent.TsNs = linkVersionTsNs + } } if applyErr := wfs.applyLocalMetadataEvent(context.Background(), createEvent); applyErr != nil { glog.Warningf("link %s: best-effort metadata apply failed: %v", newParentPath.Child(name), applyErr) @@ -195,19 +203,22 @@ func (wfs *WFS) Link(cancel <-chan struct{}, in *fuse.LinkIn, name string, out * // existing sibling link (other than the source we just wrote) returns // the old nlink from the local metacache — pjdfstest link/00.t catches // this after `link n1 n2` when it stats n0. - wfs.syncHardLinkSiblings(sourceInode, oldEntry, oldEntryPath, newEntryPath) + wfs.syncHardLinkSiblings(ctx, sourceInode, oldEntry, linkVersionTsNs, oldEntryPath, newEntryPath) wfs.outputPbEntry(out, sourceInode, request.Entry) return fuse.OK } -// syncHardLinkSiblings rewrites the cached HardLinkCounter (and ctime) on -// every sibling link of the given inode, and invalidates the kernel's inode -// attr cache. `authoritativeEntry` carries the counter/ctime that should be -// propagated. `skipPaths` are the link paths already updated by the caller -// (typically the source and/or the newly created/removed link). -func (wfs *WFS) syncHardLinkSiblings(inode uint64, authoritativeEntry *filer_pb.Entry, skipPaths ...util.FullPath) { +// syncHardLinkSiblings refreshes every sibling link of the given inode from +// the filer, so their cached HardLinkCounter (and any other fields) reflect +// the mutation. `authoritativeEntry` identifies the hard-link group; +// `versionTsNs` is the log position the mutation committed at — the fetch +// runs after the commit, so the fetched state already covers it, and the +// recorded position is the higher of the two. `skipPaths` are the link paths +// already updated by the caller (typically the source and/or the newly +// created/removed link). +func (wfs *WFS) syncHardLinkSiblings(ctx context.Context, inode uint64, authoritativeEntry *filer_pb.Entry, versionTsNs int64, skipPaths ...util.FullPath) { if authoritativeEntry == nil || len(authoritativeEntry.HardLinkId) == 0 { return } @@ -219,20 +230,48 @@ func (wfs *WFS) syncHardLinkSiblings(inode uint64, authoritativeEntry *filer_pb. for _, p := range skipPaths { skip[p] = struct{}{} } - ctx := context.Background() + // One overall deadline for the sequential sibling lookups: they run + // under the caller's exclusive hard-link lock, so a slow filer must not + // stall every other link operation on this group. + ctx, cancel := context.WithTimeout(ctx, metadataFlushTimeout) + defer cancel() for _, p := range paths { if _, skipped := skip[p]; skipped { continue } - sibling, _, err := wfs.metaCache.FindEntry(ctx, p) + sibling, siblingTsNs, err := wfs.metaCache.FindEntry(ctx, p) if err != nil || sibling == nil { continue } - // Only touch siblings that genuinely share the same hard-link id. - // inodeToPath's shared-inode invariant should already guarantee - // this, but a mismatch can occur transiently (e.g. a rename - // replaced one of the paths), and blindly stamping an unrelated - // entry's counter would corrupt it. + // Refresh the sibling from the filer rather than stamping the cached + // copy: the cache may predate an in-flight event of its own, and + // claiming this mutation's version for that stale content would + // fence the pending event out at the version gate, leaving old + // chunks and attributes until a refresh. + pbEntry, entryTsNs, _, fetchErr := filer_pb.GetEntry(ctx, wfs, p) + if fetchErr == nil { + // A path the filer no longer knows is on its way out — leave the + // cache alone. Only touch entries that genuinely share the same + // hard-link id: inodeToPath's shared-inode invariant should + // already guarantee this, but a mismatch can occur transiently + // (e.g. a rename replaced one of the paths). + if pbEntry == nil || !bytes.Equal(pbEntry.HardLinkId, authoritativeEntry.HardLinkId) { + continue + } + if entryTsNs < versionTsNs { + entryTsNs = versionTsNs + } + dir, _ := p.DirAndName() + if err := wfs.metaCache.UpdateEntryChecked(ctx, filer.FromPbEntry(dir, pbEntry), entryTsNs); err != nil { + glog.V(4).Infof("syncHardLinkSiblings update %s: %v", p, err) + } + continue + } + // The fetch failed: patch the counter in place so the sibling is + // not left stale, but claim only the version its content already + // holds — a newer in-flight event still lands and corrects it, + // while a replayed older delete stays fenced instead of removing + // a live link. if !bytes.Equal(sibling.HardLinkId, authoritativeEntry.HardLinkId) { continue } @@ -240,7 +279,7 @@ func (wfs *WFS) syncHardLinkSiblings(inode uint64, authoritativeEntry *filer_pb. if authoritativeEntry.Attributes != nil { sibling.Attr.Ctime = time.Unix(authoritativeEntry.Attributes.Ctime, int64(authoritativeEntry.Attributes.CtimeNs)) } - if err := wfs.metaCache.UpdateEntry(ctx, sibling); err != nil { + if err := wfs.metaCache.UpdateEntry(ctx, sibling, siblingTsNs); err != nil { glog.V(4).Infof("syncHardLinkSiblings update %s: %v", p, err) } } diff --git a/weed/mount/weedfs_link_test.go b/weed/mount/weedfs_link_test.go index e81bb59576..ef8b3fcb26 100644 --- a/weed/mount/weedfs_link_test.go +++ b/weed/mount/weedfs_link_test.go @@ -3,9 +3,11 @@ package mount import ( "context" "testing" + "time" "github.com/seaweedfs/go-fuse/v2/fuse" "github.com/seaweedfs/seaweedfs/weed/filer" + "github.com/seaweedfs/seaweedfs/weed/mount/meta_cache" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/util" ) @@ -239,7 +241,7 @@ func TestLinkSyncsHardLinkSiblingsOnThirdLink(t *testing.T) { } authoritative.HardLinkCounter = 4 - wfs.syncHardLinkSiblings(authoritative.Attributes.Inode, authoritative, skippedSource, secondLinkPath) + wfs.syncHardLinkSiblings(context.Background(), authoritative.Attributes.Inode, authoritative, 0, skippedSource, secondLinkPath) for _, p := range siblings { if counter := cachedLink(t, wfs, p).HardLinkCounter; counter != 3 { t.Errorf("keyed by the persisted inode the sync reached %s (counter %d), which it cannot: "+ @@ -247,10 +249,83 @@ func TestLinkSyncsHardLinkSiblingsOnThirdLink(t *testing.T) { } } - wfs.syncHardLinkSiblings(sourceInode, authoritative, skippedSource, secondLinkPath) + wfs.syncHardLinkSiblings(context.Background(), sourceInode, authoritative, 0, skippedSource, secondLinkPath) for _, p := range siblings { if counter := cachedLink(t, wfs, p).HardLinkCounter; counter != 4 { t.Errorf("%s HardLinkCounter = %d after the sibling sync, want 4", p, counter) } } } + +// When the sibling refresh cannot reach the filer, the fallback patches the +// cached counter without proving the content reflects the mutation — so it +// keeps the version the sibling already held: claiming the mutation's +// position would fence an event still in flight for the sibling, while +// claiming nothing would let a replayed older delete remove the live link. +// The test filer answers every LookupDirectoryEntry with Unimplemented, so +// the sync always lands on that fallback. +func TestSyncHardLinkSiblingFetchFailureKeepsVersion(t *testing.T) { + wfs, _ := newCreateTestWFS(t) + + hardLinkId := filer.HardLinkId([]byte{9, 9, 9, 9}) + sibPath := util.FullPath("/sib.txt") + if err := wfs.metaCache.InsertEntry(context.Background(), &filer.Entry{ + FullPath: sibPath, + Attr: filer.Attr{Crtime: time.Unix(1, 0), Mtime: time.Unix(1, 0), Mode: 0o644, FileSize: 12, Uid: 99, Gid: 100}, + HardLinkId: hardLinkId, + HardLinkCounter: 2, + }, 7000); err != nil { + t.Fatalf("seed sibling: %v", err) + } + inode := wfs.inodeToPath.Lookup(sibPath, 1, false, false, 77, true) + if inode == 0 { + t.Fatal("Lookup handed out node id 0") + } + + authoritative := &filer_pb.Entry{ + HardLinkId: hardLinkId, + HardLinkCounter: 4, + Attributes: &filer_pb.FuseAttributes{Ctime: 8}, + } + wfs.syncHardLinkSiblings(context.Background(), inode, authoritative, 9000) + + if counter := cachedLink(t, wfs, sibPath).HardLinkCounter; counter != 4 { + t.Fatalf("sibling HardLinkCounter = %d after the fallback patch, want 4", counter) + } + + // The patch claimed only the version the sibling already held (7000), so + // a delayed delete replayed from before that point is fenced out instead + // of removing the live link. + if err := wfs.metaCache.ApplyMetadataResponse(context.Background(), &filer_pb.SubscribeMetadataResponse{ + Directory: "/", + EventNotification: &filer_pb.EventNotification{ + OldEntry: &filer_pb.Entry{Name: "sib.txt"}, + }, + TsNs: 6000, + }, meta_cache.SubscriberMetadataResponseApplyOptions); err != nil { + t.Fatalf("apply delayed delete: %v", err) + } + if sib := cachedLink(t, wfs, sibPath); sib == nil { + t.Fatal("replayed older delete removed the live link") + } + + // The mutation committed at 9000 but the patch proves nothing about the + // sibling's content: an event at 8000 still in flight must land. + if err := wfs.metaCache.ApplyMetadataResponse(context.Background(), &filer_pb.SubscribeMetadataResponse{ + Directory: "/", + EventNotification: &filer_pb.EventNotification{ + NewEntry: &filer_pb.Entry{ + Name: "sib.txt", + HardLinkId: hardLinkId, + HardLinkCounter: 2, + Attributes: &filer_pb.FuseAttributes{Crtime: 1, Mtime: 5, FileMode: 0o644, FileSize: 99, Uid: 99, Gid: 100}, + }, + }, + TsNs: 8000, + }, meta_cache.SubscriberMetadataResponseApplyOptions); err != nil { + t.Fatalf("apply pending event: %v", err) + } + if sib := cachedLink(t, wfs, sibPath); sib.FileSize != 99 { + t.Errorf("pending event was fenced by the patched version: FileSize = %d, want 99", sib.FileSize) + } +}