diff --git a/weed/mount/attribute_id_intent_test.go b/weed/mount/attribute_id_intent_test.go new file mode 100644 index 0000000000..de5475e95e --- /dev/null +++ b/weed/mount/attribute_id_intent_test.go @@ -0,0 +1,69 @@ +//go:build !freebsd + +package mount + +import ( + "context" + "path/filepath" + "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" + "google.golang.org/protobuf/proto" +) + +func TestAttributeExplicitCollidingIds(t *testing.T) { + for _, valid := range []uint32{fuse.FATTR_UID, fuse.FATTR_GID, fuse.FATTR_UID | fuse.FATTR_GID, fuse.FATTR_MODE} { + t.Run(map[uint32]string{fuse.FATTR_UID: "uid", fuse.FATTR_GID: "gid", fuse.FATTR_UID | fuse.FATTR_GID: "both", fuse.FATTR_MODE: "untouched"}[valid], func(t *testing.T) { + wfs := newInvalidateTestWFS(t) + mapper, err := meta_cache.NewUidGidMapper("1000:2000", "20:30") + if err != nil { + t.Fatal(err) + } + wfs.option.UidGidMapper = mapper + wfs.metaCache = meta_cache.NewMetaCache(filepath.Join(t.TempDir(), "mapped"), mapper, util.FullPath("/"), false, func(p util.FullPath) { wfs.inodeToPath.MarkChildrenCached(p) }, func(p util.FullPath) bool { return wfs.inodeToPath.IsChildrenCached(p) }, wfs.onEntryInvalidation, nil) + t.Cleanup(wfs.metaCache.Shutdown) + wfs.atimeMap = make(map[uint64]time.Time) + current := attributeAppendEntry(3) + current.Attributes.Uid = 1000 + current.Attributes.Gid = 20 + fake := &attributeCASFiler{entry: proto.Clone(current).(*filer_pb.Entry)} + startFakeFiler(t, wfs, fake) + path := util.FullPath("/dir/file") + wfs.inodeToPath.Lookup(util.FullPath("/dir"), 1, true, false, 0, false) + wfs.inodeToPath.Lookup(path, 1, false, false, 42, true) + wfs.inodeToPath.MarkChildrenCached(util.FullPath("/dir")) + if err := wfs.metaCache.InsertEntry(context.Background(), filer.FromPbEntry("/dir", current), 1); err != nil { + t.Fatal(err) + } + in := &fuse.SetAttrIn{} + in.NodeId = 42 + in.Valid = valid + in.Uid = 1000 + in.Gid = 20 + in.Mode = 0600 + if code := wfs.SetAttr(nil, in, &fuse.AttrOut{}); code != fuse.OK { + t.Fatal(code) + } + uid, gid := uint32(1000), uint32(20) + if valid&fuse.FATTR_UID != 0 { + uid = 2000 + } + if valid&fuse.FATTR_GID != 0 { + gid = 30 + } + fake.mu.Lock() + defer fake.mu.Unlock() + if fake.entry.Attributes.Uid != uid || fake.entry.Attributes.Gid != gid { + t.Fatalf("acknowledged ids=(%d,%d), want (%d,%d) for explicitly supplied fields %x", fake.entry.Attributes.Uid, fake.entry.Attributes.Gid, uid, gid, valid) + } + if !proto.Equal(fake.entry.Chunks[0], current.Chunks[0]) { + t.Fatal("attribute operation changed original chunk") + } + }) + } +} diff --git a/weed/mount/attribute_read_lock_order_test.go b/weed/mount/attribute_read_lock_order_test.go new file mode 100644 index 0000000000..817f1c132c --- /dev/null +++ b/weed/mount/attribute_read_lock_order_test.go @@ -0,0 +1,296 @@ +//go:build !freebsd + +package mount + +import ( + "context" + "fmt" + "os" + "os/exec" + "runtime" + "strings" + "sync" + "testing" + "time" + + "github.com/seaweedfs/go-fuse/v2/fuse" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/util" + "google.golang.org/protobuf/proto" +) + +// A subprocess bounds the deliberately deadlocked negative without leaving +// blocked goroutines/cache cleanup behind in the rest of the package suite. +func TestAttributeSetAttrReadLockOrder(t *testing.T) { + if os.Getenv("MOUNT_ATTR_READ_LOCK_CHILD") == "1" { + attributeReadLockScenario(t) + return + } + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + cmd := exec.CommandContext(ctx, os.Args[0], "-test.run=^TestAttributeSetAttrReadLockOrder$", "-test.count=1", "-test.timeout=12s") + cmd.Env = append(os.Environ(), "MOUNT_ATTR_READ_LOCK_CHILD=1") + output, err := cmd.CombinedOutput() + if ctx.Err() != nil { + t.Fatalf("lock-order control failed to finish its bounded child: %v\n%s", ctx.Err(), output) + } + if err != nil { + t.Fatalf("concurrent SetAttr/Read did not complete: %v\n%s", err, output) + } +} + +func attributeReadLockScenario(t *testing.T) { + w := newInvalidateTestWFS(t) + w.atimeMap = make(map[uint64]time.Time) + data := []byte("acknowledged payload") + current := attributeAppendEntry(0) + current.Content = data + current.Attributes.FileSize = uint64(len(data)) + fake := &attributeCASFiler{entry: proto.Clone(current).(*filer_pb.Entry)} + committed, resume := make(chan struct{}), make(chan struct{}) + var resumeOnce sync.Once + t.Cleanup(func() { resumeOnce.Do(func() { close(resume) }) }) + fake.afterCommit = func() { close(committed); <-resume } + startFakeFiler(t, w, fake) + path := util.FullPath("/dir/file") + w.inodeToPath.Lookup(path, 1, false, false, 42, true) + local := proto.Clone(current).(*filer_pb.Entry) + w.mapPbIdFromFilerToLocal(local) + fh := newFileHandle(w, 1, 42, local) + fh.RememberPath(path) + w.fhMap.inode2fh[42], w.fhMap.fh2inode[1] = fh, 42 + t.Cleanup(fh.dirtyPages.Destroy) + attrDone := make(chan fuse.Status, 1) + go func() { + in := &fuse.SetAttrIn{} + in.NodeId = 42 + in.Valid = fuse.FATTR_MODE + in.Mode = 0600 + attrDone <- w.SetAttr(nil, in, &fuse.AttrOut{}) + }() + select { + case <-committed: + case <-time.After(5 * time.Second): + t.Fatal("SetAttr did not reach owner commit hook") + } + // Model the background WRITE which makes retention necessary after the owner + // has committed. No unprotected flag write overlaps a running handler. + active := w.fhLockTable.AcquireLock("test background write", fh.fh, util.ExclusiveLock) + fh.dirtyMetadata = true + w.fhLockTable.ReleaseLock(fh.fh, active) + // Hold a reader on entryLock until retention is demonstrably waiting for its + // writer lock. The old path has no active lock; the fix already owns it. + fh.entryLock.RLock() + resumeOnce.Do(func() { close(resume) }) + waitAttributeReadFrame(t, "(*WFS).retainAcknowledgedAttributes", "sync.(*RWMutex).Lock") + readDone := make(chan fuse.Status, 1) + go func() { + buf := make([]byte, len(data)) + result, status := w.Read(nil, &fuse.ReadIn{InHeader: fuse.InHeader{NodeId: 42}, Fh: 1, Size: uint32(len(data))}, buf) + if status == fuse.OK { + got, code := result.Bytes(buf) + if code != fuse.OK || string(got) != string(data) { + status = fuse.EIO + } + result.Done() + } + readDone <- status + }() + // Prove the real Read has arrived before opening the gate: old code waits + // on entryLock while holding the table entry, fixed code waits on the table. + waitAttributeReadFrame(t, "(*WFS).Read", "(*FileHandle).readFromChunksWithContext", "AcquireLock") + fh.entryLock.RUnlock() + timer := time.NewTimer(2 * time.Second) + defer timer.Stop() + for _, done := range []chan fuse.Status{attrDone, readDone} { + select { + case status := <-done: + if status != fuse.OK { + t.Fatalf("handler returned %v", status) + } + case <-timer.C: + fmt.Fprintln(os.Stderr, "LOCK_ORDER_CYCLE: owner-committed SetAttr retention and same-handle Read cannot finish") + os.Exit(23) + } + } + fh.entry.RLock() + mode := fh.entry.Attributes.FileMode + fh.entry.RUnlock() + if mode&0777 != 0600 { + t.Fatalf("acknowledged chmod was not retained: %o", mode) + } +} + +// The attribute snapshot must queue for the handle lock before it takes +// entryLock. Otherwise a Read holding the shared lock waits on entryLock, the +// snapshot waits behind a queued exclusive (an invalidation or Write), and +// that waits on the Read. +func TestAttributeSnapshotReadLockOrder(t *testing.T) { + w := newInvalidateTestWFS(t) + entry := attributeAppendEntry(0) + fh := newFileHandle(w, 1, 42, entry) + t.Cleanup(fh.dirtyPages.Destroy) + + read := w.fhLockTable.AcquireLock("test read", fh.fh, util.SharedLock) + exclusiveDone := make(chan struct{}) + go func() { + lock := w.fhLockTable.AcquireLock("test invalidate", fh.fh, util.ExclusiveLock) + w.fhLockTable.ReleaseLock(fh.fh, lock) + close(exclusiveDone) + }() + waitAttributeReadFrame(t, "TestAttributeSnapshotReadLockOrder.func1", "sync.(*Cond).Wait") + snapshotDone := make(chan bool, 1) + go func() { + _, ok := w.attributeEntrySnapshot(fh, entry) + snapshotDone <- ok + }() + waitAttributeReadFrame(t, "(*WFS).attributeEntrySnapshot", "sync.(*Cond).Wait") + + readerCanProceed := fh.entryLock.TryRLock() + if readerCanProceed { + fh.entryLock.RUnlock() + } + w.fhLockTable.ReleaseLock(fh.fh, read) + <-exclusiveDone + if ok := <-snapshotDone; !ok { + t.Fatal("clean handle snapshot was refused") + } + if !readerCanProceed { + t.Fatal("snapshot holds entryLock while queued for the handle lock behind an exclusive waiter") + } +} + +// Stack frames are arrival/blocked witnesses, not an elapsed-time performance +// assertion. Channel/lock gates define the order; timeout only refuses a broken +// test arrangement. No sleep is used to make either handler win a race. +func waitAttributeReadFrame(t *testing.T, caller string, blocked ...string) { + t.Helper() + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + buf := make([]byte, 1<<20) + n := runtime.Stack(buf, true) + for _, stack := range strings.Split(string(buf[:n]), "\n\n") { + if !strings.Contains(stack, caller) { + continue + } + for _, frame := range blocked { + if strings.Contains(stack, frame) { + return + } + } + } + runtime.Gosched() + } + t.Fatalf("did not observe %s blocked at %v", caller, blocked) +} + +// FATTR_SIZE bypasses the attribute publication fence but must not reverse +// the handle locks while an acknowledged chmod is retaining its local delta. +func TestAttributeTruncateRetainLockOrder(t *testing.T) { + if os.Getenv("MOUNT_ATTR_TRUNCATE_LOCK_CHILD") == "1" { + attributeTruncateLockScenario(t) + return + } + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + cmd := exec.CommandContext(ctx, os.Args[0], "-test.run=^TestAttributeTruncateRetainLockOrder$", "-test.count=1", "-test.timeout=12s") + cmd.Env = append(os.Environ(), "MOUNT_ATTR_TRUNCATE_LOCK_CHILD=1") + output, err := cmd.CombinedOutput() + if ctx.Err() != nil { + t.Fatalf("truncate lock control exceeded bounded child: %v\n%s", ctx.Err(), output) + } + if err != nil { + t.Fatalf("concurrent truncate/acked chmod did not finish: %v\n%s", err, output) + } +} +func attributeTruncateLockScenario(t *testing.T) { + w := newInvalidateTestWFS(t) + w.atimeMap = make(map[uint64]time.Time) + current := attributeAppendEntry(3) + fake := &attributeCASFiler{entry: proto.Clone(current).(*filer_pb.Entry)} + committed, resume := make(chan struct{}), make(chan struct{}) + var once sync.Once + t.Cleanup(func() { once.Do(func() { close(resume) }) }) + fake.afterCommit = func() { close(committed); <-resume } + startFakeFiler(t, w, fake) + path := util.FullPath("/dir/file") + w.inodeToPath.Lookup(path, 1, false, false, 42, true) + local := proto.Clone(current).(*filer_pb.Entry) + w.mapPbIdFromFilerToLocal(local) + fh := newFileHandle(w, 1, 42, local) + fh.RememberPath(path) + w.fhMap.inode2fh[42], w.fhMap.fh2inode[1] = fh, 42 + t.Cleanup(fh.dirtyPages.Destroy) + chmodDone := make(chan fuse.Status, 1) + go func() { + in := &fuse.SetAttrIn{} + in.NodeId = 42 + in.Valid = fuse.FATTR_MODE + in.Mode = 0600 + chmodDone <- w.SetAttr(nil, in, &fuse.AttrOut{}) + }() + select { + case <-committed: + case <-time.After(5 * time.Second): + t.Fatal("chmod did not reach owner commit hook") + } + active := w.fhLockTable.AcquireLock("test pending content", fh.fh, util.ExclusiveLock) + fh.dirtyMetadata = true + w.fhLockTable.ReleaseLock(fh.fh, active) + fh.entryLock.RLock() + truncateDone := make(chan fuse.Status, 1) + go func() { + in := &fuse.SetAttrIn{} + in.NodeId = 42 + in.Valid = fuse.FATTR_SIZE + in.Size = 256 + truncateDone <- w.SetAttr(nil, in, &fuse.AttrOut{}) + }() + // Truncate queued its entryLock writer before the chmod response is released. + waitAttributeLockPair(t, [][2]string{{"(*WFS).attributeEntrySnapshot", "sync.(*RWMutex).Lock"}}) + once.Do(func() { close(resume) }) + // Rejected head: retention owns active-exclusive and waits for entryLock. + // Corrected head: truncate owns active-shared and installation waits for it. + waitAttributeLockPair(t, [][2]string{{"(*WFS).retainAcknowledgedAttributes", "sync.(*RWMutex).Lock"}, {"(*FileHandle).installAckedEntry", "AcquireLock"}}) + fh.entryLock.RUnlock() + timer := time.NewTimer(2 * time.Second) + defer timer.Stop() + for _, done := range []chan fuse.Status{chmodDone, truncateDone} { + select { + case status := <-done: + if status != fuse.OK { + t.Fatalf("handler returned %v", status) + } + case <-timer.C: + fmt.Fprintln(os.Stderr, "TRUNCATE_RETENTION_CYCLE: FATTR_SIZE snapshot and acknowledged chmod retention cannot finish") + os.Exit(24) + } + } + fh.entry.RLock() + defer fh.entry.RUnlock() + e := fh.entry.Entry + if e.Attributes.FileMode&0777 != 0600 || e.Attributes.FileSize != 256 || len(e.Chunks) != 1 || !proto.Equal(e.Chunks[0], current.Chunks[0]) { + t.Fatalf("lost acknowledged chmod or pending truncate: %v", e) + } +} +func waitAttributeLockPair(t *testing.T, pairs [][2]string) { + t.Helper() + end := time.Now().Add(5 * time.Second) + for time.Now().Before(end) { + buf := make([]byte, 1<<20) + n := runtime.Stack(buf, true) + for _, stack := range strings.Split(string(buf[:n]), "\n\n") { + header, _, _ := strings.Cut(stack, "\n") + if !strings.Contains(header, "[sync.") && !strings.Contains(header, "[semacquire") { + continue + } + for _, pair := range pairs { + if strings.Contains(stack, pair[0]) && strings.Contains(stack, pair[1]) { + return + } + } + } + runtime.Gosched() + } + t.Fatalf("did not observe actual blocked lock pair %v", pairs) +} diff --git a/weed/mount/filehandle.go b/weed/mount/filehandle.go index fa26379c7c..627453fa17 100644 --- a/weed/mount/filehandle.go +++ b/weed/mount/filehandle.go @@ -20,10 +20,13 @@ type FileHandleId uint64 var IsDebugFileReadWrite = false type FileHandle struct { - fh FileHandleId - counter int64 - fhs map[FileHandleId]struct{} // live fh values, one per open; guarded by fhMap's lock - entry *LockedEntry + fh FileHandleId + counter int64 + fhs map[FileHandleId]struct{} // live fh values, one per open; guarded by fhMap's lock + entry *LockedEntry + // Nested handle lock order: wfs.fhLockTable[fh.fh] -> entryLock -> + // LockedEntry.RWMutex. Paths may use a subset, but must never acquire an + // earlier lock while holding a later one. Release in reverse order. entryLock sync.RWMutex entryChunkGroup *filer.ChunkGroup inode uint64 diff --git a/weed/mount/weedfs.go b/weed/mount/weedfs.go index 47564574c8..5cd89e448a 100644 --- a/weed/mount/weedfs.go +++ b/weed/mount/weedfs.go @@ -154,32 +154,36 @@ type WFS struct { dhMap *DirectoryHandleToInode fuseServer *fuse.Server IsOverQuota bool - fhLockTable *util.LockTable[FileHandleId] - hardLinkLockTable *util.LockTable[string] - posixLocks *PosixLockTable - posixSid uint64 // this mount's session id, for routed-lock owner identity - posixHint *posixLockHint // local fcntl-lock hint for routed mode - posixOwn *posixlock.Manager // mirror of locks this mount holds, re-asserted via keepalive - fhLocks fhLockOwners // non-flock lock owners per open fh, drained at release - peerRegistrar *PeerRegistrar - peerDirectory *PeerDirectory - peerGrpcServer *PeerGrpcServer - peerAnnouncer *PeerAnnouncer - peerConnPool *PeerConnPool - peerDirectoryStop chan struct{} // closed on unmount to stop the sweeper goroutine - FilerConf *filer.FilerConf - filerClient *wdclient.FilerClient // Cached volume location client - refreshMu sync.Mutex - refreshingDirs map[util.FullPath]struct{} - atimeMu sync.Mutex - atimeMap map[uint64]time.Time // inode -> atime, in-memory only, bounded - dirMtimeMu sync.Mutex - dirMtimeMap map[uint64]time.Time // inode -> mtime/ctime, in-memory overlay for dirs - removedDirMu sync.Mutex - removedDirs map[uint64]*filer_pb.Entry // inode -> last-known entry of a directory removed while still referenced - entryValidSec uint64 // kernel FUSE entry cache TTL in seconds - attrValidSec uint64 // kernel FUSE attr cache TTL in seconds - dirIdleEvict time.Duration + // Handle lock nesting order is defined at FileHandle.entryLock. + fhLockTable *util.LockTable[FileHandleId] + hardLinkLockTable *util.LockTable[string] + posixLocks *PosixLockTable + posixSid uint64 // this mount's session id, for routed-lock owner identity + posixHint *posixLockHint // local fcntl-lock hint for routed mode + posixOwn *posixlock.Manager // mirror of locks this mount holds, re-asserted via keepalive + fhLocks fhLockOwners // non-flock lock owners per open fh, drained at release + peerRegistrar *PeerRegistrar + peerDirectory *PeerDirectory + peerGrpcServer *PeerGrpcServer + peerAnnouncer *PeerAnnouncer + peerConnPool *PeerConnPool + peerDirectoryStop chan struct{} // closed on unmount to stop the sweeper goroutine + FilerConf *filer.FilerConf + filerClient *wdclient.FilerClient // Cached volume location client + refreshMu sync.Mutex + refreshingDirs map[util.FullPath]struct{} + atimeMu sync.Mutex + atimeMap map[uint64]time.Time // inode -> atime, in-memory only, bounded + dirMtimeMu sync.Mutex + dirMtimeMap map[uint64]time.Time // inode -> mtime/ctime, in-memory overlay for dirs + removedDirMu sync.Mutex + removedDirs map[uint64]*filer_pb.Entry // inode -> last-known entry of a directory removed while still referenced + entryValidSec uint64 // kernel FUSE entry cache TTL in seconds + attrValidSec uint64 // kernel FUSE attr cache TTL in seconds + dirIdleEvict time.Duration + + metadataPublicationOnce sync.Once + metadataPublicationLocks *util.LockTable[util.FullPath] // openMtimeCache maps inode -> [mtime_sec, mtime_ns] from the last Open. // Used to decide whether to set FOPEN_KEEP_CACHE on subsequent opens. @@ -955,6 +959,8 @@ func (wfs *WFS) invalidateOpenFileHandle(invalidation meta_cache.EntryInvalidati handled = true fhActiveLock := wfs.fhLockTable.AcquireLock("invalidateFunc", fh.fh, util.ExclusiveLock) defer wfs.fhLockTable.ReleaseLock(fh.fh, fhActiveLock) + fh.entryLock.Lock() + defer fh.entryLock.Unlock() // A rename changes which name the inode answers to, not which version of // its content the handle holds, so it is applied ahead of the version fence diff --git a/weed/mount/weedfs_attr.go b/weed/mount/weedfs_attr.go index b592ad921d..b394912d9e 100644 --- a/weed/mount/weedfs_attr.go +++ b/weed/mount/weedfs_attr.go @@ -73,6 +73,27 @@ func (wfs *WFS) SetAttr(cancel <-chan struct{}, input *fuse.SetAttrIn, out *fuse if status != fuse.OK || entry == nil { return status } + if input.Valid&fuse.FATTR_SIZE == 0 && path != "" { + unlock := wfs.lockMetadataPublication(path) + defer unlock() + } + if snapshot, ok := wfs.attributeEntrySnapshot(fh, entry); ok && input.Valid&fuse.FATTR_SIZE == 0 && path != "" { + entry, status = wfs.saveEntryAttributes(ctx, path, snapshot, input.Valid&(fuse.FATTR_UID|fuse.FATTR_GID), func(next *filer_pb.Entry) fuse.Status { + return wfs.applySetAttr(path, input, next) + }, func(live, acked *filer_pb.Entry) { + retainSetAttr(input, live, acked) + }) + if status != fuse.OK { + return status + } + if atime, ok := input.GetATime(); ok { + wfs.setAtime(input.NodeId, atime) + } + out.AttrValid = wfs.attrValidSec + wfs.setAttrByPbEntry(&out.Attr, input.NodeId, entry, true) + wfs.applyInMemoryAtime(&out.Attr, input.NodeId) + return fuse.OK + } if size, ok := input.GetSize(); ok && fh != nil && fh.dirtyPages.HasWrites() && size < filer.FileSize(entry) { // The truncation below trims chunks; dirty pages it cannot see. Left // alone, pages beyond the new size come back with the next flush and @@ -86,6 +107,8 @@ func (wfs *WFS) SetAttr(cancel <-chan struct{}, input *fuse.SetAttrIn, out *fuse } } if fh != nil { + active := wfs.fhLockTable.AcquireLock("SetAttr", fh.fh, util.ExclusiveLock) + defer wfs.fhLockTable.ReleaseLock(fh.fh, active) fh.entryLock.Lock() defer fh.entryLock.Unlock() // entry is the handle's shared LockedEntry.Entry. Async upload workers @@ -99,7 +122,7 @@ func (wfs *WFS) SetAttr(cancel <-chan struct{}, input *fuse.SetAttrIn, out *fuse entry = fh.entry.Entry } - wormEnforced, wormEnabled := wfs.wormEnforcedForEntry(path, entry) + wormEnforced, _ := wfs.wormEnforcedForEntry(path, entry) if wormEnforced { return fuse.EPERM } @@ -150,6 +173,63 @@ func (wfs *WFS) SetAttr(cancel <-chan struct{}, input *fuse.SetAttrIn, out *fuse } + if code := wfs.applySetAttr(path, input, entry); code != fuse.OK { + return code + } + if atime, ok := input.GetATime(); ok { + wfs.setAtime(input.NodeId, atime) + } + + out.AttrValid = wfs.attrValidSec + size, includeSize := input.GetSize() + if includeSize { + out.Attr.Size = size + } + wfs.setAttrByPbEntry(&out.Attr, input.NodeId, entry, !includeSize) + wfs.applyInMemoryAtime(&out.Attr, input.NodeId) + applyUnlinkedNlink(&out.Attr, path) + + if fh != nil { + fh.dirtyMetadata = true + return fuse.OK + } + if path == "" { + // removed while open: the remembered entry is all there is to update + wfs.rememberRemovedDir(input.NodeId, entry) + return fuse.OK + } + + return wfs.saveEntry(path, entry) + +} + +func retainSetAttr(input *fuse.SetAttrIn, live, acked *filer_pb.Entry) { + if _, ok := input.GetMode(); ok { + live.Attributes.FileMode = chmod(live.Attributes.FileMode, acked.Attributes.FileMode) + live.WormEnforcedAtTsNs = acked.WormEnforcedAtTsNs + } + _, uidSet := input.GetUID() + _, gidSet := input.GetGID() + if uidSet { + live.Attributes.Uid = acked.Attributes.Uid + } + if gidSet { + live.Attributes.Gid = acked.Attributes.Gid + } + if (uidSet || gidSet) && input.Uid != 0 { + live.Attributes.FileMode &^= 0o6000 + } + if _, ok := input.GetMTime(); ok { + live.Attributes.Mtime, live.Attributes.MtimeNs = acked.Attributes.Mtime, acked.Attributes.MtimeNs + } + live.Attributes.Ctime, live.Attributes.CtimeNs = acked.Attributes.Ctime, acked.Attributes.CtimeNs +} + +func (wfs *WFS) applySetAttr(path util.FullPath, input *fuse.SetAttrIn, entry *filer_pb.Entry) fuse.Status { + wormEnforced, wormEnabled := wfs.wormEnforcedForEntry(path, entry) + if wormEnforced { + return fuse.EPERM + } if mode, ok := input.GetMode(); ok { // commit the file to worm when it is set to readonly at the first time if entry.WormEnforcedAtTsNs == 0 && wormEnabled && !hasWritePermission(mode) { @@ -185,10 +265,6 @@ func (wfs *WFS) SetAttr(cancel <-chan struct{}, input *fuse.SetAttrIn, out *fuse entry.Attributes.FileMode &^= 0o6000 } - if atime, ok := input.GetATime(); ok { - wfs.setAtime(input.NodeId, atime) - } - if mtime, ok := input.GetMTime(); ok { entry.Attributes.Mtime = mtime.Unix() entry.Attributes.MtimeNs = int32(mtime.Nanosecond()) @@ -199,27 +275,7 @@ func (wfs *WFS) SetAttr(cancel <-chan struct{}, input *fuse.SetAttrIn, out *fuse entry.Attributes.Ctime = now.Unix() entry.Attributes.CtimeNs = int32(now.Nanosecond()) - out.AttrValid = wfs.attrValidSec - size, includeSize := input.GetSize() - if includeSize { - out.Attr.Size = size - } - wfs.setAttrByPbEntry(&out.Attr, input.NodeId, entry, !includeSize) - wfs.applyInMemoryAtime(&out.Attr, input.NodeId) - applyUnlinkedNlink(&out.Attr, path) - - if fh != nil { - fh.dirtyMetadata = true - return fuse.OK - } - if path == "" { - // removed while open: the remembered entry is all there is to update - wfs.rememberRemovedDir(input.NodeId, entry) - return fuse.OK - } - - return wfs.saveEntry(path, entry) - + return fuse.OK } func (wfs *WFS) setRootAttr(out *fuse.AttrOut) { diff --git a/weed/mount/weedfs_attr_cas_test.go b/weed/mount/weedfs_attr_cas_test.go new file mode 100644 index 0000000000..ac6a0bd491 --- /dev/null +++ b/weed/mount/weedfs_attr_cas_test.go @@ -0,0 +1,756 @@ +//go:build !freebsd + +package mount + +import ( + "context" + "fmt" + "runtime" + "strings" + "sync" + "syscall" + "testing" + "time" + + "github.com/seaweedfs/go-fuse/v2/fuse" + "github.com/seaweedfs/seaweedfs/weed/cluster" + "github.com/seaweedfs/seaweedfs/weed/filer" + "github.com/seaweedfs/seaweedfs/weed/pb" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/util" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + "google.golang.org/protobuf/proto" +) + +// The writer's acknowledged entry and a second mount's cached entry diverge +// before the attribute syscall. The filer implements the existing atomic +// IF_ENTRY_EQUAL contract; no sleeps or FUSE device are needed for this order. +type attributeCASFiler struct { + filer_pb.UnimplementedSeaweedFilerServer + mu sync.Mutex + entry *filer_pb.Entry + updates, conflicts int + beforeUpdate func(int, *filer_pb.Entry) + afterCommit func() + updateErr error + missing bool +} + +func (s *attributeCASFiler) LookupDirectoryEntry(ctx context.Context, req *filer_pb.LookupDirectoryEntryRequest) (*filer_pb.LookupDirectoryEntryResponse, error) { + s.mu.Lock() + defer s.mu.Unlock() + if s.missing { + return nil, status.Error(codes.NotFound, "deleted") + } + return &filer_pb.LookupDirectoryEntryResponse{Entry: proto.Clone(s.entry).(*filer_pb.Entry)}, nil +} + +func (s *attributeCASFiler) UpdateEntry(ctx context.Context, req *filer_pb.UpdateEntryRequest) (*filer_pb.UpdateEntryResponse, error) { + s.mu.Lock() + defer s.mu.Unlock() + s.updates++ + if s.updateErr != nil { + return nil, s.updateErr + } + if s.beforeUpdate != nil { + s.beforeUpdate(s.updates, s.entry) + } + if clauses := req.GetCondition().GetClauses(); len(clauses) > 0 && !proto.Equal(clauses[0].GetExpectedEntry(), s.entry) { + s.conflicts++ + return nil, status.Error(codes.FailedPrecondition, "entry changed") + } + s.entry = proto.Clone(req.Entry).(*filer_pb.Entry) + if s.afterCommit != nil { + // Release the owner lock: concurrent CreateEntry must be able to publish + // while the committed UpdateEntry response is held at this barrier. + s.mu.Unlock() + s.afterCommit() + s.mu.Lock() + } + return &filer_pb.UpdateEntryResponse{LogTsNs: 2000}, nil +} + +func TestAttributeOnlyConflictOutcomes(t *testing.T) { + for _, tc := range []struct { + name string + flags uint32 + prepare func(*attributeCASFiler) + want fuse.Status + calls int + }{ + {"bounded", 0, func(s *attributeCASFiler) { + s.beforeUpdate = func(n int, e *filer_pb.Entry) { e.Extended["concurrent"] = []byte(fmt.Sprint(n)) } + }, fuse.Status(syscall.EAGAIN), 3}, + {"replacement", 0, func(s *attributeCASFiler) { s.entry.Attributes.Inode = 99 }, fuse.Status(syscall.ESTALE), 1}, + {"deleted", 0, func(s *attributeCASFiler) { s.missing = true }, fuse.ENOENT, 1}, + {"create-existing", xattr_CREATE, func(s *attributeCASFiler) { s.entry.Extended["xattr-user.test"] = []byte("other mount") }, fuse.Status(syscall.EEXIST), 1}, + {"replace-removed", xattr_REPLACE, func(s *attributeCASFiler) { delete(s.entry.Extended, "xattr-user.test") }, fuse.ENODATA, 1}, + {"permission", 0, func(s *attributeCASFiler) { s.updateErr = status.Error(codes.PermissionDenied, "refused") }, fuse.Status(syscall.EACCES), 1}, + } { + t.Run(tc.name, func(t *testing.T) { + wfs := newInvalidateTestWFS(t) + cached := attributeAppendEntry(1) + if tc.flags == xattr_REPLACE { + cached.Extended["xattr-user.test"] = []byte("cached") + } + fake := &attributeCASFiler{entry: attributeAppendEntry(300)} + tc.prepare(fake) + startFakeFiler(t, wfs, fake) + path := util.FullPath("/dir/file") + wfs.inodeToPath.Lookup(util.FullPath("/dir"), 1, true, false, 0, false) + wfs.inodeToPath.Lookup(path, 1, false, false, 42, true) + wfs.inodeToPath.MarkChildrenCached(util.FullPath("/dir")) + if err := wfs.metaCache.InsertEntry(context.Background(), filer.FromPbEntry("/dir", cached), 1); err != nil { + t.Fatal(err) + } + in := &fuse.SetXAttrIn{Flags: tc.flags} + in.NodeId = 42 + if code := wfs.SetXAttr(nil, in, "user.test", []byte("changed")); code != tc.want { + t.Fatalf("got %v, want %v", code, tc.want) + } + fake.mu.Lock() + defer fake.mu.Unlock() + if fake.updates != tc.calls { + t.Fatalf("updates=%d want %d", fake.updates, tc.calls) + } + if len(fake.entry.Chunks) != 300 || fake.entry.Attributes.FileSize != 76800 { + t.Fatal("failed attribute op changed content") + } + }) + } +} + +func (s *attributeCASFiler) CreateEntry(ctx context.Context, req *filer_pb.CreateEntryRequest) (*filer_pb.CreateEntryResponse, error) { + s.mu.Lock() + defer s.mu.Unlock() + s.entry = proto.Clone(req.Entry).(*filer_pb.Entry) + return &filer_pb.CreateEntryResponse{LogTsNs: 2000}, nil +} + +func attributeAppendEntry(records int) *filer_pb.Entry { + e := &filer_pb.Entry{Name: "file", Attributes: &filer_pb.FuseAttributes{ + Inode: 42, Uid: 2000, Gid: 20, FileMode: 0644, FileSize: uint64(records * 256), Mtime: 10, + }, Extended: map[string][]byte{"xattr-user.other": []byte("keep"), "xattr-user.remove": []byte("remove")}} + for i := 0; i < records; i++ { + e.Chunks = append(e.Chunks, &filer_pb.FileChunk{Fid: &filer_pb.FileId{VolumeId: 1, FileKey: uint64(i + 1), Cookie: 1}, Offset: int64(i * 256), Size: 256, ModifiedTsNs: int64(i + 1)}) + } + return e +} + +func TestAttributeOnlyOpsPreserveAcknowledgedAppends(t *testing.T) { + for _, op := range []string{"chmod", "chown", "utimes", "setxattr", "removexattr"} { + for _, stale := range []bool{false, true} { + for _, open := range []bool{false, true} { + t.Run(fmt.Sprintf("%s/stale=%t/open=%t", op, stale, open), func(t *testing.T) { + wfs := newInvalidateTestWFS(t) + wfs.atimeMap = make(map[uint64]time.Time) + current := attributeAppendEntry(300) + fake := &attributeCASFiler{entry: proto.Clone(current).(*filer_pb.Entry)} + startFakeFiler(t, wfs, fake) + cached := proto.Clone(current).(*filer_pb.Entry) + if stale { + cached = attributeAppendEntry(1) + cached.Extended["xattr-user.other"] = []byte("obsolete") + cached.Attributes.Gid = 17 + } + path := util.FullPath("/dir/file") + wfs.inodeToPath.Lookup(util.FullPath("/dir"), 1, true, false, 0, false) + wfs.inodeToPath.Lookup(path, 1, false, false, 42, true) + wfs.inodeToPath.MarkChildrenCached(util.FullPath("/dir")) + if err := wfs.metaCache.InsertEntry(context.Background(), filer.FromPbEntry("/dir", cached), 1); err != nil { + t.Fatal(err) + } + var fh *FileHandle + if open { + local := proto.Clone(cached).(*filer_pb.Entry) + wfs.mapPbIdFromFilerToLocal(local) + fh = &FileHandle{fh: 1, inode: 42, wfs: wfs, entry: &LockedEntry{Entry: local}} + fh.dirtyPages = newPageWriter(fh, 1024) + fh.RememberPath(path) + wfs.fhMap.inode2fh[42], wfs.fhMap.fh2inode[1] = fh, 42 + t.Cleanup(func() { fh.dirtyPages.Destroy() }) + } + var code fuse.Status + out := &fuse.AttrOut{} + if op == "removexattr" { + code = wfs.RemoveXAttr(nil, &fuse.InHeader{NodeId: 42}, "user.remove") + } else if op == "setxattr" { + in := &fuse.SetXAttrIn{} + in.NodeId = 42 + code = wfs.SetXAttr(nil, in, "user.test", []byte("new")) + } else { + in := &fuse.SetAttrIn{} + in.NodeId = 42 + switch op { + case "chmod": + in.Valid, in.Mode = fuse.FATTR_MODE, 0600 + case "chown": + in.Valid, in.Owner.Uid, in.Owner.Gid = fuse.FATTR_UID|fuse.FATTR_GID, 1001, 21 + case "utimes": + in.Valid, in.Mtime, in.Mtimensec = fuse.FATTR_MTIME|fuse.FATTR_ATIME, 123, 456 + } + code = wfs.SetAttr(nil, in, out) + } + if code != fuse.OK { + t.Fatalf("attribute syscall: %v", code) + } + if (op == "chmod" || op == "utimes") && out.Attr.Uid != 1000 { + t.Fatalf("reply uid %d, want the local id 1000", out.Attr.Uid) + } + if open { + if code := wfs.doFlush(context.Background(), fh, 1000, 20, false); code != fuse.OK { + t.Fatalf("flush after attribute syscall: %v", code) + } + } + fake.mu.Lock() + defer fake.mu.Unlock() + if len(fake.entry.Chunks) != 300 || fake.entry.Attributes.FileSize != 76800 { + t.Fatalf("lost fsync-acknowledged appends: chunks=%d size=%d, want 300/76800", len(fake.entry.Chunks), fake.entry.Attributes.FileSize) + } + for i, chunk := range fake.entry.Chunks { + if chunk.GetFileIdString() != current.Chunks[i].GetFileIdString() || chunk.Offset != int64(i*256) || chunk.Size != 256 { + t.Fatalf("acknowledged record %d no longer references its original bytes: %v", i, chunk) + } + } + switch op { + case "chmod": + if fake.entry.Attributes.FileMode != 0600 { + t.Fatal("mode not saved") + } + case "chown": + if fake.entry.Attributes.Uid != 1001 || fake.entry.Attributes.Gid != 21 { + t.Fatal("owner not saved") + } + case "utimes": + if fake.entry.Attributes.Mtime != 123 || fake.entry.Attributes.MtimeNs != 456 { + t.Fatal("mtime not saved") + } + case "setxattr": + if string(fake.entry.Extended["xattr-user.test"]) != "new" { + t.Fatal("xattr not saved") + } + case "removexattr": + if _, exists := fake.entry.Extended["xattr-user.remove"]; exists { + t.Fatal("xattr not removed") + } + } + if string(fake.entry.Extended["xattr-user.other"]) != "keep" { + t.Fatal("unrelated xattr lost") + } + if op != "chown" && (fake.entry.Attributes.Uid != 2000 || fake.entry.Attributes.Gid != 20) { + t.Fatal("unrelated owner or uid mapping lost") + } + }) + } + } + } +} + +// The filer holds the attribute RPC while the real WRITE handler dirties the +// same clean handle. Its subsequent flush must keep both the acknowledged +// attribute and the buffered bytes. The upload adapter stores bytes in memory. +func TestAcknowledgedAttributeSurvivesWriteDuringRPC(t *testing.T) { + for _, op := range []string{"chmod", "chown", "utimes", "setxattr", "removexattr"} { + t.Run(op, func(t *testing.T) { + wfs := newInvalidateTestWFS(t) + wfs.atimeMap = make(map[uint64]time.Time) + wfs.option.ConcurrentWriters = 1 + wfs.concurrentWriters = util.NewLimitedConcurrentExecutor(1) + current := attributeAppendEntry(1) + fake := &attributeCASFiler{entry: proto.Clone(current).(*filer_pb.Entry)} + entered, resume := make(chan struct{}), make(chan struct{}) + var once sync.Once + t.Cleanup(func() { once.Do(func() { close(resume) }) }) + fake.beforeUpdate = func(n int, e *filer_pb.Entry) { + if n == 1 { + close(entered) + <-resume + } + } + startFakeFiler(t, wfs, fake) + path := util.FullPath("/dir/file") + wfs.inodeToPath.Lookup(path, 1, false, false, 42, true) + local := proto.Clone(current).(*filer_pb.Entry) + wfs.mapPbIdFromFilerToLocal(local) + fh := newFileHandle(wfs, 1, 42, local) + fh.RememberPath(path) + wfs.fhMap.inode2fh[42], wfs.fhMap.fh2inode[1] = fh, 42 + save := &flakySave{} + fh.dirtyPages.randomWriter.(*ChunkedDirtyPages).saveDataAsChunk = save.saveDataAsChunk + t.Cleanup(fh.dirtyPages.Destroy) + done := make(chan fuse.Status, 1) + go func() { + if op == "setxattr" { + done <- wfs.SetXAttr(nil, &fuse.SetXAttrIn{InHeader: fuse.InHeader{NodeId: 42}}, "user.test", []byte("acknowledged")) + } else if op == "removexattr" { + done <- wfs.RemoveXAttr(nil, &fuse.InHeader{NodeId: 42}, "user.remove") + } else { + in := &fuse.SetAttrIn{} + in.NodeId = 42 + switch op { + case "chmod": + in.Valid, in.Mode = fuse.FATTR_MODE, 0600 + case "chown": + in.Valid, in.Owner.Uid, in.Owner.Gid = fuse.FATTR_UID|fuse.FATTR_GID, 1001, 21 + case "utimes": + in.Valid, in.Mtime, in.Mtimensec = fuse.FATTR_MTIME, 123, 456 + } + done <- wfs.SetAttr(nil, in, &fuse.AttrOut{}) + } + }() + select { + case <-entered: + case <-time.After(10 * time.Second): + t.Fatal("attribute RPC never arrived") + } + payload := []byte("background write while attribute RPC waits") + in := &fuse.WriteIn{} + in.NodeId, in.Fh, in.Offset, in.Size = 42, 1, 256, uint32(len(payload)) + if n, code := wfs.Write(nil, in, payload); code != fuse.OK || n != uint32(len(payload)) { + t.Fatalf("WRITE: %d %v", n, code) + } + once.Do(func() { close(resume) }) + select { + case code := <-done: + if code != fuse.OK { + t.Fatalf("attribute: %v", code) + } + case <-time.After(10 * time.Second): + t.Fatal("attribute RPC did not finish") + } + if code := wfs.doFlush(context.Background(), fh, 1000, 20, false); code != fuse.OK { + t.Fatalf("flush: %v", code) + } + fake.mu.Lock() + defer fake.mu.Unlock() + switch op { + case "chmod": + if fake.entry.Attributes.FileMode != 0600 { + t.Fatal("flush undid acknowledged chmod") + } + case "chown": + if fake.entry.Attributes.Uid != 1001 || fake.entry.Attributes.Gid != 21 { + t.Fatal("flush undid acknowledged chown") + } + case "utimes": + if fake.entry.Attributes.Mtime != 123 || fake.entry.Attributes.MtimeNs != 456 { + t.Fatal("flush undid acknowledged utimes") + } + case "setxattr": + if string(fake.entry.Extended["xattr-user.test"]) != "acknowledged" { + t.Fatal("flush undid acknowledged setxattr") + } + case "removexattr": + if _, exists := fake.entry.Extended["xattr-user.remove"]; exists { + t.Fatal("flush undid acknowledged removexattr") + } + } + if len(save.bodies) != 1 || save.bodies[0] != string(payload) { + t.Fatalf("pending bytes changed: %q", save.bodies) + } + if fake.entry.Attributes.FileSize != 256+uint64(len(payload)) || len(fake.entry.Chunks) != 2 { + t.Fatalf("pending content lost: %v", fake.entry) + } + original, appended := false, false + for _, chunk := range fake.entry.Chunks { + switch chunk.Offset { + case 0: + original = chunk.GetFileIdString() == current.Chunks[0].GetFileIdString() && chunk.Size == 256 + case 256: + appended = chunk.GetFileIdString() == "1,01" && chunk.Size == uint64(len(payload)) + } + } + if !original || !appended { + t.Fatalf("chunk identity/range changed: %v", fake.entry.Chunks) + } + + }) + } +} + +// Force owner-commit -> concurrent WRITE -> whole-entry flush -> RPC response. +// The corrected caller must serialize publication across the entire interval. +func TestAttributePublicationBeforeResponse(t *testing.T) { + for _, contention := range []string{"response", "installation", "retention"} { + for _, op := range []string{"chmod", "chown", "utimes", "setxattr", "removexattr"} { + t.Run(contention+"/"+op, func(t *testing.T) { + wfs := newInvalidateTestWFS(t) + wfs.atimeMap = make(map[uint64]time.Time) + wfs.option.ConcurrentWriters = 1 + wfs.concurrentWriters = util.NewLimitedConcurrentExecutor(1) + current := attributeAppendEntry(1) + fake := &attributeCASFiler{entry: proto.Clone(current).(*filer_pb.Entry)} + entered, resume := make(chan struct{}), make(chan struct{}) + var once sync.Once + t.Cleanup(func() { once.Do(func() { close(resume) }) }) + fake.afterCommit = func() { close(entered); <-resume } + + startFakeFiler(t, wfs, fake) + path := util.FullPath("/dir/file") + wfs.inodeToPath.Lookup(path, 1, false, false, 42, true) + local := proto.Clone(current).(*filer_pb.Entry) + wfs.mapPbIdFromFilerToLocal(local) + fh := newFileHandle(wfs, 1, 42, local) + fh.RememberPath(path) + wfs.fhMap.inode2fh[42], wfs.fhMap.fh2inode[1] = fh, 42 + save := &flakySave{} + fh.dirtyPages.randomWriter.(*ChunkedDirtyPages).saveDataAsChunk = save.saveDataAsChunk + t.Cleanup(fh.dirtyPages.Destroy) + done := make(chan fuse.Status, 1) + go func() { + if op == "setxattr" { + done <- wfs.SetXAttr(nil, &fuse.SetXAttrIn{InHeader: fuse.InHeader{NodeId: 42}}, "user.test", []byte("acknowledged")) + } else if op == "removexattr" { + done <- wfs.RemoveXAttr(nil, &fuse.InHeader{NodeId: 42}, "user.remove") + } else { + in := &fuse.SetAttrIn{} + in.NodeId = 42 + switch op { + case "chmod": + in.Valid, in.Mode = fuse.FATTR_MODE, 0600 + case "chown": + in.Valid, in.Owner.Uid, in.Owner.Gid = fuse.FATTR_UID|fuse.FATTR_GID, 1001, 21 + case "utimes": + in.Valid, in.Mtime, in.Mtimensec = fuse.FATTR_MTIME, 123, 456 + } + done <- wfs.SetAttr(nil, in, &fuse.AttrOut{}) + } + }() + select { + case <-entered: + case <-time.After(10 * time.Second): + t.Fatal("attribute RPC never arrived") + } + payload := []byte("background write while attribute RPC waits") + in := &fuse.WriteIn{} + in.NodeId, in.Fh, in.Offset, in.Size = 42, 1, 256, uint32(len(payload)) + if n, code := wfs.Write(nil, in, payload); code != fuse.OK || n != uint32(len(payload)) { + t.Fatalf("WRITE: %d %v", n, code) + } + flushDone := make(chan fuse.Status, 1) + flushSettled, publicationEscaped := false, false + flush := func() { flushDone <- wfs.doFlush(context.Background(), fh, 1000, 20, false) } + // Probe ownership while the committed response is held. On the old caller + // no publication boundary exists, so force its whole-entry flush to commit + // BEFORE releasing the response. With coordination, queue that same flush + // behind the attribute and force contention at installation/retention. + table := wfs.metadataPublicationTable() + probe, unguarded := table.TryAcquireLock("test publication witness", path, util.ExclusiveLock) + if unguarded { + table.ReleaseLock(path, probe) + flush() + if code := <-flushDone; code != fuse.OK { + t.Fatalf("pre-response flush: %v", code) + } + once.Do(func() { close(resume) }) + } else { + release := func() {} + if contention == "installation" { + active := wfs.fhLockTable.AcquireLock("test installation contention", fh.fh, util.ExclusiveLock) + release = func() { wfs.fhLockTable.ReleaseLock(fh.fh, active) } + } else if contention == "retention" { + fh.entryLock.Lock() + release = fh.entryLock.Unlock + } + go flush() + // Do not open the RPC response gate merely because flush was started: + // witness this caller actually asleep on the publication condition. + if blocked, code := waitFlushPublication(t, flushDone); !blocked { + flushSettled, publicationEscaped = true, true + if code != fuse.OK { + t.Errorf("flush before response gate: %v", code) + } + } + once.Do(func() { close(resume) }) + if contention != "response" { + frame := "installAckedEntry" + if contention == "retention" { + frame = "retainAcknowledgedAttributes" + } + // A stack witness proves the real caller reached the held lock. Time + // only bounds a broken test; no latency threshold decides correctness. + blocked := waitAttributeFrame(frame) + if blocked && !flushSettled { + // Check again before releasing installation/retention. A prematurely + // unlocked publication lets the real flush finish while this gate is + // held: join that flush BEFORE retention, forcing the lost-attr order. + if waiting, code := waitFlushPublication(t, flushDone); !waiting { + flushSettled, publicationEscaped = true, true + if code != fuse.OK { + t.Errorf("flush before %s gate: %v", contention, code) + } + } + } + release() + if !blocked { + t.Fatalf("caller never contended at %s", frame) + } + } + } + select { + case code := <-done: + if code != fuse.OK { + t.Fatalf("attribute: %v", code) + } + case <-time.After(10 * time.Second): + t.Fatal("attribute did not finish") + } + if !unguarded && !flushSettled { + select { + case code := <-flushDone: + if code != fuse.OK { + t.Fatalf("flush: %v", code) + } + case <-time.After(10 * time.Second): + t.Fatal("flush did not finish") + } + } + + fake.mu.Lock() + defer fake.mu.Unlock() + switch op { + case "chmod": + if fake.entry.Attributes.FileMode != 0600 { + t.Fatal("flush undid acknowledged chmod") + } + case "chown": + if fake.entry.Attributes.Uid != 1001 || fake.entry.Attributes.Gid != 21 { + t.Fatal("flush undid acknowledged chown") + } + case "utimes": + if fake.entry.Attributes.Mtime != 123 || fake.entry.Attributes.MtimeNs != 456 { + t.Fatal("flush undid acknowledged utimes") + } + case "setxattr": + if string(fake.entry.Extended["xattr-user.test"]) != "acknowledged" { + t.Fatal("flush undid acknowledged setxattr") + } + case "removexattr": + if _, exists := fake.entry.Extended["xattr-user.remove"]; exists { + t.Fatal("flush undid acknowledged removexattr") + } + } + if len(save.bodies) != 1 || save.bodies[0] != string(payload) { + t.Fatalf("pending bytes changed: %q", save.bodies) + } + if fake.entry.Attributes.FileSize != 256+uint64(len(payload)) || len(fake.entry.Chunks) != 2 { + t.Fatalf("pending content lost: %v", fake.entry) + } + original, appended := false, false + for _, chunk := range fake.entry.Chunks { + switch chunk.Offset { + case 0: + original = chunk.GetFileIdString() == current.Chunks[0].GetFileIdString() && chunk.Size == 256 + case 256: + appended = chunk.GetFileIdString() == "1,01" && chunk.Size == uint64(len(payload)) + } + } + if !original || !appended { + t.Fatalf("chunk identity/range changed: %v", fake.entry.Chunks) + } + if publicationEscaped { + t.Fatal("flush escaped publication before its attribute gate opened") + } + + }) + } + } +} + +func waitAttributeFrame(frame string) bool { + deadline := time.Now().Add(10 * time.Second) + for time.Now().Before(deadline) { + buf := make([]byte, 1<<20) + n := runtime.Stack(buf, true) + if strings.Contains(string(buf[:n]), "."+frame+"(") { + return true + } + runtime.Gosched() + } + return false +} + +// Require the flush goroutine's waiting state AND both publication frames. +// Seeing a frame alone also admits a runnable waiter awakened by an early +// unlock. Broadcast makes that goroutine runnable before retention is reached, +// so the early-unlock mutant must progress to its real CreateEntry instead. +func waitFlushPublication(t *testing.T, done <-chan fuse.Status) (bool, fuse.Status) { + t.Helper() + deadline := time.Now().Add(10 * time.Second) + for time.Now().Before(deadline) { + select { + case code := <-done: + return false, code + default: + } + buf := make([]byte, 1<<20) + n := runtime.Stack(buf, true) + for _, stack := range strings.Split(string(buf[:n]), "\n\n") { + header, _, _ := strings.Cut(stack, "\n") + if strings.Contains(header, "[sync.Cond.Wait]") && + strings.Contains(stack, ".flushMetadataToFiler(") && strings.Contains(stack, ".lockMetadataPublication(") { + t.Log("flush arrived and is blocked at publication before opening gate") + return true, fuse.OK + } + } + runtime.Gosched() + } + t.Error("flush neither blocked at publication nor settled") + return false, fuse.EIO +} + +// The filer applies the update even though the caller's syscall was +// interrupted while the RPC was in flight. +type interruptingAttributeFiler struct { + *attributeCASFiler + once sync.Once + interrupt chan struct{} +} + +func (s *interruptingAttributeFiler) UpdateEntry(ctx context.Context, req *filer_pb.UpdateEntryRequest) (*filer_pb.UpdateEntryResponse, error) { + s.once.Do(func() { close(s.interrupt) }) + select { + case <-ctx.Done(): + case <-time.After(100 * time.Millisecond): + } + return s.attributeCASFiler.UpdateEntry(context.Background(), req) +} + +func TestAttributeSaveSurvivesInterrupt(t *testing.T) { + for _, op := range []string{"chmod", "setxattr"} { + t.Run(op, func(t *testing.T) { + wfs := newInvalidateTestWFS(t) + wfs.atimeMap = make(map[uint64]time.Time) + current := attributeAppendEntry(300) + fake := &interruptingAttributeFiler{attributeCASFiler: &attributeCASFiler{entry: proto.Clone(current).(*filer_pb.Entry)}, interrupt: make(chan struct{})} + startFakeFiler(t, wfs, fake) + path := util.FullPath("/dir/file") + wfs.inodeToPath.Lookup(util.FullPath("/dir"), 1, true, false, 0, false) + wfs.inodeToPath.Lookup(path, 1, false, false, 42, true) + wfs.inodeToPath.MarkChildrenCached(util.FullPath("/dir")) + if err := wfs.metaCache.InsertEntry(context.Background(), filer.FromPbEntry("/dir", current), 1); err != nil { + t.Fatal(err) + } + var code fuse.Status + if op == "setxattr" { + in := &fuse.SetXAttrIn{} + in.NodeId = 42 + code = wfs.SetXAttr(fake.interrupt, in, "user.test", []byte("new")) + } else { + in := &fuse.SetAttrIn{} + in.NodeId = 42 + in.Valid, in.Mode = fuse.FATTR_MODE, 0600 + code = wfs.SetAttr(fake.interrupt, in, &fuse.AttrOut{}) + } + if code != fuse.OK { + t.Fatalf("interrupted %s reported %v for an update the filer applied", op, code) + } + cached, _, err := wfs.metaCache.FindEntry(context.Background(), path) + if err != nil { + t.Fatal(err) + } + if op == "chmod" && cached.Attr.Mode.Perm() != 0600 { + t.Fatalf("local cache mode %v, want 0600", cached.Attr.Mode.Perm()) + } + if op == "setxattr" && string(cached.Extended["xattr-user.test"]) != "new" { + t.Fatal("local cache missing the saved xattr") + } + }) + } +} + +// The test mount maps local uid 1000 to filer uid 2000, so a file the filer +// records as uid 1000 reads back as local 1000 and maps forward to 2000. +func TestAttributeSaveKeepsCollidingFilerIds(t *testing.T) { + for _, op := range []string{"chmod", "setxattr"} { + t.Run(op, func(t *testing.T) { + wfs := newInvalidateTestWFS(t) + wfs.atimeMap = make(map[uint64]time.Time) + current := attributeAppendEntry(3) + current.Attributes.Uid = 1000 + fake := &attributeCASFiler{entry: proto.Clone(current).(*filer_pb.Entry)} + startFakeFiler(t, wfs, fake) + path := util.FullPath("/dir/file") + wfs.inodeToPath.Lookup(util.FullPath("/dir"), 1, true, false, 0, false) + wfs.inodeToPath.Lookup(path, 1, false, false, 42, true) + wfs.inodeToPath.MarkChildrenCached(util.FullPath("/dir")) + if err := wfs.metaCache.InsertEntry(context.Background(), filer.FromPbEntry("/dir", current), 1); err != nil { + t.Fatal(err) + } + var code fuse.Status + if op == "setxattr" { + in := &fuse.SetXAttrIn{} + in.NodeId = 42 + code = wfs.SetXAttr(nil, in, "user.test", []byte("new")) + } else { + in := &fuse.SetAttrIn{} + in.NodeId = 42 + in.Valid, in.Mode = fuse.FATTR_MODE, 0600 + code = wfs.SetAttr(nil, in, &fuse.AttrOut{}) + } + if code != fuse.OK { + t.Fatalf("%s on a colliding uid: %v", op, code) + } + fake.mu.Lock() + defer fake.mu.Unlock() + if fake.entry.Attributes.Uid != 1000 { + t.Fatalf("%s rewrote the filer uid to %d", op, fake.entry.Attributes.Uid) + } + cached, _, err := wfs.metaCache.FindEntry(context.Background(), path) + if err != nil { + t.Fatal(err) + } + if cached.Attr.Uid != 1000 { + t.Fatalf("local cache uid %d, want 1000", cached.Attr.Uid) + } + }) + } +} + +// The seed filer has not applied the owner's latest update yet and forwards +// conditional writes to the owner, as filers do under the lock ring. +type laggingSeedFiler struct { + filer_pb.UnimplementedSeaweedFilerServer + stale *filer_pb.Entry + owner *attributeCASFiler +} + +func (s *laggingSeedFiler) LookupDirectoryEntry(ctx context.Context, req *filer_pb.LookupDirectoryEntryRequest) (*filer_pb.LookupDirectoryEntryResponse, error) { + return &filer_pb.LookupDirectoryEntryResponse{Entry: proto.Clone(s.stale).(*filer_pb.Entry)}, nil +} + +func (s *laggingSeedFiler) UpdateEntry(ctx context.Context, req *filer_pb.UpdateEntryRequest) (*filer_pb.UpdateEntryResponse, error) { + return s.owner.UpdateEntry(ctx, req) +} + +func TestAttributeRetryReadsOwnerFiler(t *testing.T) { + wfs := newInvalidateTestWFS(t) + wfs.atimeMap = make(map[uint64]time.Time) + stale := attributeAppendEntry(1) + owner := &attributeCASFiler{entry: attributeAppendEntry(300)} + startFakeFiler(t, wfs, owner) + ownerAddress := wfs.option.FilerAddresses[0] + startFakeFiler(t, wfs, &laggingSeedFiler{stale: stale, owner: owner}) + wfs.lockClient = cluster.NewLockClient(nil, wfs.option.FilerAddresses[0]) + wfs.lockClient.SetRing([]pb.ServerAddress{ownerAddress}, 1) + + path := util.FullPath("/dir/file") + wfs.inodeToPath.Lookup(util.FullPath("/dir"), 1, true, false, 0, false) + wfs.inodeToPath.Lookup(path, 1, false, false, 42, true) + wfs.inodeToPath.MarkChildrenCached(util.FullPath("/dir")) + if err := wfs.metaCache.InsertEntry(context.Background(), filer.FromPbEntry("/dir", stale), 1); err != nil { + t.Fatal(err) + } + in := &fuse.SetAttrIn{} + in.NodeId = 42 + in.Valid, in.Mode = fuse.FATTR_MODE, 0600 + if code := wfs.SetAttr(nil, in, &fuse.AttrOut{}); code != fuse.OK { + t.Fatalf("chmod behind a lagging seed filer: %v", code) + } + owner.mu.Lock() + defer owner.mu.Unlock() + if len(owner.entry.Chunks) != 300 || owner.entry.Attributes.FileMode&0777 != 0600 { + t.Fatalf("owner entry: %d chunks, mode %o", len(owner.entry.Chunks), owner.entry.Attributes.FileMode) + } +} diff --git a/weed/mount/weedfs_attr_race_test.go b/weed/mount/weedfs_attr_race_test.go index 805e4261b3..73374efd18 100644 --- a/weed/mount/weedfs_attr_race_test.go +++ b/weed/mount/weedfs_attr_race_test.go @@ -45,6 +45,9 @@ func TestAttrChunkRace(t *testing.T) { wfs: wfs, entry: &LockedEntry{Entry: entry}, entryChunkGroup: chunkGroup, + // The upload worker below represents pending local content writes, + // so metadata remains on the handle until their normal flush. + dirtyMetadata: true, } fh.dirtyPages = newPageWriter(fh, 1<<20) wfs.fhMap.inode2fh[inode] = fh diff --git a/weed/mount/weedfs_file_sync.go b/weed/mount/weedfs_file_sync.go index a7d5c8eea5..89f5652858 100644 --- a/weed/mount/weedfs_file_sync.go +++ b/weed/mount/weedfs_file_sync.go @@ -206,6 +206,11 @@ func (wfs *WFS) doFlush(ctx context.Context, fh *FileHandle, uid, gid uint32, al // needed here. The local fhLockTable lock below serializes within this mount. func (wfs *WFS) flushMetadataToFiler(ctx context.Context, fh *FileHandle, dir, name string, uid, gid uint32) error { fileFullPath := fh.FullPath() + // Attribute acknowledgement and local retention must complete before a + // whole-entry flush can publish. Take this before the handle's active lock. + unlock := wfs.lockMetadataPublication(fileFullPath) + defer unlock() + glog.V(4).Infof("flushMetadataToFiler %s/%s inode %d fh %d", dir, name, fh.inode, fh.fh) fhActiveLock := fh.wfs.fhLockTable.AcquireLock("doFlush", fh.fh, util.ExclusiveLock) diff --git a/weed/mount/weedfs_filehandle.go b/weed/mount/weedfs_filehandle.go index 16220d26c4..1809f53b88 100644 --- a/weed/mount/weedfs_filehandle.go +++ b/weed/mount/weedfs_filehandle.go @@ -115,9 +115,22 @@ func (wfs *WFS) AcquireHandle(ctx context.Context, inode uint64, flags, uid, gid } // lookupFilerEntry reads path from the filer, bypassing the local cache, in -// local ids and with the log position the answer reflects. It asks the filer -// that checks path's write conditions, since another filer may lag it. +// local ids and with the log position the answer reflects. func (wfs *WFS) lookupFilerEntry(ctx context.Context, path util.FullPath) (*filer_pb.Entry, entryVersion, error) { + entry, version, err := wfs.lookupOwnerEntry(ctx, path) + if err != nil { + return nil, entryVersion{}, err + } + if entry.Attributes == nil { + entry.Attributes = &filer_pb.FuseAttributes{} + } + wfs.mapPbIdFromFilerToLocal(entry) + return entry, version, nil +} + +// lookupOwnerEntry reads path in filer ids from the filer that checks path's +// write conditions, since another filer may lag it. +func (wfs *WFS) lookupOwnerEntry(ctx context.Context, path util.FullPath) (*filer_pb.Entry, entryVersion, error) { dir, name := path.DirAndName() var entry *filer_pb.Entry var version entryVersion @@ -138,10 +151,6 @@ func (wfs *WFS) lookupFilerEntry(ctx context.Context, path util.FullPath) (*file if err != nil { return nil, entryVersion{}, err } - if entry.Attributes == nil { - entry.Attributes = &filer_pb.FuseAttributes{} - } - wfs.mapPbIdFromFilerToLocal(entry) return entry, version, nil } diff --git a/weed/mount/weedfs_xattr.go b/weed/mount/weedfs_xattr.go index da3815ccbd..a84df3ae3d 100644 --- a/weed/mount/weedfs_xattr.go +++ b/weed/mount/weedfs_xattr.go @@ -8,6 +8,8 @@ import ( "syscall" "github.com/seaweedfs/go-fuse/v2/fuse" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/util" ) const ( @@ -116,30 +118,31 @@ func (wfs *WFS) SetXAttr(cancel <-chan struct{}, input *fuse.SetXAttrIn, attr st if entry == nil { return fuse.ENOENT } + if path != "" { + unlock := wfs.lockMetadataPublication(path) + defer unlock() + } + if snapshot, ok := wfs.attributeEntrySnapshot(fh, entry); ok && path != "" { + _, code := wfs.saveEntryAttributes(ctx, path, snapshot, 0, func(next *filer_pb.Entry) fuse.Status { + return applyXAttr(next, input.Flags, attr, data) + }, func(live, acked *filer_pb.Entry) { + if live.Extended == nil { + live.Extended = make(map[string][]byte) + } + live.Extended[XATTR_PREFIX+attr] = append([]byte(nil), acked.Extended[XATTR_PREFIX+attr]...) + }) + return code + } if fh != nil { + active := wfs.fhLockTable.AcquireLock("xattr", fh.fh, util.ExclusiveLock) + defer wfs.fhLockTable.ReleaseLock(fh.fh, active) fh.entryLock.Lock() defer fh.entryLock.Unlock() } - if entry.Extended == nil { - entry.Extended = make(map[string][]byte) + if code := applyXAttr(entry, input.Flags, attr, data); code != fuse.OK { + return code } - _, exists := entry.Extended[XATTR_PREFIX+attr] - switch input.Flags { - case xattr_CREATE: - if exists { - return fuse.Status(syscall.EEXIST) - } - case xattr_REPLACE: - if !exists { - return fuse.ENODATA - } - } - - // data aliases the FUSE request's pooled input buffer, which is - // recycled once this handler returns. Copy before storing so a - // later request reusing the buffer cannot corrupt the value. - entry.Extended[XATTR_PREFIX+attr] = append([]byte(nil), data...) if fh != nil { fh.dirtyMetadata = true @@ -212,21 +215,28 @@ func (wfs *WFS) RemoveXAttr(cancel <-chan struct{}, header *fuse.InHeader, attr if entry == nil { return fuse.OK } + if path != "" { + unlock := wfs.lockMetadataPublication(path) + defer unlock() + } + if snapshot, ok := wfs.attributeEntrySnapshot(fh, entry); ok && path != "" { + _, code := wfs.saveEntryAttributes(ctx, path, snapshot, 0, func(next *filer_pb.Entry) fuse.Status { + return removeXAttr(next, attr) + }, func(live, _ *filer_pb.Entry) { + delete(live.Extended, XATTR_PREFIX+attr) + }) + return code + } if fh != nil { + active := wfs.fhLockTable.AcquireLock("xattr", fh.fh, util.ExclusiveLock) + defer wfs.fhLockTable.ReleaseLock(fh.fh, active) fh.entryLock.Lock() defer fh.entryLock.Unlock() } - if entry.Extended == nil { - return fuse.ENOATTR + if code := removeXAttr(entry, attr); code != fuse.OK { + return code } - _, found := entry.Extended[XATTR_PREFIX+attr] - - if !found { - return fuse.ENOATTR - } - - delete(entry.Extended, XATTR_PREFIX+attr) if fh != nil { fh.dirtyMetadata = true @@ -240,3 +250,35 @@ func (wfs *WFS) RemoveXAttr(cancel <-chan struct{}, header *fuse.InHeader, attr return wfs.saveEntry(path, entry) } + +func removeXAttr(entry *filer_pb.Entry, attr string) fuse.Status { + if _, found := entry.Extended[XATTR_PREFIX+attr]; !found { + return fuse.ENOATTR + } + delete(entry.Extended, XATTR_PREFIX+attr) + return fuse.OK +} + +func applyXAttr(entry *filer_pb.Entry, flags uint32, attr string, data []byte) fuse.Status { + if entry.Extended == nil { + entry.Extended = make(map[string][]byte) + } + _, exists := entry.Extended[XATTR_PREFIX+attr] + switch flags { + case xattr_CREATE: + if exists { + return fuse.Status(syscall.EEXIST) + } + case xattr_REPLACE: + if !exists { + return fuse.ENODATA + } + } + + // data aliases the FUSE request's pooled input buffer, which is + // recycled once this handler returns. Copy before storing so a + // later request reusing the buffer cannot corrupt the value. + entry.Extended[XATTR_PREFIX+attr] = append([]byte(nil), data...) + + return fuse.OK +} diff --git a/weed/mount/weedfs_xattr_test.go b/weed/mount/weedfs_xattr_test.go index 035553c277..da0fa0b330 100644 --- a/weed/mount/weedfs_xattr_test.go +++ b/weed/mount/weedfs_xattr_test.go @@ -34,6 +34,9 @@ func TestSetXAttrCopiesValueBuffer(t *testing.T) { }, }, 0, 0) fh.RememberPath(path) + // Model a newly copied file whose data has not reached the filer yet; + // this is the deferred persistence path whose buffer ownership we test. + fh.dirtyMetadata = true // Caller buffer aliases a pool that the kernel will overwrite on // the very next request. SetXAttr must defensively copy. diff --git a/weed/mount/wfs_save.go b/weed/mount/wfs_save.go index e66712830a..8c30e37659 100644 --- a/weed/mount/wfs_save.go +++ b/weed/mount/wfs_save.go @@ -53,6 +53,13 @@ func (wfs *WFS) saveEntry(path util.FullPath, entry *filer_pb.Entry) (code fuse. return fuseStatus } + wfs.applySavedEntry(path, entry, resp) + return fuse.OK +} + +// applySavedEntry publishes an acknowledged entry with filer-side uid/gid. +func (wfs *WFS) applySavedEntry(path util.FullPath, entry *filer_pb.Entry, resp *filer_pb.UpdateEntryResponse) { + parentDir, _ := path.DirAndName() // The mutation is acknowledged; bring any open handle for this path up // to the acknowledged state — a handle opened while this save was in // flight holds an older entry, and advancing its version alone would @@ -79,7 +86,6 @@ func (wfs *WFS) saveEntry(path util.FullPath, entry *filer_pb.Entry) (code fuse. wfs.inodeToPath.InvalidateChildrenCache(util.FullPath(parentDir)) } - return fuse.OK } func (wfs *WFS) mapPbIdFromFilerToLocal(entry *filer_pb.Entry) { diff --git a/weed/mount/wfs_save_attrs.go b/weed/mount/wfs_save_attrs.go new file mode 100644 index 0000000000..50e58c96b4 --- /dev/null +++ b/weed/mount/wfs_save_attrs.go @@ -0,0 +1,164 @@ +package mount + +import ( + "context" + "errors" + "syscall" + + "github.com/seaweedfs/go-fuse/v2/fuse" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/util" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + "google.golang.org/protobuf/proto" +) + +// A clean open handle must not become a dirty whole-entry write merely from +// chmod or setxattr. Pending content writes and deferred creates still use the +// handle's existing flush path, which owns their unsaved data. +func (wfs *WFS) attributeEntrySnapshot(fh *FileHandle, entry *filer_pb.Entry) (*filer_pb.Entry, bool) { + if fh == nil { + return entry, true + } + // Match Read: the handle's active lock precedes entryLock. + lock := wfs.fhLockTable.AcquireLock("attributeEntrySnapshot", fh.fh, util.SharedLock) + defer wfs.fhLockTable.ReleaseLock(fh.fh, lock) + fh.entryLock.Lock() + defer fh.entryLock.Unlock() + if fh.dirtyMetadata { + return nil, false + } + fh.entry.RLock() + defer fh.entry.RUnlock() + return proto.Clone(fh.entry.Entry).(*filer_pb.Entry), true +} + +// saveEntryAttributes retries only an entry comparison miss, always replaying +// the syscall's attribute change on the new entry. In particular, the cached +// chunks, size, inline content and unrelated attributes never win a conflict. +func (wfs *WFS) saveEntryAttributes(ctx context.Context, path util.FullPath, base *filer_pb.Entry, idFields uint32, change func(*filer_pb.Entry) fuse.Status, retain func(live, acked *filer_pb.Entry)) (*filer_pb.Entry, fuse.Status) { + // Like the other metadata mutations, ignore a FUSE interrupt: the owner + // may already have applied the update the caller would be told failed. + ctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), metadataFlushTimeout) + defer cancel() + dir, _ := path.DirAndName() + inode := base.GetAttributes().GetInode() + localUid, localGid := base.GetAttributes().GetUid(), base.GetAttributes().GetGid() + // The snapshot has local ids only. Where the id mapping is not one-to-one + // the filer ids derived from them can miss; the re-read supplies the real ones. + filerUid, filerGid := wfs.option.UidGidMapper.LocalToFiler(localUid, localGid) + for attempt := 0; attempt < 3; attempt++ { + if attempt > 0 { + // Bypass the mount cache: it supplied the snapshot just refused. + entry, _, err := wfs.lookupOwnerEntry(ctx, path) + if errors.Is(err, filer_pb.ErrNotFound) { + return nil, fuse.ENOENT + } + if err != nil { + return nil, grpcErrorToFuseStatus(err) + } + base = entry + filerUid, filerGid = base.GetAttributes().GetUid(), base.GetAttributes().GetGid() + wfs.mapPbIdFromFilerToLocal(base) + if inode != 0 && base.GetAttributes().GetInode() != inode { + return nil, fuse.Status(syscall.ESTALE) + } + } + next := proto.Clone(base).(*filer_pb.Entry) + if code := change(next); code != fuse.OK { + return nil, code + } + // Whole-entry clones are as large as the chunk list; switch ids in place. + // Only fields absent from the syscall keep the filer's raw ids. + // Equal local values still express an explicit chown/chgrp intent. + toFiler := func(uid, gid uint32) (uint32, uint32) { + mappedUid, mappedGid := wfs.option.UidGidMapper.LocalToFiler(uid, gid) + if idFields&fuse.FATTR_UID == 0 { + mappedUid = filerUid + } + if idFields&fuse.FATTR_GID == 0 { + mappedGid = filerGid + } + return mappedUid, mappedGid + } + restoreBase := mapIdsToFiler(base, func(_, _ uint32) (uint32, uint32) { return filerUid, filerGid }) + restoreNext := mapIdsToFiler(next, toFiler) + req := &filer_pb.UpdateEntryRequest{Directory: dir, Entry: next, Condition: &filer_pb.WriteCondition{Clauses: []*filer_pb.WriteCondition_Clause{{Kind: filer_pb.WriteCondition_IF_ENTRY_EQUAL, ExpectedEntry: base}}}, Signatures: []int32{wfs.signature}} + var resp *filer_pb.UpdateEntryResponse + // The stream currently reduces FailedPrecondition to EIO. Use the unary + // RPC so a comparison miss remains distinguishable from an I/O failure. + err := wfs.withFilerClient(ctx, false, wfs.filerRPCWait(), func(ctx context.Context, client filer_pb.SeaweedFilerClient) error { + var err error + resp, err = client.UpdateEntry(ctx, req) + return err + }) + restoreBase() + if status.Code(err) == codes.FailedPrecondition { + continue + } + if err != nil { + return nil, grpcErrorToFuseStatus(err) + } + wfs.applySavedEntry(path, next, resp) + restoreNext() + wfs.retainAcknowledgedAttributes(path, next, retain) + return next, fuse.OK + } + return nil, fuse.Status(syscall.EAGAIN) +} + +// mapIdsToFiler maps entry's uid/gid to the filer's and returns a func that +// puts the local ids back exactly; mapping back can differ when the ids collide. +func mapIdsToFiler(entry *filer_pb.Entry, toFiler func(uid, gid uint32) (uint32, uint32)) (restore func()) { + attr := entry.GetAttributes() + if attr == nil { + return func() {} + } + uid, gid := attr.Uid, attr.Gid + attr.Uid, attr.Gid = toFiler(uid, gid) + return func() { attr.Uid, attr.Gid = uid, gid } +} + +// A background WRITE may dirty the handle while the attribute RPC is in +// flight. installAckedEntry intentionally leaves its content alone. Retain +// only the acknowledged attribute delta there, so the next content flush +// cannot undo a successful syscall. Do not recheck CREATE/REPLACE or WORM: +// those preconditions were already validated by the acknowledged mutation. +func (wfs *WFS) retainAcknowledgedAttributes(path util.FullPath, acked *filer_pb.Entry, retain func(live, acked *filer_pb.Entry)) { + inode, found := wfs.inodeToPath.GetInode(path) + if !found { + return + } + fh, found := wfs.fhMap.FindFileHandle(inode) + if !found { + return + } + // Match Read: the handle's active lock precedes entryLock. + lock := wfs.fhLockTable.AcquireLock("retainAcknowledgedAttributes", fh.fh, util.ExclusiveLock) + defer wfs.fhLockTable.ReleaseLock(fh.fh, lock) + fh.entryLock.Lock() + defer fh.entryLock.Unlock() + if !fh.dirtyMetadata || fh.isDeleted { + return + } + fh.entry.Lock() + defer fh.entry.Unlock() + live := fh.entry.Entry + if acked.GetAttributes().GetInode() != 0 && live.GetAttributes().GetInode() != acked.GetAttributes().GetInode() { + return + } + retain(live, acked) +} + +// A path lock covers the owner RPC through local installation/retention, and +// whole-entry metadata publication. WRITE and chunk upload remain independent. +// Keying by path also covers a handle opened while the RPC is in flight. +func (wfs *WFS) metadataPublicationTable() *util.LockTable[util.FullPath] { + wfs.metadataPublicationOnce.Do(func() { wfs.metadataPublicationLocks = util.NewLockTable[util.FullPath]() }) + return wfs.metadataPublicationLocks +} +func (wfs *WFS) lockMetadataPublication(path util.FullPath) func() { + table := wfs.metadataPublicationTable() + lock := table.AcquireLock("metadata publication", path, util.ExclusiveLock) + return func() { table.ReleaseLock(path, lock) } +}