diff --git a/weed/cluster/cluster.go b/weed/cluster/cluster.go
index 7436474dce..2c47d3a2c4 100644
--- a/weed/cluster/cluster.go
+++ b/weed/cluster/cluster.go
@@ -13,6 +13,7 @@ const (
 	VolumeServerType = "volumeServer"
 	FilerType        = "filer"
 	BrokerType       = "broker"
+	MountType        = "mount"
 	S3Type           = "s3"
 	VectorType       = "vector"
 )
diff --git a/weed/mount/content_flush_condition_test.go b/weed/mount/content_flush_condition_test.go
new file mode 100644
index 0000000000..4eb719a253
--- /dev/null
+++ b/weed/mount/content_flush_condition_test.go
@@ -0,0 +1,248 @@
+//go:build !freebsd
+
+package mount
+
+import (
+	"context"
+	"sync"
+	"testing"
+
+	"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"
+)
+
+type contentFlushConditionFiler struct {
+	filer_pb.UnimplementedSeaweedFilerServer
+	mu    sync.Mutex
+	entry *filer_pb.Entry
+	calls int
+}
+
+func (s *contentFlushConditionFiler) CreateEntry(ctx context.Context, r *filer_pb.CreateEntryRequest) (*filer_pb.CreateEntryResponse, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.calls++
+	for _, c := range r.GetCondition().GetClauses() {
+		matches := false
+		switch c.Kind {
+		case filer_pb.WriteCondition_IF_ENTRY_EQUAL:
+			matches = proto.Equal(c.ExpectedEntry, s.entry)
+		case filer_pb.WriteCondition_IF_NOT_EXISTS:
+			matches = s.entry == nil
+		}
+		if !matches {
+			return &filer_pb.CreateEntryResponse{Error: "precondition failed", ErrorCode: filer_pb.FilerError_PRECONDITION_FAILED}, nil
+		}
+	}
+	// Like the filer, stamp the stored version and announce what was stored.
+	old := s.entry
+	s.entry = proto.Clone(r.Entry).(*filer_pb.Entry)
+	s.entry.CommitTsNs = 1000 + int64(s.calls)
+	return &filer_pb.CreateEntryResponse{LogTsNs: 2000, MetadataEvent: &filer_pb.SubscribeMetadataResponse{
+		Directory: r.Directory,
+		TsNs:      s.entry.CommitTsNs,
+		EventNotification: &filer_pb.EventNotification{
+			OldEntry: old,
+			NewEntry: proto.Clone(s.entry).(*filer_pb.Entry),
+		},
+	}}, nil
+}
+func (s *contentFlushConditionFiler) LookupDirectoryEntry(ctx context.Context, r *filer_pb.LookupDirectoryEntryRequest) (*filer_pb.LookupDirectoryEntryResponse, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if s.entry == nil {
+		return nil, status.Error(codes.NotFound, filer_pb.ErrNotFound.Error())
+	}
+	return &filer_pb.LookupDirectoryEntryResponse{Entry: proto.Clone(s.entry).(*filer_pb.Entry)}, nil
+}
+
+func flushConditionEntry(ids ...uint64) *filer_pb.Entry {
+	e := &filer_pb.Entry{Name: "file", Attributes: &filer_pb.FuseAttributes{Inode: 42, FileMode: 0644, Uid: 2000, Gid: 20, FileSize: 3}}
+	for i, id := range ids {
+		e.Chunks = append(e.Chunks, &filer_pb.FileChunk{Fid: &filer_pb.FileId{VolumeId: 1, FileKey: id, Cookie: 1}, Offset: int64(i), Size: 1, ModifiedTsNs: int64(id)})
+	}
+	return e
+}
+
+// Exercise the actual mount content-publication caller. A peer's successful
+// append or replacement is authoritative; refusing our stale flush must retain
+// our dirty entry, instead of refreshing a guard and replaying stale content.
+func TestContentFlushPreservesPeerChunks(t *testing.T) {
+	for _, name := range []string{"single-writer", "peer-append", "replaced", "deleted", "deferred-create", "deferred-create-race"} {
+		t.Run(name, func(t *testing.T) {
+			wfs := newInvalidateTestWFS(t)
+			observed := flushConditionEntry(1)
+			live := flushConditionEntry(1, 3)
+			live.Chunks[1].Offset = 2
+			current := flushConditionEntry(1)
+			wantError := false
+			switch name {
+			case "peer-append":
+				current = flushConditionEntry(1, 2)
+				wantError = true
+			case "replaced":
+				current = flushConditionEntry(4)
+				current.Attributes.Inode = 99
+				wantError = true
+			case "deleted":
+				current = nil
+				wantError = true
+			case "deferred-create":
+				observed = nil
+				current = nil
+			case "deferred-create-race":
+				observed = nil
+				current = flushConditionEntry(4)
+				current.Attributes.Inode = 99
+				wantError = true
+			}
+			if observed != nil {
+				wfs.mapPbIdFromFilerToLocal(observed)
+			}
+			wfs.mapPbIdFromFilerToLocal(live)
+			fake := &contentFlushConditionFiler{entry: current}
+			startFakeFiler(t, wfs, fake)
+			fh := newFileHandle(wfs, 1, 42, live)
+			t.Cleanup(func() { fh.dirtyPages.Destroy() })
+			fh.baseEntry.Store(observed)
+			fh.dirtyMetadata = true
+			fh.RememberPath(util.FullPath("/dir/file"))
+			before := proto.Clone(current)
+			code := wfs.doFlush(context.Background(), fh, 1000, 20, false)
+			fake.mu.Lock()
+			defer fake.mu.Unlock()
+			if wantError {
+				if code == fuse.OK {
+					t.Fatal("stale content flush acknowledged after a peer changed the entry")
+				}
+				if !proto.Equal(fake.entry, before) {
+					t.Fatalf("peer entry changed: %v", fake.entry)
+				}
+				if !fh.dirtyMetadata {
+					t.Fatal("refused flush lost dirty metadata")
+				}
+				if len(fh.entry.Entry.Chunks) != 2 || fh.entry.Entry.Chunks[1].GetFileIdString() != live.Chunks[1].GetFileIdString() {
+					t.Fatal("refused flush lost pending content")
+				}
+			} else {
+				if code != fuse.OK {
+					t.Fatal(code)
+				}
+				if fh.dirtyMetadata {
+					t.Fatal("successful flush remains dirty")
+				}
+				if len(fake.entry.Chunks) != 2 || fake.entry.Chunks[1].Offset != 2 || fake.entry.Chunks[1].Size != 1 {
+					t.Fatalf("normal flush lost content: %v", fake.entry)
+				}
+			}
+			if fake.calls != 1 {
+				t.Fatalf("unexpected replay: %d calls", fake.calls)
+			}
+		})
+	}
+}
+
+// Each acknowledged flush must leave the handle a base the next flush on the
+// same handle can match: the stored entry, with the filer's own commit stamp.
+func TestContentFlushRepeatsOnOneHandle(t *testing.T) {
+	wfs := newInvalidateTestWFS(t)
+	stored := flushConditionEntry(1)
+	stored.CommitTsNs = 7
+	fake := &contentFlushConditionFiler{entry: proto.Clone(stored).(*filer_pb.Entry)}
+	startFakeFiler(t, wfs, fake)
+	observed := proto.Clone(stored).(*filer_pb.Entry)
+	wfs.mapPbIdFromFilerToLocal(observed)
+	fh := newFileHandle(wfs, 1, 42, proto.Clone(observed).(*filer_pb.Entry))
+	t.Cleanup(func() { fh.dirtyPages.Destroy() })
+	fh.RememberPath(util.FullPath("/dir/file"))
+	for i := uint64(2); i <= 3; i++ {
+		fh.UpdateEntry(func(e *filer_pb.Entry) {
+			e.Chunks = append(e.Chunks, &filer_pb.FileChunk{Fid: &filer_pb.FileId{VolumeId: 1, FileKey: i, Cookie: 1}, Offset: int64(i - 1), Size: 1, ModifiedTsNs: int64(i)})
+			e.Attributes.FileSize = i
+		})
+		fh.dirtyMetadata = true
+		if code := wfs.doFlush(context.Background(), fh, 1000, 20, false); code != fuse.OK {
+			t.Fatalf("flush %d: %v", i-1, code)
+		}
+	}
+	fake.mu.Lock()
+	defer fake.mu.Unlock()
+	if len(fake.entry.Chunks) != 3 || fake.calls != 2 {
+		t.Fatalf("stored %d chunks after %d calls", len(fake.entry.Chunks), fake.calls)
+	}
+}
+
+// The periodic metadata flush publishes the same whole entry, so it must not
+// overwrite a peer's append either.
+func TestPeriodicMetadataFlushPreservesPeerChunks(t *testing.T) {
+	wfs := newInvalidateTestWFS(t)
+	current := flushConditionEntry(1, 2)
+	fake := &contentFlushConditionFiler{entry: proto.Clone(current).(*filer_pb.Entry)}
+	startFakeFiler(t, wfs, fake)
+	observed := flushConditionEntry(1)
+	wfs.mapPbIdFromFilerToLocal(observed)
+	live := flushConditionEntry(1, 3)
+	live.Chunks[1].Offset = 2
+	wfs.mapPbIdFromFilerToLocal(live)
+	fh := newFileHandle(wfs, 1, 42, live)
+	t.Cleanup(func() { fh.dirtyPages.Destroy() })
+	fh.baseEntry.Store(observed)
+	fh.dirtyMetadata = true
+	fh.RememberPath(util.FullPath("/dir/file"))
+	if err := wfs.flushFileMetadata(fh); err == nil {
+		t.Fatal("periodic flush overwrote a peer's append")
+	}
+	fake.mu.Lock()
+	defer fake.mu.Unlock()
+	if !proto.Equal(fake.entry, current) || fake.calls != 1 {
+		t.Fatalf("peer entry changed after %d calls: %v", fake.calls, fake.entry)
+	}
+	if !fh.dirtyMetadata {
+		t.Fatal("refused periodic flush lost dirty metadata")
+	}
+}
+
+// A miss on state no peer changed is republished once against the stored
+// entry: a rename while open, or a filer uid the local id mapping cannot
+// reproduce (newInvalidateTestWFS maps local 1000 to filer 2000).
+func TestContentFlushRepublishesUnchangedEntry(t *testing.T) {
+	for _, name := range []string{"renamed", "colliding-uid"} {
+		t.Run(name, func(t *testing.T) {
+			wfs := newInvalidateTestWFS(t)
+			stored := flushConditionEntry(1)
+			stored.CommitTsNs = 7
+			observed := proto.Clone(stored).(*filer_pb.Entry)
+			path := util.FullPath("/dir/file")
+			switch name {
+			case "renamed":
+				wfs.mapPbIdFromFilerToLocal(observed)
+				stored.Name, stored.CommitTsNs = "moved", 8
+				path = "/dir/moved"
+			case "colliding-uid":
+				stored.Attributes.Uid, observed.Attributes.Uid = 1000, 1000
+			}
+			fake := &contentFlushConditionFiler{entry: proto.Clone(stored).(*filer_pb.Entry)}
+			startFakeFiler(t, wfs, fake)
+			live := proto.Clone(observed).(*filer_pb.Entry)
+			live.Name = stored.Name
+			live.Chunks = append(live.Chunks, &filer_pb.FileChunk{Fid: &filer_pb.FileId{VolumeId: 1, FileKey: 3, Cookie: 1}, Offset: 1, Size: 1, ModifiedTsNs: 3})
+			fh := newFileHandle(wfs, 1, 42, live)
+			t.Cleanup(func() { fh.dirtyPages.Destroy() })
+			fh.baseEntry.Store(observed)
+			fh.dirtyMetadata = true
+			fh.RememberPath(path)
+			if code := wfs.doFlush(context.Background(), fh, 1000, 20, false); code != fuse.OK {
+				t.Fatal(code)
+			}
+			fake.mu.Lock()
+			defer fake.mu.Unlock()
+			if len(fake.entry.Chunks) != 2 || fake.calls != 2 {
+				t.Fatalf("stored %d chunks after %d calls", len(fake.entry.Chunks), fake.calls)
+			}
+		})
+	}
+}
diff --git a/weed/mount/error_classifier.go b/weed/mount/error_classifier.go
index 22e664d321..be2bb24552 100644
--- a/weed/mount/error_classifier.go
+++ b/weed/mount/error_classifier.go
@@ -1,12 +1,15 @@
 package mount
 
 import (
+	"errors"
 	"strings"
 	"syscall"
 
 	"github.com/seaweedfs/go-fuse/v2/fuse"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 )
 
 func grpcErrorToFuseStatus(err error) fuse.Status {
@@ -81,6 +84,10 @@ func isRetryableFilerError(err error) bool {
 	if err == nil {
 		return false
 	}
+	// CreateEntry reports a failed write condition in its response, not as a status.
+	if errors.Is(err, filer_pb.ErrPreconditionFailed) {
+		return false
+	}
 	if s, ok := status.FromError(err); ok {
 		switch s.Code() {
 		case codes.NotFound,
diff --git a/weed/mount/error_classifier_test.go b/weed/mount/error_classifier_test.go
index 04cb44cabb..7099265ed0 100644
--- a/weed/mount/error_classifier_test.go
+++ b/weed/mount/error_classifier_test.go
@@ -11,6 +11,8 @@ import (
 	"github.com/seaweedfs/go-fuse/v2/fuse"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 )
 
 // Regression: wfs_save.go used to wrap the gRPC error with fmt.Errorf(... %v ...)
@@ -97,6 +99,7 @@ func TestIsRetryableFilerError(t *testing.T) {
 		{"permission_denied", status.Error(codes.PermissionDenied, "no access"), false},
 		{"unauthenticated", status.Error(codes.Unauthenticated, "missing creds"), false},
 		{"failed_precondition", status.Error(codes.FailedPrecondition, "not empty"), false},
+		{"create_condition_failed", fmt.Errorf("fh flush create /f: CreateEntry /f: %w", filer_pb.ErrPreconditionFailed), false},
 		{"plain_error_retries", errors.New("random network glitch"), true},
 		{"wrapped_canceled_still_retries", fmt.Errorf("ctx: %w", status.Error(codes.Canceled, "closing")), true},
 		{"wrapped_not_found_still_skipped", fmt.Errorf("ctx: %w", status.Error(codes.NotFound, "gone")), false},
diff --git a/weed/mount/weedfs.go b/weed/mount/weedfs.go
index 2bfe531f86..47564574c8 100644
--- a/weed/mount/weedfs.go
+++ b/weed/mount/weedfs.go
@@ -469,6 +469,10 @@ func (wfs *WFS) StartBackgroundTasks() error {
 		glog.V(0).Infof("writebackCache enabled: async flush on close() for improved small file performance")
 	}
 
+	if wfs.lockClient != nil {
+		wfs.followLockRing()
+	}
+
 	follower, err := wfs.subscribeFilerConfEvents()
 	if err != nil {
 		return err
diff --git a/weed/mount/weedfs_file_mkrm.go b/weed/mount/weedfs_file_mkrm.go
index 213a97eea8..78ad8d819d 100644
--- a/weed/mount/weedfs_file_mkrm.go
+++ b/weed/mount/weedfs_file_mkrm.go
@@ -45,31 +45,7 @@ func (wfs *WFS) Create(cancel <-chan struct{}, in *fuse.CreateIn, name string, o
 	defer cancelFunc()
 	newEntry, _, code := wfs.maybeLoadEntry(ctx, entryFullPath)
 	if code == fuse.OK {
-		if newEntry == nil || newEntry.Attributes == nil {
-			return fuse.EIO
-		}
-		if in.Flags&syscall.O_EXCL != 0 {
-			glog.V(0).Infof("Create O_EXCL %s: already exists (uid=%d gid=%d mode=%o)",
-				entryFullPath, newEntry.Attributes.Uid, newEntry.Attributes.Gid, newEntry.Attributes.FileMode)
-			return fuse.Status(syscall.EEXIST)
-		}
-		inode = wfs.inodeToPath.Lookup(entryFullPath, newEntry.Attributes.Crtime, false, len(newEntry.HardLinkId) > 0, newEntry.Attributes.Inode, true)
-		fileHandle, openFh, status := wfs.AcquireHandle(ctx, inode, in.Flags, in.Uid, in.Gid)
-		if status != fuse.OK {
-			return status
-		}
-		if in.Flags&syscall.O_TRUNC != 0 && in.Flags&fuse.O_ANYWRITE != 0 {
-			if code = wfs.truncateEntry(entryFullPath, newEntry); code != fuse.OK {
-				wfs.ReleaseHandle(openFh)
-				return code
-			}
-			newEntry = fileHandle.GetEntry().GetEntry()
-		}
-
-		wfs.outputPbEntry(&out.EntryOut, inode, newEntry)
-		out.Fh = uint64(openFh)
-		out.OpenFlags = 0
-		return fuse.OK
+		return wfs.openExistingForCreate(ctx, in, entryFullPath, newEntry, out)
 	}
 	if code != fuse.ENOENT {
 		return code
@@ -86,6 +62,14 @@ func (wfs *WFS) Create(cancel <-chan struct{}, in *fuse.CreateIn, name string, o
 		if dlmLock == nil {
 			return fuse.Status(syscall.EAGAIN)
 		}
+		// A deferred create does not ask the filer, so find a file the
+		// previous lock holder created while this create waited.
+		if !wfs.option.EagerFilerCreate {
+			if peerEntry, _, err := wfs.lookupFilerEntry(ctx, entryFullPath); err == nil {
+				dlmLock.Stop()
+				return wfs.openExistingForCreate(ctx, in, entryFullPath, peerEntry, out)
+			}
+		}
 	}
 
 	inode, newEntry, code = wfs.createRegularFile(ctx, dirFullPath, name, in.Mode, in.Uid, in.Gid, 0, !wfs.option.EagerFilerCreate, true)
@@ -99,25 +83,7 @@ func (wfs *WFS) Create(cancel <-chan struct{}, in *fuse.CreateIn, name string, o
 		if code != fuse.OK {
 			return code
 		}
-		if newEntry == nil || newEntry.Attributes == nil {
-			return fuse.EIO
-		}
-		inode = wfs.inodeToPath.Lookup(entryFullPath, newEntry.Attributes.Crtime, false, len(newEntry.HardLinkId) > 0, newEntry.Attributes.Inode, true)
-		fileHandle, openFh, status := wfs.AcquireHandle(ctx, inode, in.Flags, in.Uid, in.Gid)
-		if status != fuse.OK {
-			return status
-		}
-		if in.Flags&syscall.O_TRUNC != 0 && in.Flags&fuse.O_ANYWRITE != 0 {
-			if code = wfs.truncateEntry(entryFullPath, newEntry); code != fuse.OK {
-				wfs.ReleaseHandle(openFh)
-				return code
-			}
-			newEntry = fileHandle.GetEntry().GetEntry()
-		}
-		wfs.outputPbEntry(&out.EntryOut, inode, newEntry)
-		out.Fh = uint64(openFh)
-		out.OpenFlags = 0
-		return fuse.OK
+		return wfs.openExistingForCreate(ctx, in, entryFullPath, newEntry, out)
 	} else if code != fuse.OK {
 		if dlmLock != nil {
 			dlmLock.Stop()
@@ -156,6 +122,12 @@ func (wfs *WFS) Create(cancel <-chan struct{}, in *fuse.CreateIn, name string, o
 	// file is closed without any writes. An eager create has already
 	// persisted the entry, so its handle starts clean.
 	fileHandle.dirtyMetadata = !wfs.option.EagerFilerCreate
+	if !wfs.option.EagerFilerCreate {
+		// This is a local placeholder, not an acknowledged filer base.
+		fileHandle.baseEntry.Store(nil)
+	} else {
+		fileHandle.setAuthoritativeBase(proto.Clone(newEntry).(*filer_pb.Entry))
+	}
 
 	// Create bypasses AcquireHandle, so attach the lock acquired above.
 	// A surviving handle may already hold one; ours is redundant then.
@@ -174,6 +146,35 @@ func (wfs *WFS) Create(cancel <-chan struct{}, in *fuse.CreateIn, name string, o
 	return fuse.OK
 }
 
+// openExistingForCreate answers a create whose name already exists: an
+// exclusive create fails, any other opens the entry, truncating it if asked.
+func (wfs *WFS) openExistingForCreate(ctx context.Context, in *fuse.CreateIn, entryFullPath util.FullPath, entry *filer_pb.Entry, out *fuse.CreateOut) fuse.Status {
+	if entry == nil || entry.Attributes == nil {
+		return fuse.EIO
+	}
+	if in.Flags&syscall.O_EXCL != 0 {
+		glog.V(0).Infof("Create O_EXCL %s: already exists (uid=%d gid=%d mode=%o)",
+			entryFullPath, entry.Attributes.Uid, entry.Attributes.Gid, entry.Attributes.FileMode)
+		return fuse.Status(syscall.EEXIST)
+	}
+	inode := wfs.inodeToPath.Lookup(entryFullPath, entry.Attributes.Crtime, false, len(entry.HardLinkId) > 0, entry.Attributes.Inode, true)
+	fileHandle, openFh, status := wfs.AcquireHandle(ctx, inode, in.Flags, in.Uid, in.Gid)
+	if status != fuse.OK {
+		return status
+	}
+	if in.Flags&syscall.O_TRUNC != 0 && in.Flags&fuse.O_ANYWRITE != 0 {
+		if code := wfs.truncateEntry(entryFullPath, entry); code != fuse.OK {
+			wfs.ReleaseHandle(openFh)
+			return code
+		}
+		entry = fileHandle.GetEntry().GetEntry()
+	}
+	wfs.outputPbEntry(&out.EntryOut, inode, entry)
+	out.Fh = uint64(openFh)
+	out.OpenFlags = 0
+	return fuse.OK
+}
+
 /** Create a file node
  *
  * This is called for creation of all non-directory, non-symlink
@@ -473,7 +474,7 @@ func (wfs *WFS) createRegularFile(ctx context.Context, dirFullPath util.FullPath
 		return 0, nil, grpcErrorToFuseStatus(err)
 	}
 
-	return inode, newEntry, fuse.OK
+	return inode, wfs.ackedBase(resp, newEntry), fuse.OK
 }
 
 // markHandleDeleted flags the inode's open handle so its flushes stop writing
diff --git a/weed/mount/weedfs_file_sync.go b/weed/mount/weedfs_file_sync.go
index de47440abe..a7d5c8eea5 100644
--- a/weed/mount/weedfs_file_sync.go
+++ b/weed/mount/weedfs_file_sync.go
@@ -3,6 +3,7 @@ package mount
 import (
 	"bytes"
 	"context"
+	"errors"
 	"fmt"
 	"io"
 	"syscall"
@@ -180,9 +181,9 @@ func (wfs *WFS) doFlush(ctx context.Context, fh *FileHandle, uid, gid uint32, al
 		return fuse.Status(syscall.ENOSPC)
 	}
 
-	if err := retryMetadataFlush(ctx, func() error {
+	if err := retryMetadataFlushIf(ctx, func() error {
 		return wfs.flushMetadataToFiler(ctx, fh, dir, name, uid, gid)
-	}, func(nextAttempt, totalAttempts int, backoff time.Duration, err error) {
+	}, isRetryableFilerError, func(nextAttempt, totalAttempts int, backoff time.Duration, err error) {
 		glog.Warningf("%v fh %d flush: retrying metadata flush (attempt %d/%d) after %v: %v",
 			fileFullPath, fh.fh, nextAttempt, totalAttempts, backoff, err)
 	}); err != nil {
@@ -275,7 +276,7 @@ func (wfs *WFS) flushMetadataToFiler(ctx context.Context, fh *FileHandle, dir, n
 	baseSnapshot := proto.Clone(requestEntry).(*filer_pb.Entry)
 	wfs.mapPbIdFromLocalToFiler(request.Entry)
 
-	resp, err := wfs.streamCreateEntry(ctx, request)
+	resp, err := wfs.createContentEntry(ctx, fh, request)
 	if err != nil {
 		glog.Errorf("fh flush create %s: %v", fileFullPath, err)
 		return fmt.Errorf("fh flush create %s: %w", fileFullPath, err)
@@ -291,7 +292,7 @@ func (wfs *WFS) flushMetadataToFiler(ctx context.Context, fh *FileHandle, dir, n
 	// The filer acknowledged this state at the event's log position (or, for
 	// a no-op create, at the response's log position); older queued
 	// subscription events must not roll the handle back.
-	fh.setAuthoritativeBase(baseSnapshot)
+	fh.setAuthoritativeBase(wfs.ackedBase(resp, baseSnapshot))
 	fh.advanceEntryVersion(ackVersionTsNs(resp), resp.GetLogSignature())
 	if applyErr := wfs.applyLocalMetadataEvent(context.Background(), event); applyErr != nil {
 		glog.Warningf("flush %s: best-effort metadata apply failed: %v", fileFullPath, applyErr)
@@ -305,6 +306,77 @@ func (wfs *WFS) flushMetadataToFiler(ctx context.Context, fh *FileHandle, dir, n
 	return err
 }
 
+// createContentEntry publishes a handle's whole entry only over the filer
+// state the handle observed. Never refresh the guard and replay a stale whole
+// entry after a conflict: that would discard a peer's acknowledged chunks.
+// Deferred creates expect no entry; existing handles compare the full base,
+// including inode identity. The unary call keeps a refused condition apart
+// from an I/O failure; the mutation stream reduces it to EIO.
+//
+// A miss is retried once, against the stored entry, only when that entry is
+// still the observed state: renamed, restamped, or encoded with filer ids the
+// local mapping cannot reproduce.
+func (wfs *WFS) createContentEntry(ctx context.Context, fh *FileHandle, request *filer_pb.CreateEntryRequest) (resp *filer_pb.CreateEntryResponse, err error) {
+	base := fh.baseEntry.Load()
+	guard := &filer_pb.WriteCondition_Clause{Kind: filer_pb.WriteCondition_IF_NOT_EXISTS}
+	if base != nil {
+		expected := proto.Clone(base).(*filer_pb.Entry)
+		wfs.mapPbIdFromLocalToFiler(expected)
+		guard = &filer_pb.WriteCondition_Clause{Kind: filer_pb.WriteCondition_IF_ENTRY_EQUAL, ExpectedEntry: expected}
+	}
+	for attempt := 0; attempt < 2; attempt++ {
+		request.Condition = &filer_pb.WriteCondition{Clauses: []*filer_pb.WriteCondition_Clause{guard}}
+		var stored *filer_pb.Entry
+		err = wfs.withFilerClient(ctx, false, wfs.filerRPCWait(), func(ctx context.Context, client filer_pb.SeaweedFilerClient) error {
+			var callErr error
+			resp, callErr = filer_pb.CreateEntryWithResponse(ctx, client, request)
+			if base == nil || attempt > 0 || !errors.Is(callErr, filer_pb.ErrPreconditionFailed) {
+				return callErr
+			}
+			if found, lookupErr := filer_pb.LookupEntry(ctx, client, &filer_pb.LookupDirectoryEntryRequest{Directory: request.Directory, Name: request.Entry.Name}); lookupErr == nil {
+				stored = found.Entry
+			}
+			return callErr
+		})
+		if stored == nil || !wfs.sameObservedEntry(stored, base) {
+			return resp, err
+		}
+		guard = &filer_pb.WriteCondition_Clause{Kind: filer_pb.WriteCondition_IF_ENTRY_EQUAL, ExpectedEntry: stored}
+	}
+	return resp, err
+}
+
+// sameObservedEntry reports whether the filer's stored entry still holds the
+// state the handle observed, ignoring what no peer write is carried by: the
+// name, the commit stamp, the id encoding and the derived file size.
+func (wfs *WFS) sameObservedEntry(stored, base *filer_pb.Entry) bool {
+	current := proto.Clone(stored).(*filer_pb.Entry)
+	wfs.mapPbIdFromFilerToLocal(current)
+	observed := proto.Clone(base).(*filer_pb.Entry)
+	for _, e := range []*filer_pb.Entry{current, observed} {
+		e.Name, e.CommitTsNs = "", 0
+		if e.Attributes != nil {
+			e.Attributes.FileSize = filer.FileSize(e)
+		}
+		filer_pb.BeforeEntrySerialization(e.Chunks)
+	}
+	return proto.Equal(current, observed)
+}
+
+// ackedBase is the base an acknowledged create leaves: the entry the filer
+// stored, in local ids. The filer stamps commit_ts_ns, crtime, inode and atime
+// and may rewrite chunks, so the requested entry would never match it again.
+// A filer that sends no event leaves the requested entry.
+func (wfs *WFS) ackedBase(resp *filer_pb.CreateEntryResponse, requested *filer_pb.Entry) *filer_pb.Entry {
+	stored := resp.GetMetadataEvent().GetEventNotification().GetNewEntry()
+	if stored == nil {
+		return requested
+	}
+	base := proto.Clone(stored).(*filer_pb.Entry)
+	wfs.mapPbIdFromFilerToLocal(base)
+	return base
+}
+
 // shouldMergeChunks reports whether the non-manifest chunks are bloated
 // enough to justify re-reading and re-uploading the file. The condition
 // is: sum of compacted chunk sizes > 2 * logical file size.
diff --git a/weed/mount/weedfs_filehandle.go b/weed/mount/weedfs_filehandle.go
index e401d20756..16220d26c4 100644
--- a/weed/mount/weedfs_filehandle.go
+++ b/weed/mount/weedfs_filehandle.go
@@ -105,10 +105,46 @@ func (wfs *WFS) AcquireHandle(ctx context.Context, inode uint64, flags, uid, gid
 			return nil, 0, fuse.Status(syscall.EAGAIN)
 		}
 		glog.V(1).Infof("DLM lock acquired for %s", path)
+		// The previous holder's flush may not have reached the local cache,
+		// and the lock orders this writer after it: take the filer's entry.
+		if entry, version, err := wfs.lookupFilerEntry(ctx, path); err == nil {
+			fileHandle.installAckedEntry(entry, version.tsNs, version.signature)
+		}
 	}
 	return fileHandle, openFh, fuse.OK
 }
 
+// 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.
+func (wfs *WFS) lookupFilerEntry(ctx context.Context, path util.FullPath) (*filer_pb.Entry, entryVersion, error) {
+	dir, name := path.DirAndName()
+	var entry *filer_pb.Entry
+	var version entryVersion
+	lookup := func(ctx context.Context, client filer_pb.SeaweedFilerClient) error {
+		resp, err := filer_pb.LookupEntry(ctx, client, &filer_pb.LookupDirectoryEntryRequest{Directory: dir, Name: name})
+		if err != nil {
+			return err
+		}
+		entry, version = resp.Entry, entryVersion{tsNs: resp.LogTsNs, signature: resp.LogSignature}
+		return nil
+	}
+	var err error
+	if owner := wfs.entryWriteOwner(path); owner != "" {
+		err = wfs.callFiler(ctx, false, wfs.filerRPCWait(), owner.ToGrpcAddress(), lookup)
+	} else {
+		err = wfs.withFilerClient(ctx, false, wfs.filerRPCWait(), lookup)
+	}
+	if err != nil {
+		return nil, entryVersion{}, err
+	}
+	if entry.Attributes == nil {
+		entry.Attributes = &filer_pb.FuseAttributes{}
+	}
+	wfs.mapPbIdFromFilerToLocal(entry)
+	return entry, version, nil
+}
+
 // ReleaseHandle is called from FUSE Release.  For handles with a pending
 // async flush, the map removal and the pendingAsyncFlush registration are
 // done under a single lock hold so that a concurrent AcquireHandle cannot
diff --git a/weed/mount/weedfs_lock_ring.go b/weed/mount/weedfs_lock_ring.go
new file mode 100644
index 0000000000..13c08b3489
--- /dev/null
+++ b/weed/mount/weedfs_lock_ring.go
@@ -0,0 +1,75 @@
+package mount
+
+import (
+	"context"
+	"strings"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/cluster"
+	"github.com/seaweedfs/seaweedfs/weed/glog"
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+	"github.com/seaweedfs/seaweedfs/weed/wdclient"
+)
+
+// lockRingWait bounds how long startup waits for the first lock ring; until
+// one arrives, locks start at the seed filer and lookups use any filer.
+const lockRingWait = 10 * time.Second
+
+// followLockRing feeds the masters' lock ring into lockClient, so locks and
+// owner reads go straight to the filer that owns a file.
+func (wfs *WFS) followLockRing() {
+	var masters []string
+	var filerGroup string
+	err := wfs.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
+		resp, err := client.GetFilerConfiguration(context.Background(), &filer_pb.GetFilerConfigurationRequest{})
+		if err != nil {
+			return err
+		}
+		masters, filerGroup = resp.Masters, resp.FilerGroup
+		return nil
+	})
+	if err != nil || len(masters) == 0 {
+		glog.Warningf("dlm: cannot follow the lock ring, masters %v: %v", masters, err)
+		return
+	}
+
+	masterClient := wdclient.NewMasterClient(wfs.option.GrpcDialOption, filerGroup, cluster.MountType, "", wfs.option.DataCenter, "",
+		*pb.ServerAddresses(strings.Join(masters, ",")).ToServiceDiscovery())
+	masterClient.SetOnLockRingUpdateFn(func(update *master_pb.LockRingUpdate) {
+		servers := make([]pb.ServerAddress, 0, len(update.Servers))
+		for _, s := range update.Servers {
+			servers = append(servers, pb.ServerAddress(s))
+		}
+		wfs.lockClient.SetRing(servers, update.Version)
+	})
+	masterClient.SetOnMasterChangeFn(func(previous, current pb.ServerAddress) {
+		wfs.lockClient.ResetRing()
+	})
+	go masterClient.KeepConnectedToMaster(context.Background())
+
+	root := wfs.option.FilerMountRootPath
+	for deadline := time.Now().Add(lockRingWait); wfs.lockClient.PrimaryForKey(root) == ""; time.Sleep(100 * time.Millisecond) {
+		if time.Now().After(deadline) {
+			glog.Warningf("dlm: no lock ring from masters %v after %v", masters, lockRingWait)
+			return
+		}
+	}
+}
+
+// entryWriteOwner is the filer that serializes path's writes per this mount's
+// ring view, matching the filer's routing: the prior owner while a rebalance
+// cools off, else the primary. "" before a ring arrives.
+func (wfs *WFS) entryWriteOwner(path util.FullPath) pb.ServerAddress {
+	if wfs.lockClient == nil {
+		return ""
+	}
+	key := s3_constants.ObjectWriteRouteKeyPrefix + string(path)
+	if prior := wfs.lockClient.PriorOwnerForKey(key); prior != "" {
+		return prior
+	}
+	return wfs.lockClient.PrimaryForKey(key)
+}
diff --git a/weed/mount/weedfs_lock_ring_test.go b/weed/mount/weedfs_lock_ring_test.go
new file mode 100644
index 0000000000..1e8e993f2d
--- /dev/null
+++ b/weed/mount/weedfs_lock_ring_test.go
@@ -0,0 +1,44 @@
+package mount
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/cluster"
+	"github.com/seaweedfs/seaweedfs/weed/cluster/lock_manager"
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+)
+
+func TestEntryWriteOwnerMatchesFilerRouting(t *testing.T) {
+	wfs := &WFS{}
+	path := util.FullPath("/dir/file")
+	if got := wfs.entryWriteOwner(path); got != "" {
+		t.Fatalf("no lock client: owner %q, want none", got)
+	}
+
+	wfs.lockClient = cluster.NewLockClient(nil, "seed:8888")
+	if got := wfs.entryWriteOwner(path); got != "" {
+		t.Fatalf("no ring yet: owner %q, want none", got)
+	}
+
+	servers := []pb.ServerAddress{"filer1:8888", "filer2:8888", "filer3:8888"}
+	wfs.lockClient.SetRing(servers, 1)
+	filerRing := lock_manager.NewLockRing(time.Hour)
+	filerRing.SetSnapshot(servers, 1)
+
+	owners := map[pb.ServerAddress]bool{}
+	for i := 0; i < 64; i++ {
+		p := util.FullPath(fmt.Sprintf("/dir/file%d", i))
+		want := filerRing.WriteOwner(s3_constants.ObjectWriteRouteKeyPrefix + string(p))
+		if got := wfs.entryWriteOwner(p); got != want {
+			t.Fatalf("%s: mount routes to %q, filer to %q", p, got, want)
+		}
+		owners[want] = true
+	}
+	if len(owners) < 2 {
+		t.Fatalf("all paths hashed to %v; the test does not tell routing keys apart", owners)
+	}
+}
diff --git a/weed/mount/weedfs_metadata_flush.go b/weed/mount/weedfs_metadata_flush.go
index bce2e9e8d9..fdcafa0ef5 100644
--- a/weed/mount/weedfs_metadata_flush.go
+++ b/weed/mount/weedfs_metadata_flush.go
@@ -179,7 +179,7 @@ func (wfs *WFS) flushFileMetadata(fh *FileHandle) error {
 	baseSnapshot := proto.Clone(requestEntry).(*filer_pb.Entry)
 	wfs.mapPbIdFromLocalToFiler(request.Entry)
 
-	resp, err := wfs.streamCreateEntry(context.Background(), request)
+	resp, err := wfs.createContentEntry(context.Background(), fh, request)
 	if err != nil {
 		return err
 	}
@@ -191,7 +191,7 @@ func (wfs *WFS) flushFileMetadata(fh *FileHandle) error {
 			event.TsNs = ackVersionTsNs(resp)
 		}
 	}
-	fh.setAuthoritativeBase(baseSnapshot)
+	fh.setAuthoritativeBase(wfs.ackedBase(resp, baseSnapshot))
 	fh.advanceEntryVersion(ackVersionTsNs(resp), resp.GetLogSignature())
 	if applyErr := wfs.applyLocalMetadataEvent(context.Background(), event); applyErr != nil {
 		glog.Warningf("flushFileMetadata %s: best-effort metadata apply failed: %v", fileFullPath, applyErr)
diff --git a/weed/server/master_grpc_server.go b/weed/server/master_grpc_server.go
index 7d7b97818c..e046eba54c 100644
--- a/weed/server/master_grpc_server.go
+++ b/weed/server/master_grpc_server.go
@@ -667,10 +667,10 @@ func (ms *MasterServer) initialLockRingUpdate(clientType string, filerGroup stri
 	if ms.LockRingManager == nil {
 		return nil
 	}
-	// Filers are ring members; S3 gateways are lock clients that need the same
-	// view to dial a key's primary directly. Both get the initial snapshot;
-	// later membership changes already broadcast to every connected client.
-	if clientType != cluster.FilerType && clientType != cluster.S3Type {
+	// Filers are ring members; S3 gateways and mounts are lock clients that
+	// need the same view to dial a key's primary directly. All get the initial
+	// snapshot; later membership changes already broadcast to every client.
+	if clientType != cluster.FilerType && clientType != cluster.S3Type && clientType != cluster.MountType {
 		return nil
 	}
 
diff --git a/weed/server/master_grpc_server_test.go b/weed/server/master_grpc_server_test.go
index 787d1f1913..1a83f07136 100644
--- a/weed/server/master_grpc_server_test.go
+++ b/weed/server/master_grpc_server_test.go
@@ -148,6 +148,21 @@ func TestInitialLockRingUpdateSkipsNonFilers(t *testing.T) {
 	assert.Nil(t, ms.initialLockRingUpdate(cluster.BrokerType, "group-a"))
 }
 
+func TestInitialLockRingUpdateReachesLockClients(t *testing.T) {
+	ms := &MasterServer{
+		LockRingManager: cluster.NewLockRingManager(nil),
+	}
+
+	ms.LockRingManager.AddServer("group-a", "filer1:8888")
+	ms.LockRingManager.FlushPending("group-a")
+
+	for _, clientType := range []string{cluster.S3Type, cluster.MountType} {
+		resp := ms.initialLockRingUpdate(clientType, "group-a")
+		require.NotNil(t, resp, clientType)
+		assert.Equal(t, []string{"filer1:8888"}, resp.LockRingUpdate.Servers, clientType)
+	}
+}
+
 // TestReconnectedClientSurvivesOldHandlerCleanup covers a client reconnecting
 // KeepConnected under the same name before the old handler has exited: the old
 // handler's deferred cleanup must not close the channel the reconnected stream