mount: merge concurrent content changes from another mount instead of failing or dropping them (#12596)

* mount: merge a refused content flush onto the stored entry

A content flush conditioned on the handle's base failed with EIO whenever a
peer had changed the file since, so two mounts appending to one file could
not both close it cleanly. On a refused condition, reread the entry and
replay this handle's changes since its base onto it: the chunks it added or
dropped, and the attributes, extended keys and inline content it changed.
A local truncation keeps its size and clips the peer's chunks to it;
otherwise the larger size wins. The merged entry is published against the
stored one, up to three times, and installed in the handle so its next
flush keeps the peer's chunks. A replaced or deleted file is still refused.

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

* mount: merge a peer's content change into a dirty handle

A subscription event carrying a peer's content change replaced a dirty
handle's entry and discarded its unflushed pages, so the handle's writes
were lost and its close still returned OK. Fold the peer's entry into the
handle with the same merge a refused flush uses, keep the dirty pages, and
take the peer's entry as the new base. A replaced file still invalidates.

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

* mount: stop a flush merge once a filer accepts the write

withFilerClient can call the callback on several filers. A refusal from
one left its stored entry behind, so a later filer's success went on to
another merge round.

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

* mount: refuse a merge that puts inline content next to chunks

A read inside the inline bytes returns them without looking at chunks,
so one side's inline content merged with the other side's chunks hid
data after a successful flush.

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

* mount: refuse to merge a whole-file rewrite over a peer's chunk change

maybeMergeChunks re-uploads the handle's whole view as unstamped chunks.
Merged after a peer's change, they tie with a peer's own rewrite or bring
back a range the peer truncated, so that flush fails as before.

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

* mount: clip peer chunks at the lowest truncate, not the final size

A handle that truncated and then wrote past the old end ends up larger
than its base, so the merge did not see the truncate and kept the peer's
chunks inside the range the truncate cleared. SetAttr now records the
lowest size truncated to over the current base.

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

* mount: keep a peer's truncate when the handle left the size alone

The merge took the larger of the handle's and the stored size, so a peer
truncate came back as a zero-filled tail when the handle only overwrote
bytes inside it. The stored size now stands unless the handle changed
its size or wrote past it, and rebasing keeps only the growth writes made
after the flush was taken.

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

* mount: retry a read view whose merged manifests did not resolve

A failed SetChunks leaves the chunk group answering every read with the
resolution error, and nothing re-ran it while the entry stayed the same.
The handle now marks such a view (on rebase and on SetEntry) and the
next read or lseek resolves the entry's chunks again.

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

* mount: keep a dirty handle's writes when a peer change cannot merge

The new merge refusals sent a peer event on the same file down the old
path, which discarded the dirty pages and adopted the peer's entry, so
close returned OK without the writes. The handle now keeps its writes and
base, and the flush meets the conflict and fails.

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

* mount: keep an unpublished truncate across a peer's base change

The truncate mark was tied to the base pointer, so a peer event that moved
a dirty handle's base (a merged content change or a metadata-only change)
dropped it, and a later merge on a regrown file kept peer chunks inside
the truncated range. Those base moves now carry the mark; a flush ack or a
discarded handle state still clears it.

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

* mount: wait for the flush before merging a shrink over buffered writes

A write inside the file leaves its bytes in dirty pages without moving
the size. Merging a peer's truncate then adopted the smaller size, and the
chunk uploaded later sat past it, unreadable through the handle. The event
path and the periodic metadata flush now refuse a merge that shrinks the
handle; the close flush uploads the pages first and merges then.

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

* mount: tell a shortened base chunk from a whole-file rewrite

A truncate shortens a base chunk in place, so it gets a new reference
key, and an old chunk can carry no timestamp. The rewrite refusal now
applies only to unstamped chunks whose file id is new to the base.

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

* mount: check the creation time when the base has no inode

An entry without an inode or hard link id matched any replacement at the
same path, so a stale handle could merge into a different file. Without
an inode the merge now needs the same creation time. The filer keeps it
across updates; only a synthesized base has neither and still merges.

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

* mount: keep a dirty handle's writes when a peer replaces the file

A peer entry with another inode at the handle's path went down the
fallback that destroys the dirty pages and installs the replacement, so
close returned OK without the writes. The handle now keeps its writes
and base like it does for an unmergeable change, and its flush refuses to
update the replacement.

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

* mount: build the content flush tests on freebsd

The invalidation tests, which every platform compiles, now use the
content flush helpers, so the freebsd cross-compile vet failed with
undefined: flushConditionEntry. Nothing in the file is freebsd-specific.

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

* mount: keep a dirty handle's base when a same-content file replaces it

A replacement with the base's chunks but another inode counted as a
metadata-only change, so the dirty handle adopted it as its base and the
next conditional flush overwrote the new file.

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

* mount: bring a peer's metadata change into a dirty handle

A metadata-only event moved a dirty handle's base but left its live
entry, so the next flush read the peer's chmod as a local edit and
published the old mode back. Replay the handle's own changes onto the
peer's entry instead, as for a content change.

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

* mount: wait for the flush before merging peer inline content over buffered writes

A buffered write is not a chunk yet, so a merge could install a peer's
inline content that later reads return ahead of the write's chunk.

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

* mount: track a whole-file chunk rewrite on the handle

The merge told a rewrite by its unstamped chunks, which a manifest fold
hides: manifests are unstamped either way. Record the rewrite against the
base it was made from, carried across peer base changes like a pending
truncate, and refuse the merge from that.

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

* mount: refuse a merge that shortens a manifest at the truncate

Clipping a manifest reference leaves its members whole, so peer bytes past
the handle's truncate could reappear once the file grew again.

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

* mount: flush a dirty handle after a peer replaces its file

The replacement test checked only the retained handle state. It now
flushes against a filer holding the replacement, including one with the
base's content, and checks the flush fails and leaves it unchanged.

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

* mount: test an append fsync after a peer's attribute-only change

A second mount's acknowledged chmod, chown, utimes or setxattr leaves
the content alone, so the writer's conditional content flush must still
publish its append and keep that change. On master the refused flush
surfaces as EIO from Fsync.

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

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

weed/mount/content_flush_condition_test.go

@@ -1,13 +1,15 @@
-//go:build !freebsd
-
 package mount
 
 import (
 	"context"
+	"errors"
+	"slices"
 	"sync"
 	"testing"
+	"time"
 
 	"github.com/seaweedfs/go-fuse/v2/fuse"
+	"github.com/seaweedfs/seaweedfs/weed/filer"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/util"
 	"google.golang.org/grpc/codes"
@@ -51,6 +53,17 @@ func (s *contentFlushConditionFiler) CreateEntry(ctx context.Context, r *filer_p
 		},
 	}}, nil
 }
+func (s *contentFlushConditionFiler) UpdateEntry(ctx context.Context, r *filer_pb.UpdateEntryRequest) (*filer_pb.UpdateEntryResponse, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if clauses := r.GetCondition().GetClauses(); len(clauses) > 0 && !proto.Equal(clauses[0].GetExpectedEntry(), s.entry) {
+		return nil, status.Error(codes.FailedPrecondition, "entry changed")
+	}
+	s.calls++
+	s.entry = proto.Clone(r.Entry).(*filer_pb.Entry)
+	s.entry.CommitTsNs = 1000 + int64(s.calls)
+	return &filer_pb.UpdateEntryResponse{LogTsNs: 2000}, nil
+}
 func (s *contentFlushConditionFiler) LookupDirectoryEntry(ctx context.Context, r *filer_pb.LookupDirectoryEntryRequest) (*filer_pb.LookupDirectoryEntryResponse, error) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
