fix(mount): preserve appended chunks during stale attribute updates (#12593)

* fix(mount): preserve acknowledged chunks during attribute updates

* mount: keep the WFS field block aligned

Adding the publication-lock fields inside the aligned block made gofmt
re-indent 25 unrelated lines. Put them in their own group after
dirIdleEvict, matching the enterprise layout, so syncs do not conflict there.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* mount: finish attribute saves through a FUSE interrupt

saveEntryAttributes derived its RPC context from the syscall's cancel
channel, so an interrupt while the conditional UpdateEntry was at the
owner returned EIO for an update the owner then applied, and skipped the
local install and retention. Keep the metadata deadline but drop the
cancellation, like the other metadata mutations.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* mount: map attribute CAS ids in place instead of cloning the entry

saveEntryAttributes cloned the whole entry three times per attempt only
to rewrite uid/gid for the filer, and each clone is as large as the
chunk list. Switch the ids in place and restore the exact local values
after the call, keeping one clone. The test now checks the reply carries
the local uid.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* mount: compare attribute updates against the filer's own uid/gid

With -map.uid/-map.gid, a filer id that is also a mapped local id reads
back unchanged and maps forward to another filer id. The IF_ENTRY_EQUAL
expected entry then never matched, so chmod, utimes and xattr changes on
such files failed with EAGAIN after three attempts, and an update would
have rewritten the owner. Keep the filer ids read with the entry for the
ids the syscall leaves alone.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* fix(mount): align attribute retention with read lock order

* mount: take the handle lock before entryLock in the attribute snapshot

attributeEntrySnapshot held entryLock while it queued for the shared
handle lock. A Read holding that lock waits on entryLock, the snapshot
waits behind a queued exclusive (an invalidation or a Write), and the
exclusive waits on the Read.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* fix(mount): complete handle lock order and preserve explicit ID intent

* mount: re-read a refused attribute update from the file's owner filer

The filer checks the IF_ENTRY_EQUAL condition on the owner of the path's
write route key. The retry read any filer, so behind one that had not
applied the owner's latest update, every attempt sent the same stale base
and the syscall failed with EAGAIN. Read from the owner once the lock ring
is known, like lookupFilerEntry, but keep the filer's raw ids.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Chris Lu <chrislusf@users.noreply.github.com>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

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

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

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

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

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) {

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

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

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)

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
 }
 

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

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.

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) {

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