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