@@ -68,9 +81,9 @@ func flushConditionEntry(ids ...uint64) *filer_pb.Entry {
 	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.
+// Exercise the actual mount content-publication caller. A peer's append is
+// merged with ours; a replaced or deleted file is not ours to merge into, so
+// the refused flush keeps our dirty entry instead of overwriting the peer.
 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) {
@@ -79,11 +92,11 @@ func TestContentFlushPreservesPeerChunks(t *testing.T) {
 			live := flushConditionEntry(1, 3)
 			live.Chunks[1].Offset = 2
 			current := flushConditionEntry(1)
-			wantError := false
+			wantError, wantCalls := false, 1
 			switch name {
 			case "peer-append":
 				current = flushConditionEntry(1, 2)
-				wantError = true
+				wantCalls = 2
 			case "replaced":
 				current = flushConditionEntry(4)
 				current.Attributes.Inode = 99
@@ -135,11 +148,18 @@ func TestContentFlushPreservesPeerChunks(t *testing.T) {
 				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)
+				want := []uint64{1, 3}
+				if name == "peer-append" {
+					want = []uint64{1, 2, 3}
+				}
+				if got := chunkFileKeys(fake.entry); !slices.Equal(got, want) {
+					t.Fatalf("stored chunks %v, want %v", got, want)
+				}
+				if got := chunkFileKeys(fh.GetEntry().GetEntry()); !slices.Equal(got, want) {
+					t.Fatalf("handle chunks %v, want %v", got, want)
 				}
 			}
-			if fake.calls != 1 {
+			if fake.calls != wantCalls {
 				t.Fatalf("unexpected replay: %d calls", fake.calls)
 			}
 		})
@@ -176,6 +196,13 @@ func TestContentFlushRepeatsOnOneHandle(t *testing.T) {
 	}
 }
 
+func chunkFileKeys(e *filer_pb.Entry) (keys []uint64) {
+	for _, c := range e.GetChunks() {
+		keys = append(keys, c.GetFid().GetFileKey())
+	}
+	return keys
+}
+
 // The periodic metadata flush publishes the same whole entry, so it must not
 // overwrite a peer's append either.
 func TestPeriodicMetadataFlushPreservesPeerChunks(t *testing.T) {
@@ -193,16 +220,182 @@ func TestPeriodicMetadataFlushPreservesPeerChunks(t *testing.T) {
 	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")
+	if err := wfs.flushFileMetadata(fh); err != nil {
+		t.Fatal(err)
 	}
 	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 got := chunkFileKeys(fake.entry); !slices.Equal(got, []uint64{1, 2, 3}) || fake.calls != 2 {
+		t.Fatalf("stored chunks %v after %d calls", got, fake.calls)
+	}
+	if got := chunkFileKeys(fh.GetEntry().GetEntry()); !slices.Equal(got, []uint64{1, 2, 3}) {
+		t.Fatalf("handle chunks %v", got)
 	}
 	if !fh.dirtyMetadata {
-		t.Fatal("refused periodic flush lost dirty metadata")
+		t.Fatal("periodic flush cleared dirty metadata")
+	}
+}
+
+// Every merged flush is followed by a flush that matches what it stored, so
+// the peer's chunk is not dropped by the next publication from this handle.
+func TestContentFlushMergeThenRepeat(t *testing.T) {
+	wfs := newInvalidateTestWFS(t)
+	fake := &contentFlushConditionFiler{entry: flushConditionEntry(1, 2)}
+	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.RememberPath(util.FullPath("/dir/file"))
+	for i := uint64(4); i <= 5; i++ {
+		fh.dirtyMetadata = true
+		if code := wfs.doFlush(context.Background(), fh, 1000, 20, false); code != fuse.OK {
+			t.Fatalf("flush before chunk %d: %v", i, code)
+		}
+		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.Fatal(code)
+	}
+	fake.mu.Lock()
+	defer fake.mu.Unlock()
+	if got := chunkFileKeys(fake.entry); !slices.Equal(got, []uint64{1, 2, 3, 4, 5}) || fake.calls != 4 {
+		t.Fatalf("stored chunks %v after %d calls", got, fake.calls)
+	}
+}
+
+// The merge keeps a peer's attribute and extended changes the handle did not
+// make, applies the ones it did, and lets a local truncation clip the peer's
+// chunks past the new end.
+func TestMergeContentEntry(t *testing.T) {
+	base := flushConditionEntry(1, 2)
+	base.Extended = map[string][]byte{"user.a": []byte("1"), "user.b": []byte("1")}
+	stored := proto.Clone(base).(*filer_pb.Entry)
+	stored.Chunks = append(stored.Chunks, &filer_pb.FileChunk{Fid: &filer_pb.FileId{VolumeId: 1, FileKey: 9, Cookie: 1}, Offset: 2, Size: 4})
+	stored.Attributes.FileMode, stored.Attributes.FileSize = 0600, 6
+	stored.Extended["user.b"] = []byte("peer")
+	stored.CommitTsNs = 77
+
+	local := proto.Clone(base).(*filer_pb.Entry)
+	local.Attributes.Mtime = 123
+	local.Extended["user.a"] = []byte("mine")
+	delete(local.Extended, "user.b")
+	local.Chunks = append(local.Chunks, &filer_pb.FileChunk{FileId: "1,0300000001", Offset: 2, Size: 1, ModifiedTsNs: 3})
+
+	merged, ok := mergeContentEntry(stored, base, local, local, -1, false)
+	if !ok {
+		t.Fatal("not merged")
+	}
+	if got := len(merged.Chunks); got != 4 || merged.Attributes.FileSize != 6 {
+		t.Fatalf("merged %d chunks, size %d", got, merged.Attributes.FileSize)
+	}
+	if merged.Attributes.FileMode != 0600 || merged.Attributes.Mtime != 123 || merged.CommitTsNs != 77 {
+		t.Fatalf("attributes %v", merged.Attributes)
+	}
+	if string(merged.Extended["user.a"]) != "mine" || merged.Extended["user.b"] != nil {
+		t.Fatalf("extended %v", merged.Extended)
+	}
+
+	truncated := proto.Clone(base).(*filer_pb.Entry)
+	truncated.Chunks = truncated.Chunks[:1]
+	truncated.Attributes.FileSize = 1
+	merged, ok = mergeContentEntry(stored, base, truncated, truncated, -1, false)
+	if !ok {
+		t.Fatal("truncation not merged")
+	}
+	if got := chunkFileKeys(merged); !slices.Equal(got, []uint64{1}) || merged.Attributes.FileSize != 1 {
+		t.Fatalf("truncated merge kept %v, size %d", got, merged.Attributes.FileSize)
+	}
+
+	// Truncated to 1, then written at 7: the peer chunk at 2 stays cut.
+	regrown := proto.Clone(truncated).(*filer_pb.Entry)
+	regrown.Chunks = append(regrown.Chunks, &filer_pb.FileChunk{Fid: &filer_pb.FileId{VolumeId: 1, FileKey: 7, Cookie: 1}, Offset: 7, Size: 1, ModifiedTsNs: 7})
+	regrown.Attributes.FileSize = 8
+	merged, ok = mergeContentEntry(stored, base, regrown, regrown, 1, false)
+	if !ok {
+		t.Fatal("truncate then write not merged")
+	}
+	if got := chunkFileKeys(merged); !slices.Equal(got, []uint64{1, 7}) || merged.Attributes.FileSize != 8 {
+		t.Fatalf("truncate then write kept %v, size %d", got, merged.Attributes.FileSize)
+	}
+
+	// Shortening a manifest reference leaves its members whole.
+	peerManifest := proto.Clone(base).(*filer_pb.Entry)
+	peerManifest.Chunks = append(peerManifest.Chunks, &filer_pb.FileChunk{Fid: &filer_pb.FileId{VolumeId: 1, FileKey: 5, Cookie: 1}, Offset: 0, Size: 10, IsChunkManifest: true})
+	if _, ok := mergeContentEntry(peerManifest, base, regrown, regrown, 1, false); ok {
+		t.Fatal("merged a peer manifest across the truncate")
+	}
+
+	// A peer truncate stands when the handle only overwrote inside it.
+	peerTruncated := proto.Clone(base).(*filer_pb.Entry)
+	peerTruncated.Chunks, peerTruncated.Attributes.FileSize = peerTruncated.Chunks[:1], 1
+	overwrote := proto.Clone(base).(*filer_pb.Entry)
+	overwrote.Chunks = append(overwrote.Chunks, &filer_pb.FileChunk{Fid: &filer_pb.FileId{VolumeId: 1, FileKey: 6, Cookie: 1}, Size: 1, ModifiedTsNs: 6})
+	merged, ok = mergeContentEntry(peerTruncated, base, overwrote, overwrote, -1, false)
+	if !ok {
+		t.Fatal("overwrite over a peer truncate not merged")
+	}
+	if got := chunkFileKeys(merged); !slices.Equal(got, []uint64{1, 6}) || merged.Attributes.FileSize != 1 {
+		t.Fatalf("overwrite over a peer truncate kept %v, size %d", got, merged.Attributes.FileSize)
+	}
+
+	// A rewrite folded into a manifest is still a rewrite.
+	rewritten := proto.Clone(base).(*filer_pb.Entry)
+	rewritten.Chunks = []*filer_pb.FileChunk{{Fid: &filer_pb.FileId{VolumeId: 1, FileKey: 8, Cookie: 1}, Size: 2, IsChunkManifest: true}}
+	if _, ok := mergeContentEntry(stored, base, rewritten, rewritten, -1, true); ok {
+		t.Fatal("merged a stale whole-file rewrite over a peer append")
+	}
+	if _, ok := mergeContentEntry(stored, base, rewritten, rewritten, -1, false); !ok {
+		t.Fatal("refused new chunks the handle did not rewrite")
+	}
+	if _, ok := mergeContentEntry(proto.Clone(base).(*filer_pb.Entry), base, rewritten, rewritten, -1, true); !ok {
+		t.Fatal("refused a rewrite with no peer chunk change")
+	}
+
+	stored.Attributes.Inode = 99
+	if _, ok := mergeContentEntry(stored, base, local, local, -1, false); ok {
+		t.Fatal("merged into a replaced file")
+	}
+	unstamped := flushConditionEntry(4)
+	unstamped.Chunks[0].Size, unstamped.Chunks[0].ModifiedTsNs = 3, 0
+	shortened := proto.Clone(unstamped).(*filer_pb.Entry)
+	shortened.Chunks[0].Size, shortened.Attributes.FileSize = 2, 2
+	peerAppend := proto.Clone(unstamped).(*filer_pb.Entry)
+	peerAppend.Chunks = append(peerAppend.Chunks, &filer_pb.FileChunk{Fid: &filer_pb.FileId{VolumeId: 1, FileKey: 9, Cookie: 1}, Offset: 3, Size: 1, ModifiedTsNs: 9})
+	if _, ok := mergeContentEntry(peerAppend, unstamped, shortened, shortened, 2, false); !ok {
+		t.Fatal("refused a truncate that shortened an unstamped base chunk")
+	}
+
+	noInode := proto.Clone(base).(*filer_pb.Entry)
+	noInode.Attributes.Inode, noInode.Attributes.Crtime = 0, 5
+	sameFile := proto.Clone(noInode).(*filer_pb.Entry)
+	sameFile.Chunks = append(sameFile.Chunks, stored.Chunks[2])
+	if _, ok := mergeContentEntry(sameFile, noInode, local, local, -1, false); !ok {
+		t.Fatal("refused an inode-less file with the same creation time")
+	}
+	sameFile.Attributes.Crtime = 6
+	if _, ok := mergeContentEntry(sameFile, noInode, local, local, -1, false); ok {
+		t.Fatal("merged into an inode-less replacement")
+	}
+
+	empty := flushConditionEntry()
+	inline := proto.Clone(empty).(*filer_pb.Entry)
+	inline.Content = []byte("peer")
+	appended := proto.Clone(empty).(*filer_pb.Entry)
+	appended.Chunks = flushConditionEntry(5).Chunks
+	if _, ok := mergeContentEntry(inline, empty, appended, appended, -1, false); ok {
+		t.Fatal("merged chunks next to inline content")
+	}
+	if _, ok := mergeContentEntry(appended, empty, inline, inline, -1, false); ok {
+		t.Fatal("merged inline content next to chunks")
 	}
 }
 
@@ -246,3 +439,59 @@ func TestContentFlushRepublishesUnchangedEntry(t *testing.T) {
 		})
 	}
 }
+
+// Rebasing keeps a size the handle reached after the flush was taken, not the
+// size it sent: a peer's truncate merged in must stand.
+func TestRebaseEntryKeepsMergedSize(t *testing.T) {
+	wfs := newInvalidateTestWFS(t)
+	inode := wfs.inodeToPath.Lookup(util.FullPath("/dir/file"), time.Now().Unix(), false, false, 0, false)
+	fh, _, _, _ := wfs.fhMap.AcquireFileHandle(wfs, inode, flushConditionEntry(1, 2), 0, 0)
+	sent := proto.Clone(fh.GetEntry().GetEntry()).(*filer_pb.Entry)
+	merged := flushConditionEntry(1)
+	merged.Attributes.FileSize = 1
+
+	fh.rebaseEntry(merged, sent)
+	if size := fh.GetEntry().GetEntry().Attributes.FileSize; size != 1 {
+		t.Fatalf("size %d after rebasing onto a truncate, want 1", size)
+	}
+
+	fh.GetEntry().GetEntry().Attributes.FileSize = 5
+	fh.rebaseEntry(merged, sent)
+	if size := fh.GetEntry().GetEntry().Attributes.FileSize; size != 5 {
+		t.Fatalf("size %d after a write grew the handle, want 5", size)
+	}
+}
+
+// A merged entry whose manifest cannot be resolved yet leaves the read view
+// marked, and the next read resolves it again instead of failing for good.
+func TestRebaseEntryRetriesUnresolvedChunkView(t *testing.T) {
+	wfs := newInvalidateTestWFS(t)
+	inode := wfs.inodeToPath.Lookup(util.FullPath("/dir/file"), time.Now().Unix(), false, false, 0, false)
+	fh, _, _, _ := wfs.fhMap.AcquireFileHandle(wfs, inode, flushConditionEntry(1), 0, 0)
+	lookupDown := true
+	group, err := filer.NewChunkGroup(func(ctx context.Context, fileId string) ([]string, error) {
+		if lookupDown {
+			return nil, errors.New("volume lookup down")
+		}
+		return nil, nil
+	}, nil, nil, 1, nil, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	fh.entryChunkGroup = group
+
+	sent := proto.Clone(fh.GetEntry().GetEntry()).(*filer_pb.Entry)
+	merged := flushConditionEntry(1, 2)
+	merged.Chunks[1].IsChunkManifest = true
+	fh.rebaseEntry(merged, sent)
+	if !fh.chunkViewStale.Load() {
+		t.Fatal("unresolved manifest not marked")
+	}
+
+	lookupDown = false
+	fh.UpdateEntry(func(e *filer_pb.Entry) { e.Chunks = e.Chunks[:1] })
+	fh.refreshChunkView()
+	if fh.chunkViewStale.Load() {
+		t.Fatal("read view still marked after the chunks resolved")
+	}
+}

weed/mount/filehandle.go

@@ -1,6 +1,7 @@
 package mount
 
 import (
+	"bytes"
 	"os"
 	"sync"
 	"sync/atomic"
@@ -66,6 +67,15 @@ type FileHandle struct {
 	// must be judged here, not against the live entry. Always store a clone.
 	baseEntry atomic.Pointer[filer_pb.Entry]
 
+	// sinceBase marks a truncate or chunk rewrite over baseEntry that a
+	// merge must honor: later writes can grow the file past a truncate, and
+	// rewritten chunks carry the handle's view of the base.
+	sinceBase atomic.Pointer[baseMark]
+
+	// chunkViewStale marks a read view whose manifest resolution failed; the
+	// next read or lseek resolves the entry's chunks again.
+	chunkViewStale atomic.Bool
+
 	// dlmLock holds the distributed lock for cross-mount write coordination.
 	// Non-nil only when -dlm is enabled and the file was opened for writing.
 	// Acquired in AcquireHandle, released in ReleaseHandle.
@@ -141,6 +151,7 @@ func (fh *FileHandle) SetEntry(entry *filer_pb.Entry) {
 		if resolveManifestErr != nil {
 			glog.Warningf("failed to resolve manifest chunks in %+v", filer_pb.ScrubCipherKeys(entry))
 		}
+		fh.chunkViewStale.Store(resolveManifestErr != nil)
 	} else {
 		glog.Fatalf("setting file handle entry to nil")
 	}
@@ -174,6 +185,125 @@ func (fh *FileHandle) installAckedEntry(entry *filer_pb.Entry, versionTsNs int64
 	fh.advanceEntryVersion(versionTsNs, signature)
 }
 
+// rebaseEntry installs an entry merged with a peer's changes, keeping the
+// chunks and growth writers added after sent was taken, and refreshes the
+// read view.
+func (fh *FileHandle) rebaseEntry(merged, sent *filer_pb.Entry) {
+	inSent := make(map[string]bool, len(sent.Chunks))
+	for _, c := range sent.Chunks {
+		inSent[contentChunkKey(c)] = true
+	}
+	merged = proto.Clone(merged).(*filer_pb.Entry)
+	fh.UpdateEntry(func(e *filer_pb.Entry) {
+		chunks := merged.Chunks
+		for _, c := range e.Chunks {
+			if !inSent[contentChunkKey(c)] {
+				chunks = append(chunks, c)
+			}
+		}
+		attrs := merged.Attributes
+		if size := e.GetAttributes().GetFileSize(); size > sent.GetAttributes().GetFileSize() && size > attrs.FileSize {
+			attrs.FileSize = size
+		}
+		e.Attributes, e.Extended, e.Content, e.Chunks = attrs, merged.Extended, merged.Content, chunks
+		err := fh.entryChunkGroup.SetChunks(chunks)
+		if err != nil {
+			glog.Warningf("rebase %s: resolve chunks: %v", fh.FullPath(), err)
+		}
+		fh.chunkViewStale.Store(err != nil)
+	})
+}
+
+// refreshChunkView retries a read view whose manifest resolution failed.
+func (fh *FileHandle) refreshChunkView() {
+	if !fh.chunkViewStale.Load() {
+		return
+	}
+	fh.UpdateEntry(func(e *filer_pb.Entry) {
+		fh.chunkViewStale.Store(fh.entryChunkGroup.SetChunks(e.Chunks) != nil)
+	})
+}
+
+// mergePeerContent folds a peer's change into a dirty handle, so its own
+// writes stay on top of the peer's instead of being discarded.
+func (fh *FileHandle) mergePeerContent(peer, base *filer_pb.Entry) bool {
+	var local *filer_pb.Entry
+	fh.UpdateEntry(func(e *filer_pb.Entry) {
+		local = proto.Clone(e).(*filer_pb.Entry)
+	})
+	mark := fh.markSince(base)
+	merged, ok := mergeContentEntry(peer, base, local, local, mark.truncatedTo, mark.rewrote)
+	ok = ok && mergeKeepsBuffered(merged, local)
+	if ok {
+		fh.rebaseEntry(merged, local)
+	}
+	return ok
+}
+
+// mergeKeepsBuffered reports whether merged leaves room for writes still
+// buffered over local. They are not chunks yet, so a merge that shrinks the
+// file or brings in inline content (read ahead of chunks) waits for the flush.
+func mergeKeepsBuffered(merged, local *filer_pb.Entry) bool {
+	return merged.Attributes.FileSize >= filer.FileSize(local) && bytes.Equal(merged.Content, local.Content)
+}
+
+// baseMark records what the handle did over base that its entry does not
+// show: the lowest size a truncate set (-1 for none), and a whole-file
+// rewrite of its chunks.
+type baseMark struct {
+	base        *filer_pb.Entry
+	truncatedTo int64
+	rewrote     bool
+}
+
+func (fh *FileHandle) markBase(update func(m *baseMark)) {
+	for {
+		base, old := fh.baseEntry.Load(), fh.sinceBase.Load()
+		next := baseMark{base: base, truncatedTo: -1}
+		if old != nil && old.base == base {
+			next = *old
+		}
+		update(&next)
+		if fh.sinceBase.CompareAndSwap(old, &next) {
+			return
+		}
+	}
+}
+
+// noteTruncate records a truncate to size over the current base.
+func (fh *FileHandle) noteTruncate(size uint64) {
+	fh.markBase(func(m *baseMark) {
+		if m.truncatedTo < 0 || int64(size) < m.truncatedTo {
+			m.truncatedTo = int64(size)
+		}
+	})
+}
+
+// noteRewrite records that the handle's chunks were rewritten from its view
+// of the current base.
+func (fh *FileHandle) noteRewrite() {
+	fh.markBase(func(m *baseMark) { m.rewrote = true })
+}
+
+// adoptPeerBase moves a dirty handle's base to a peer's entry. What the
+// handle has not published yet still applies over the new base.
+func (fh *FileHandle) adoptPeerBase(base *filer_pb.Entry) {
+	old := fh.baseEntry.Swap(base)
+	if m := fh.sinceBase.Load(); m != nil && m.base == old {
+		moved := *m
+		moved.base = base
+		fh.sinceBase.CompareAndSwap(m, &moved)
+	}
+}
+
+// markSince is what the handle did over base.
+func (fh *FileHandle) markSince(base *filer_pb.Entry) baseMark {
+	if m := fh.sinceBase.Load(); m != nil && m.base == base {
+		return *m
+	}
+	return baseMark{base: base, truncatedTo: -1}
+}
+
 // setAuthoritativeBase installs the base snapshot a local ack acknowledged.
 func (fh *FileHandle) setAuthoritativeBase(base *filer_pb.Entry) {
 	fh.baseEntry.Store(base)

weed/mount/filehandle_read.go

@@ -100,6 +100,7 @@ func (fh *FileHandle) readFromChunksWithContext(ctx context.Context, buff []byte
 	}
 
 	// Fall back to normal chunk reading
+	fh.refreshChunkView()
 	totalRead, ts, err := fh.entryChunkGroup.ReadDataAt(ctx, fileSize, buff, offset)
 
 	if err != nil && err != io.EOF {

weed/mount/weedfs.go

@@ -1044,18 +1044,24 @@ func (wfs *WFS) invalidateOpenFileHandle(invalidation meta_cache.EntryInvalidati
 		fh.advanceEntryVersion(candidateTsNs, 0)
 		return
 	}
+	// A dirty handle replays its own changes onto the peer's entry, keeping
+	// its pages. An unmergeable change or a replaced file keeps the writes;
+	// the flush reports the conflict.
+	if fh.dirtyMetadata && base != nil {
+		if sameContentFile(candidate, base) && fh.mergePeerContent(candidate, base) {
+			fh.adoptPeerBase(proto.Clone(candidate).(*filer_pb.Entry))
+			fh.advanceEntryVersion(candidateTsNs, 0)
+		}
+		return
+	}
 	// Dirty pages overlay content, so only a content change invalidates them:
 	// a metadata-only event (chmod, touch, or a committed copy's own event
-	// against an approximate base) leaves them valid. A dirty handle likewise
-	// keeps its diverged entry unless foreign content supersedes it.
-	contentChanged := base == nil || !sameEntryContent(candidate, base)
-	if contentChanged {
+	// against an approximate base) leaves them valid.
+	if base == nil || !sameEntryContent(candidate, base) {
 		fh.dirtyPages.Destroy()
 		fh.dirtyPages = newPageWriter(fh, wfs.option.ChunkSizeLimit)
 	}
-	if contentChanged || !fh.dirtyMetadata {
-		fh.SetEntry(candidate)
-	}
+	fh.SetEntry(candidate)
 	fh.baseEntry.Store(proto.Clone(candidate).(*filer_pb.Entry))
 	fh.advanceEntryVersion(candidateTsNs, 0)
 	return

weed/mount/weedfs_attr.go

@@ -134,6 +134,9 @@ func (wfs *WFS) SetAttr(cancel <-chan struct{}, input *fuse.SetAttrIn, out *fuse
 		wfs.invalidateOpenMtimeCache(input.NodeId)
 		oldFileSize := filer.FileSize(entry)
 		if size < oldFileSize {
+			if fh != nil {
+				fh.noteTruncate(size)
+			}
 			// fmt.Printf("truncate %v \n", fullPath)
 			var chunks []*filer_pb.FileChunk
 			var truncatedChunks []*filer_pb.FileChunk

weed/mount/weedfs_attr_cas_test.go

@@ -138,6 +138,82 @@ func attributeAppendEntry(records int) *filer_pb.Entry {
 	return e
 }
 
+// A peer's acknowledged attribute change leaves the content alone, so our
+// append's conditional flush must still publish, keeping the peer's change.
+func TestAppendFlushAfterPeerAttributeChange(t *testing.T) {
+	for _, op := range []string{"none", "chmod", "chown", "utimes", "setxattr"} {
+		t.Run(op, func(t *testing.T) {
+			writer, peer := newInvalidateTestWFS(t), newInvalidateTestWFS(t)
+			peer.atimeMap = make(map[uint64]time.Time)
+			base := attributeAppendEntry(1)
+			base.CommitTsNs = 100
+			fake := &contentFlushConditionFiler{entry: proto.Clone(base).(*filer_pb.Entry)}
+			startFakeFiler(t, writer, fake)
+			startFakeFiler(t, peer, fake)
+			path := util.FullPath("/dir/file")
+			local := proto.Clone(base).(*filer_pb.Entry)
+			writer.mapPbIdFromFilerToLocal(local)
+			fh := newFileHandle(writer, 1, 42, local)
+			fh.RememberPath(path)
+			t.Cleanup(fh.dirtyPages.Destroy)
+			writer.fhMap.inode2fh[42], writer.fhMap.fh2inode[1] = fh, 42
+			appended := attributeAppendEntry(2).Chunks[1]
+			fh.UpdateEntry(func(e *filer_pb.Entry) {
+				e.Chunks = append(e.Chunks, appended)
+				e.Attributes.FileSize = 512
+			})
+			fh.dirtyMetadata = true
+
+			peer.inodeToPath.Lookup("/dir", 1, true, false, 0, false)
+			peer.inodeToPath.Lookup(path, 1, false, false, 42, true)
+			peer.inodeToPath.MarkChildrenCached("/dir")
+			if err := peer.metaCache.InsertEntry(context.Background(), filer.FromPbEntry("/dir", base), 1); err != nil {
+				t.Fatal(err)
+			}
+			code := fuse.OK
+			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 = fuse.FATTR_MTIME, 123
+			case "setxattr":
+				code = peer.SetXAttr(nil, &fuse.SetXAttrIn{InHeader: fuse.InHeader{NodeId: 42}}, "user.test", []byte("peer"))
+			}
+			if in.Valid != 0 {
+				code = peer.SetAttr(nil, in, &fuse.AttrOut{})
+			}
+			if code != fuse.OK {
+				t.Fatalf("peer %s: %v", op, code)
+			}
+
+			if code = writer.Fsync(nil, &fuse.FsyncIn{InHeader: fuse.InHeader{NodeId: 42}, Fh: 1}); code != fuse.OK {
+				t.Fatalf("append fsync after peer %s: %v", op, code)
+			}
+			fake.mu.Lock()
+			defer fake.mu.Unlock()
+			got := fake.entry
+			if len(got.Chunks) != 2 || got.Attributes.FileSize != 512 {
+				t.Fatalf("append not published after peer %s: %v", op, got)
+			}
+			a := got.Attributes
+			kept := map[string]bool{
+				"none":     true,
+				"chmod":    a.FileMode&0777 == 0600,
+				"chown":    a.Uid == 1001 && a.Gid == 21,
+				"utimes":   a.Mtime == 123,
+				"setxattr": string(got.Extended["xattr-user.test"]) == "peer",
+			}[op]
+			if !kept {
+				t.Fatalf("append flush dropped the peer's %s: %v", op, got)
+			}
+		})
+	}
+}
+
 func TestAttributeOnlyOpsPreserveAcknowledgedAppends(t *testing.T) {
 	for _, op := range []string{"chmod", "chown", "utimes", "setxattr", "removexattr"} {
 		for _, stale := range []bool{false, true} {

weed/mount/weedfs_file_lseek.go

@@ -70,6 +70,7 @@ func (wfs *WFS) Lseek(cancel <-chan struct{}, in *fuse.LseekIn, out *fuse.LseekO
 	}()
 
 	// search chunks for the offset
+	fh.refreshChunkView()
 	found, offset, err := fh.entryChunkGroup.SearchChunks(ctx, offset, fileSize, in.Whence)
 	if err != nil {
 		glog.Errorf("Lseek %s: %v", fh.FullPath(), err)

weed/mount/weedfs_file_sync.go

@@ -254,6 +254,7 @@ func (wfs *WFS) flushMetadataToFiler(ctx context.Context, fh *FileHandle, dir, n
 	} else if mergedChunks != nil {
 		chunks = mergedChunks
 		manifestChunks = nil
+		fh.noteRewrite()
 	}
 
 	// no chunk deleter here: a failed fold reports the blobs it saved
@@ -281,11 +282,15 @@ 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.createContentEntry(ctx, fh, request)
+	resp, merged, err := wfs.createContentEntry(ctx, fh, request, baseSnapshot, false)
 	if err != nil {
 		glog.Errorf("fh flush create %s: %v", fileFullPath, err)
 		return fmt.Errorf("fh flush create %s: %w", fileFullPath, err)
 	}
+	if merged != nil {
+		fh.rebaseEntry(merged, requestEntry)
+		baseSnapshot = merged
+	}
 
 	event := resp.GetMetadataEvent()
 	if event == nil {
@@ -312,16 +317,16 @@ func (wfs *WFS) flushMetadataToFiler(ctx context.Context, fh *FileHandle, dir, n
 }
 
 // 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.
+// state the handle observed. 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) {
+// When the stored entry moved on, the handle's own changes since its base are
+// replayed onto it and published against it, so a peer's acknowledged chunks
+// survive. merged is that entry in local ids; nil when the request went as is.
+// With writes still buffered, the merge must keep room for them.
+func (wfs *WFS) createContentEntry(ctx context.Context, fh *FileHandle, request *filer_pb.CreateEntryRequest, local *filer_pb.Entry, buffered bool) (resp *filer_pb.CreateEntryResponse, merged *filer_pb.Entry, err error) {
 	base := fh.baseEntry.Load()
 	guard := &filer_pb.WriteCondition_Clause{Kind: filer_pb.WriteCondition_IF_NOT_EXISTS}
 	if base != nil {
@@ -329,13 +334,15 @@ func (wfs *WFS) createContentEntry(ctx context.Context, fh *FileHandle, request
 		wfs.mapPbIdFromLocalToFiler(expected)
 		guard = &filer_pb.WriteCondition_Clause{Kind: filer_pb.WriteCondition_IF_ENTRY_EQUAL, ExpectedEntry: expected}
 	}
-	for attempt := 0; attempt < 2; attempt++ {
+	sent := request.Entry
+	for attempt := 0; ; 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
+			stored = nil
 			resp, callErr = filer_pb.CreateEntryWithResponse(ctx, client, request)
-			if base == nil || attempt > 0 || !errors.Is(callErr, filer_pb.ErrPreconditionFailed) {
+			if base == nil || attempt >= maxContentFlushMerges || !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 {
@@ -343,29 +350,175 @@ func (wfs *WFS) createContentEntry(ctx context.Context, fh *FileHandle, request
 			}
 			return callErr
 		})
-		if stored == nil || !wfs.sameObservedEntry(stored, base) {
-			return resp, err
+		if err == nil && request.Entry != sent {
+			merged = proto.Clone(request.Entry).(*filer_pb.Entry)
+			wfs.mapPbIdFromFilerToLocal(merged)
 		}
+		if err == nil || stored == nil {
+			return resp, merged, err
+		}
+		mark := fh.markSince(base)
+		next, ok := mergeContentEntry(stored, base, local, sent, mark.truncatedTo, mark.rewrote)
+		if !ok || buffered && !mergeKeepsBuffered(next, local) {
+			return resp, nil, err
+		}
+		glog.V(1).Infof("content flush %s/%s: entry changed since the handle's base, merging", request.Directory, request.Entry.Name)
+		request.Entry = next
 		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)
+// maxContentFlushMerges bounds how often one flush rebases onto a moving entry.
+const maxContentFlushMerges = 3
+
+// mergeContentEntry replays what local changed since base onto stored: the
+// chunks it added or dropped, and the attributes, extended keys and inline
+// content it changed, taking those values from sent (local in stored's id
+// space). Everything else keeps the stored value. A local truncation keeps
+// the handle's size and clips stored chunks at the lowest size truncated to
+// (truncatedTo, or -1); otherwise the stored size stands unless the handle
+// changed its size or wrote past it.
+// A different file at the path, inline content changed on both sides,
+// inline content next to chunks (reads stop at the inline bytes), a manifest
+// across the truncate (its members keep their sizes), or a local
+// whole-file rewrite (rewrote) over a peer's chunk change cannot be merged:
+// the rewrite carries the handle's stale view.
+func mergeContentEntry(stored, base, local, sent *filer_pb.Entry, truncatedTo int64, rewrote bool) (*filer_pb.Entry, bool) {
+	if !sameContentFile(stored, base) {
+		return nil, false
 	}
-	return proto.Equal(current, observed)
+	merged := proto.Clone(stored).(*filer_pb.Entry)
+	merged.Name = sent.Name
+	if !bytes.Equal(local.Content, base.Content) {
+		if !bytes.Equal(stored.Content, base.Content) {
+			return nil, false
+		}
+		merged.Content = sent.Content
+	}
+
+	inBase := make(map[string]bool, len(base.Chunks))
+	for _, c := range base.Chunks {
+		inBase[contentChunkKey(c)] = true
+	}
+	inLocal := make(map[string]bool, len(local.Chunks))
+	for _, c := range local.Chunks {
+		inLocal[contentChunkKey(c)] = true
+	}
+	localSize := filer.FileSize(local)
+	if truncatedTo < 0 && localSize < filer.FileSize(base) {
+		truncatedTo = int64(localSize)
+	}
+	peerChanged := len(stored.Chunks) != len(base.Chunks)
+	merged.Chunks = merged.Chunks[:0]
+	for _, c := range stored.Chunks {
+		key := contentChunkKey(c)
+		peerChanged = peerChanged || !inBase[key]
+		if inBase[key] && !inLocal[key] || truncatedTo >= 0 && c.Offset >= truncatedTo {
+			continue
+		}
+		c = proto.Clone(c).(*filer_pb.FileChunk)
+		if truncatedTo >= 0 && c.Offset+int64(c.Size) > truncatedTo {
+			if c.IsChunkManifest {
+				return nil, false
+			}
+			c.Size = uint64(truncatedTo - c.Offset)
+		}
+		merged.Chunks = append(merged.Chunks, c)
+	}
+	if rewrote && peerChanged {
+		return nil, false
+	}
+	size := int64(filer.FileSize(stored))
+	if localSize != filer.FileSize(base) {
+		size = max(size, int64(localSize))
+	}
+	for _, c := range local.Chunks {
+		if !inBase[contentChunkKey(c)] {
+			merged.Chunks = append(merged.Chunks, proto.Clone(c).(*filer_pb.FileChunk))
+			size = max(size, c.Offset+int64(c.Size))
+		}
+	}
+	if len(merged.Content) > 0 && len(merged.Chunks) > 0 {
+		return nil, false
+	}
+
+	if merged.Attributes == nil {
+		merged.Attributes = &filer_pb.FuseAttributes{}
+	}
+	baseAttrs, localAttrs := base.GetAttributes(), local.GetAttributes()
+	if baseAttrs == nil {
+		baseAttrs = &filer_pb.FuseAttributes{}
+	}
+	if localAttrs == nil {
+		localAttrs = &filer_pb.FuseAttributes{}
+	}
+	m, b, l := merged.Attributes.ProtoReflect(), baseAttrs.ProtoReflect(), localAttrs.ProtoReflect()
+	s := proto.Clone(sent.GetAttributes()).(*filer_pb.FuseAttributes).ProtoReflect()
+	fields := m.Descriptor().Fields()
+	for i := 0; i < fields.Len(); i++ {
+		fd := fields.Get(i)
+		if fd.Name() == "file_size" || l.Get(fd).Equal(b.Get(fd)) {
+			continue
+		}
+		if s.Has(fd) {
+			m.Set(fd, s.Get(fd))
+		} else {
+			m.Clear(fd)
+		}
+	}
+
+	for k, v := range local.Extended {
+		if old, found := base.Extended[k]; !found || !bytes.Equal(old, v) {
+			if merged.Extended == nil {
+				merged.Extended = make(map[string][]byte)
+			}
+			merged.Extended[k] = sent.Extended[k]
+		}
+	}
+	for k := range base.Extended {
+		if _, found := local.Extended[k]; !found {
+			delete(merged.Extended, k)
+		}
+	}
+
+	if truncatedTo >= 0 {
+		size = int64(localSize)
+	}
+	merged.Attributes.FileSize = uint64(size)
+	return merged, true
+}
+
+// sameContentFile reports whether stored is still the regular file base was:
+// the same inode, or for an entry without one, the same creation time. A
+// synthesized base with neither cannot tell.
+func sameContentFile(stored, base *filer_pb.Entry) bool {
+	if stored == nil || base == nil || stored.IsDirectory || !bytes.Equal(stored.HardLinkId, base.HardLinkId) {
+		return false
+	}
+	if inode := base.GetAttributes().GetInode(); inode != 0 {
+		return stored.GetAttributes().GetInode() == inode
+	}
+	crtime := base.GetAttributes().GetCrtime()
+	return crtime == 0 || stored.GetAttributes().GetCrtime() == crtime
+}
+
+// contentChunkKey identifies a chunk reference: truncation shrinks a chunk in
+// place, so the same file id at another size is a different reference. Both
+// file id encodings give one key, and the chunk is left as is.
+func contentChunkKey(c *filer_pb.FileChunk) string {
+	return fmt.Sprintf("%s@%d+%d", contentFileIdKey(c), c.Offset, c.Size)
+}
+
+func contentFileIdKey(c *filer_pb.FileChunk) string {
+	fid := c.Fid
+	if fid == nil {
+		if parsed, err := filer_pb.ToFileIdObject(c.FileId); err == nil {
+			fid = parsed
+		} else {
+			return c.FileId
+		}
+	}
+	return fmt.Sprintf("%d,%x,%x", fid.VolumeId, fid.FileKey, fid.Cookie)
 }
 
 // ackedBase is the base an acknowledged create leaves: the entry the filer

weed/mount/weedfs_invalidate_open_handle_test.go

@@ -20,6 +20,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/util"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/protobuf/proto"
 )
 
 func newInvalidateTestWFS(t *testing.T) *WFS {
@@ -1241,7 +1242,8 @@ func TestCommittedCopyWithFailedReadbackAdoptsItsEvent(t *testing.T) {
 		t.Fatalf("handle version = %d, want 1500", got)
 	}
 
-	// The adoption is one-shot: a genuinely foreign event still invalidates.
+	// The adoption is one-shot: a genuinely foreign event still installs,
+	// merged under the handle's own writes.
 	if err := wfs.metaCache.ApplyMetadataResponse(context.Background(), updateEventFor("file", 300, 1600), meta_cache.SubscriberMetadataResponseApplyOptions); err != nil {
 		t.Fatalf("apply foreign event: %v", err)
 	}
@@ -1249,8 +1251,11 @@ func TestCommittedCopyWithFailedReadbackAdoptsItsEvent(t *testing.T) {
 	if size := fh.GetEntry().GetEntry().Attributes.FileSize; size != 300 {
 		t.Fatalf("live entry file size = %d, want 300 (foreign event after adoption must install)", size)
 	}
-	if fh.dirtyPages == pagesBefore {
-		t.Fatal("foreign event after adoption must invalidate dirty pages")
+	if base := fh.baseEntry.Load(); base.GetAttributes().GetFileSize() != 300 {
+		t.Fatalf("base %v, want the foreign entry", base)
+	}
+	if fh.dirtyPages != pagesBefore {
+		t.Fatal("dirty pages destroyed by a mergeable foreign event")
 	}
 }
 
@@ -2075,3 +2080,308 @@ func TestAckFromAnotherFilerIsNotDroppedBehindForeignFence(t *testing.T) {
 		t.Fatal("an older ack from the fencing filer was installed; within one clock domain it must be refused")
 	}
 }
+
+// A peer's append reaching a dirty handle is merged under the handle's own
+// writes: the uploaded chunk and the unflushed pages both survive.
+func TestPeerAppendMergesIntoDirtyHandle(t *testing.T) {
+	wfs := newInvalidateTestWFS(t)
+
+	inode := wfs.inodeToPath.Lookup(util.FullPath("/dir/file"), time.Now().Unix(), false, false, 0, false)
+	fh, _, _, _ := wfs.fhMap.AcquireFileHandle(wfs, inode, flushConditionEntry(1), 0, 0)
+	fh.AddChunks([]*filer_pb.FileChunk{{FileId: "1,0300000001", Offset: 2, Size: 1, ModifiedTsNs: 3}})
+	fh.dirtyMetadata = true
+	pagesBefore := fh.dirtyPages
+
+	peer := flushConditionEntry(1, 2)
+	peer.Attributes.FileSize = 4
+	peer.Chunks[1].Offset, peer.Chunks[1].Size = 3, 1
+	appendResp := &filer_pb.SubscribeMetadataResponse{
+		Directory: "/dir",
+		TsNs:      1500,
+		EventNotification: &filer_pb.EventNotification{
+			OldEntry:      flushConditionEntry(1),
+			NewEntry:      peer,
+			NewParentPath: "/dir",
+		},
+	}
+	if err := wfs.metaCache.ApplyMetadataResponse(context.Background(), appendResp, meta_cache.SubscriberMetadataResponseApplyOptions); err != nil {
+		t.Fatalf("apply append: %v", err)
+	}
+	wfs.metaCache.WaitForEntryInvalidations()
+
+	if fh.dirtyPages != pagesBefore {
+		t.Fatal("dirty pages destroyed by a peer append")
+	}
+	entry := fh.GetEntry().GetEntry()
+	if got := len(entry.Chunks); got != 3 || entry.Attributes.FileSize != 4 {
+		t.Fatalf("handle has %d chunks, size %d; want the peer's and its own", got, entry.Attributes.FileSize)
+	}
+	if base := fh.baseEntry.Load(); len(base.GetChunks()) != 2 {
+		t.Fatalf("base %v, want the peer's entry", base)
+	}
+}
+
+// A peer change that cannot be merged (inline content over the handle's new
+// chunk) leaves the dirty handle alone, so its flush fails instead of the
+// event silently dropping the writes.
+func TestUnmergeablePeerChangeKeepsDirtyHandle(t *testing.T) {
+	wfs := newInvalidateTestWFS(t)
+
+	inode := wfs.inodeToPath.Lookup(util.FullPath("/dir/file"), time.Now().Unix(), false, false, 0, false)
+	fh, _, _, _ := wfs.fhMap.AcquireFileHandle(wfs, inode, flushConditionEntry(1), 0, 0)
+	fh.AddChunks([]*filer_pb.FileChunk{{FileId: "1,0300000001", Offset: 2, Size: 1, ModifiedTsNs: 3}})
+	fh.dirtyMetadata = true
+	pagesBefore := fh.dirtyPages
+
+	peer := flushConditionEntry()
+	peer.Content = []byte("peer")
+	peer.Attributes.FileSize = 4
+	resp := &filer_pb.SubscribeMetadataResponse{
+		Directory: "/dir",
+		TsNs:      1500,
+		EventNotification: &filer_pb.EventNotification{
+			OldEntry:      flushConditionEntry(1),
+			NewEntry:      peer,
+			NewParentPath: "/dir",
+		},
+	}
+	if err := wfs.metaCache.ApplyMetadataResponse(context.Background(), resp, meta_cache.SubscriberMetadataResponseApplyOptions); err != nil {
+		t.Fatalf("apply: %v", err)
+	}
+	wfs.metaCache.WaitForEntryInvalidations()
+
+	if fh.dirtyPages != pagesBefore {
+		t.Fatal("dirty pages destroyed by an unmergeable peer change")
+	}
+	if got := len(fh.GetEntry().GetEntry().Chunks); got != 2 {
+		t.Fatalf("handle has %d chunks, want its own 2", got)
+	}
+	if base := fh.baseEntry.Load(); len(base.GetChunks()) != 1 || len(base.GetContent()) != 0 {
+		t.Fatalf("base %v moved to the peer's entry", base)
+	}
+}
+
+// A peer event that moves a dirty handle's base keeps its unpublished
+// truncate and chunk rewrite, so a later merge still honors them.
+func TestPeerEventKeepsPendingTruncate(t *testing.T) {
+	wfs := newInvalidateTestWFS(t)
+
+	inode := wfs.inodeToPath.Lookup(util.FullPath("/dir/file"), time.Now().Unix(), false, false, 0, false)
+	fh, _, _, _ := wfs.fhMap.AcquireFileHandle(wfs, inode, flushConditionEntry(1, 2), 0, 0)
+	fh.dirtyMetadata = true
+	fh.noteTruncate(1)
+	fh.noteRewrite()
+
+	chmod := flushConditionEntry(1, 2)
+	chmod.Attributes.FileMode = 0600
+	resp := &filer_pb.SubscribeMetadataResponse{
+		Directory: "/dir",
+		TsNs:      1500,
+		EventNotification: &filer_pb.EventNotification{
+			OldEntry:      flushConditionEntry(1, 2),
+			NewEntry:      chmod,
+			NewParentPath: "/dir",
+		},
+	}
+	if err := wfs.metaCache.ApplyMetadataResponse(context.Background(), resp, meta_cache.SubscriberMetadataResponseApplyOptions); err != nil {
+		t.Fatalf("apply: %v", err)
+	}
+	wfs.metaCache.WaitForEntryInvalidations()
+
+	base := fh.baseEntry.Load()
+	if base.GetAttributes().GetFileMode() != 0600 {
+		t.Fatalf("base %v did not move to the peer's entry", base)
+	}
+	if got := fh.markSince(base); got.truncatedTo != 1 || !got.rewrote {
+		t.Fatalf("pending truncate %d, rewrite %v after the base moved", got.truncatedTo, got.rewrote)
+	}
+}
+
+// A peer truncate below a dirty handle's size waits for the flush: the
+// handle's buffered writes past the new end are not chunks yet.
+func TestPeerTruncateWaitsForBufferedWrites(t *testing.T) {
+	wfs := newInvalidateTestWFS(t)
+
+	inode := wfs.inodeToPath.Lookup(util.FullPath("/dir/file"), time.Now().Unix(), false, false, 0, false)
+	fh, _, _, _ := wfs.fhMap.AcquireFileHandle(wfs, inode, flushConditionEntry(1, 2), 0, 0)
+	fh.dirtyMetadata = true
+	pagesBefore := fh.dirtyPages
+
+	peer := flushConditionEntry(1)
+	peer.Attributes.FileSize = 1
+	resp := &filer_pb.SubscribeMetadataResponse{
+		Directory: "/dir",
+		TsNs:      1500,
+		EventNotification: &filer_pb.EventNotification{
+			OldEntry:      flushConditionEntry(1, 2),
+			NewEntry:      peer,
+			NewParentPath: "/dir",
+		},
+	}
+	if err := wfs.metaCache.ApplyMetadataResponse(context.Background(), resp, meta_cache.SubscriberMetadataResponseApplyOptions); err != nil {
+		t.Fatalf("apply: %v", err)
+	}
+	wfs.metaCache.WaitForEntryInvalidations()
+
+	if fh.dirtyPages != pagesBefore {
+		t.Fatal("dirty pages destroyed by a peer truncate")
+	}
+	if size := fh.GetEntry().GetEntry().Attributes.FileSize; size != 3 {
+		t.Fatalf("handle size %d, want its own 3 until the flush", size)
+	}
+	if base := fh.baseEntry.Load(); len(base.GetChunks()) != 2 {
+		t.Fatalf("base %v moved to the peer's truncate", base)
+	}
+}
+
+// A peer replacing the file at the handle's path (another inode) leaves the
+// dirty handle's writes and base alone instead of installing the newcomer,
+// and the handle's flush then fails without touching the replacement.
+func TestPeerReplacementKeepsDirtyHandle(t *testing.T) {
+	for _, name := range []string{"new-content", "same-content"} {
+		t.Run(name, func(t *testing.T) {
+			wfs := newInvalidateTestWFS(t)
+
+			replacement := flushConditionEntry(7)
+			if name == "same-content" {
+				replacement = flushConditionEntry(1)
+			}
+			replacement.Attributes.Inode = 99
+			fake := &contentFlushConditionFiler{entry: proto.Clone(replacement).(*filer_pb.Entry)}
+			startFakeFiler(t, wfs, fake)
+
+			inode := wfs.inodeToPath.Lookup(util.FullPath("/dir/file"), time.Now().Unix(), false, false, 0, false)
+			base := flushConditionEntry(1)
+			wfs.mapPbIdFromFilerToLocal(base)
+			fh, _, _, _ := wfs.fhMap.AcquireFileHandle(wfs, inode, base, 0, 0)
+			t.Cleanup(func() { fh.dirtyPages.Destroy() })
+			fh.AddChunks([]*filer_pb.FileChunk{{FileId: "1,0300000001", Offset: 2, Size: 1, ModifiedTsNs: 3}})
+			fh.dirtyMetadata = true
+			pagesBefore := fh.dirtyPages
+
+			resp := &filer_pb.SubscribeMetadataResponse{
+				Directory: "/dir",
+				TsNs:      1500,
+				EventNotification: &filer_pb.EventNotification{
+					OldEntry:      flushConditionEntry(1),
+					NewEntry:      replacement,
+					NewParentPath: "/dir",
+				},
+			}
+			if err := wfs.metaCache.ApplyMetadataResponse(context.Background(), resp, meta_cache.SubscriberMetadataResponseApplyOptions); err != nil {
+				t.Fatalf("apply: %v", err)
+			}
+			wfs.metaCache.WaitForEntryInvalidations()
+
+			if fh.dirtyPages != pagesBefore {
+				t.Fatal("dirty pages destroyed by a replacement")
+			}
+			if got := chunkFileKeys(fh.GetEntry().GetEntry()); len(got) != 2 || got[0] != 1 {
+				t.Fatalf("handle chunks %v, want its own", got)
+			}
+			if base := fh.baseEntry.Load(); base.GetAttributes().GetInode() != 42 {
+				t.Fatalf("base %v moved to the replacement", base)
+			}
+
+			if code := wfs.doFlush(context.Background(), fh, 1000, 20, false); code == fuse.OK {
+				t.Fatal("flush into the replacement acknowledged")
+			}
+			fake.mu.Lock()
+			defer fake.mu.Unlock()
+			if !proto.Equal(fake.entry, replacement) {
+				t.Fatalf("replacement changed: %v", fake.entry)
+			}
+			if !fh.dirtyMetadata {
+				t.Fatal("refused flush lost dirty metadata")
+			}
+		})
+	}
+}
+
+// A peer's metadata-only change reaches a dirty handle's live entry, so its
+// next flush does not publish the stale value back; the handle's own edits
+// stay on top.
+func TestPeerChmodReachesDirtyHandle(t *testing.T) {
+	for _, name := range []string{"peer-only", "both-changed"} {
+		t.Run(name, func(t *testing.T) {
+			wfs := newInvalidateTestWFS(t)
+
+			inode := wfs.inodeToPath.Lookup(util.FullPath("/dir/file"), time.Now().Unix(), false, false, 0, false)
+			fh, _, _, _ := wfs.fhMap.AcquireFileHandle(wfs, inode, flushConditionEntry(1), 0, 0)
+			fh.AddChunks([]*filer_pb.FileChunk{{FileId: "1,0300000001", Offset: 2, Size: 1, ModifiedTsNs: 3}})
+			want := uint32(0600)
+			if name == "both-changed" {
+				fh.UpdateEntry(func(e *filer_pb.Entry) { e.Attributes.FileMode = 0640 })
+				want = 0640
+			}
+			fh.dirtyMetadata = true
+			pagesBefore := fh.dirtyPages
+
+			peer := flushConditionEntry(1)
+			peer.Attributes.FileMode = 0600
+			resp := &filer_pb.SubscribeMetadataResponse{
+				Directory: "/dir",
+				TsNs:      1500,
+				EventNotification: &filer_pb.EventNotification{
+					OldEntry:      flushConditionEntry(1),
+					NewEntry:      peer,
+					NewParentPath: "/dir",
+				},
+			}
+			if err := wfs.metaCache.ApplyMetadataResponse(context.Background(), resp, meta_cache.SubscriberMetadataResponseApplyOptions); err != nil {
+				t.Fatalf("apply: %v", err)
+			}
+			wfs.metaCache.WaitForEntryInvalidations()
+
+			if fh.dirtyPages != pagesBefore {
+				t.Fatal("dirty pages destroyed by a peer chmod")
+			}
+			entry := fh.GetEntry().GetEntry()
+			if got := entry.Attributes.FileMode; got != want || len(entry.Chunks) != 2 {
+				t.Fatalf("handle mode %o with %d chunks, want %o with its own chunk", got, len(entry.Chunks), want)
+			}
+			if base := fh.baseEntry.Load(); base.GetAttributes().GetFileMode() != 0600 {
+				t.Fatalf("base %v, want the peer's entry", base)
+			}
+		})
+	}
+}
+
+// Inline content from a peer would be read ahead of the chunk a buffered
+// write becomes, so it waits for the flush instead of merging.
+func TestPeerInlineContentWaitsForBufferedWrite(t *testing.T) {
+	wfs := newInvalidateTestWFS(t)
+
+	empty := func() *filer_pb.Entry {
+		e := flushConditionEntry()
+		e.Attributes.FileSize = 0
+		return e
+	}
+	inode := wfs.inodeToPath.Lookup(util.FullPath("/dir/file"), time.Now().Unix(), false, false, 0, false)
+	fh, _, _, _ := wfs.fhMap.AcquireFileHandle(wfs, inode, empty(), 0, 0)
+	fh.UpdateEntry(func(e *filer_pb.Entry) { e.Attributes.FileSize = 4 })
+	fh.dirtyMetadata = true
+
+	peer := empty()
+	peer.Content, peer.Attributes.FileSize = []byte("peer"), 4
+	resp := &filer_pb.SubscribeMetadataResponse{
+		Directory: "/dir",
+		TsNs:      1500,
+		EventNotification: &filer_pb.EventNotification{
+			OldEntry:      empty(),
+			NewEntry:      peer,
+			NewParentPath: "/dir",
+		},
+	}
+	if err := wfs.metaCache.ApplyMetadataResponse(context.Background(), resp, meta_cache.SubscriberMetadataResponseApplyOptions); err != nil {
+		t.Fatalf("apply: %v", err)
+	}
+	wfs.metaCache.WaitForEntryInvalidations()
+
+	if content := fh.GetEntry().GetEntry().Content; len(content) != 0 {
+		t.Fatalf("handle took the peer's inline content %q over a buffered write", content)
+	}
+	if base := fh.baseEntry.Load(); len(base.GetContent()) != 0 {
+		t.Fatalf("base %v moved to the peer's entry", base)
+	}
+}

weed/mount/weedfs_metadata_flush.go

@@ -179,10 +179,14 @@ func (wfs *WFS) flushFileMetadata(fh *FileHandle) error {
 	baseSnapshot := proto.Clone(requestEntry).(*filer_pb.Entry)
 	wfs.mapPbIdFromLocalToFiler(request.Entry)
 
-	resp, err := wfs.createContentEntry(context.Background(), fh, request)
+	resp, merged, err := wfs.createContentEntry(context.Background(), fh, request, baseSnapshot, true)
 	if err != nil {
 		return err
 	}
+	if merged != nil {
+		fh.rebaseEntry(merged, requestEntry)
+		baseSnapshot = merged
+	}
 
 	event := resp.GetMetadataEvent()
 	if event == nil {