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>

weed/mount/content_flush_condition_test.go

@@ -4,6 +4,7 @@ package mount
 
 import (
 	"context"
+	"slices"
 	"sync"
 	"testing"
 
@@ -68,9 +69,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 +80,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 +136,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 +184,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 +208,104 @@ 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)
+	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)
+	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)
+	}
+
+	stored.Attributes.Inode = 99
+	if _, ok := mergeContentEntry(stored, base, local, local); ok {
+		t.Fatal("merged into a replaced file")
 	}
 }
 

weed/mount/filehandle.go

@@ -174,6 +174,30 @@ 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 uploaders attached after sent was taken, and refreshes the read view.
+func (fh *FileHandle) rebaseEntry(merged *filer_pb.Entry, sent []*filer_pb.FileChunk) {
+	inSent := make(map[string]bool, len(sent))
+	for _, c := range sent {
+		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 > attrs.FileSize {
+			attrs.FileSize = size
+		}
+		e.Attributes, e.Extended, e.Content, e.Chunks = attrs, merged.Extended, merged.Content, chunks
+		fh.entryChunkGroup.SetChunks(chunks)
+	})
+}
+
 // setAuthoritativeBase installs the base snapshot a local ack acknowledged.
 func (fh *FileHandle) setAuthoritativeBase(base *filer_pb.Entry) {
 	fh.baseEntry.Store(base)

weed/mount/weedfs_file_sync.go

@@ -281,11 +281,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)
 	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.Chunks)
+		baseSnapshot = merged
+	}
 
 	event := resp.GetMetadataEvent()
 	if event == nil {
@@ -312,16 +316,15 @@ 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.
+func (wfs *WFS) createContentEntry(ctx context.Context, fh *FileHandle, request *filer_pb.CreateEntryRequest, local *filer_pb.Entry) (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 +332,14 @@ 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
 			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 +347,141 @@ 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 stored == nil {
+			return resp, merged, err
+		}
+		next, ok := mergeContentEntry(stored, base, local, sent)
+		if !ok {
+			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
+// its size and clips stored chunks to it; otherwise the larger size wins.
+// A different file at the path, or inline content changed on both sides,
+// cannot be merged.
+func mergeContentEntry(stored, base, local, sent *filer_pb.Entry) (*filer_pb.Entry, bool) {
+	if stored == nil || base == nil || stored.IsDirectory || !bytes.Equal(stored.HardLinkId, base.HardLinkId) {
+		return nil, false
 	}
-	return proto.Equal(current, observed)
+	if inode := base.GetAttributes().GetInode(); inode != 0 && stored.GetAttributes().GetInode() != inode {
+		return nil, false
+	}
+	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
+	}
+	merged.Chunks = merged.Chunks[:0]
+	for _, c := range stored.Chunks {
+		if key := contentChunkKey(c); !inBase[key] || inLocal[key] {
+			merged.Chunks = append(merged.Chunks, proto.Clone(c).(*filer_pb.FileChunk))
+		}
+	}
+	for _, c := range local.Chunks {
+		if !inBase[contentChunkKey(c)] {
+			merged.Chunks = append(merged.Chunks, proto.Clone(c).(*filer_pb.FileChunk))
+		}
+	}
+
+	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)
+		}
+	}
+
+	localSize := filer.FileSize(local)
+	if localSize < filer.FileSize(base) {
+		clipped := merged.Chunks[:0]
+		for _, c := range merged.Chunks {
+			if c.Offset >= int64(localSize) {
+				continue
+			}
+			if c.Offset+int64(c.Size) > int64(localSize) {
+				c.Size = localSize - uint64(c.Offset)
+			}
+			clipped = append(clipped, c)
+		}
+		merged.Chunks = clipped
+		merged.Attributes.FileSize = localSize
+	} else {
+		merged.Attributes.FileSize = uint64(max(int64(localSize), int64(filer.FileSize(stored))))
+	}
+	return merged, true
+}
+
+// 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 {
+	fid := c.Fid
+	if fid == nil {
+		if parsed, err := filer_pb.ToFileIdObject(c.FileId); err == nil {
+			fid = parsed
+		} else {
+			return fmt.Sprintf("%s@%d+%d", c.FileId, c.Offset, c.Size)
+		}
+	}
+	return fmt.Sprintf("%d,%x,%x@%d+%d", fid.VolumeId, fid.FileKey, fid.Cookie, c.Offset, c.Size)
 }
 
 // ackedBase is the base an acknowledged create leaves: the entry the filer

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)
 	if err != nil {
 		return err
 	}
+	if merged != nil {
+		fh.rebaseEntry(merged, requestEntry.Chunks)
+		baseSnapshot = merged
+	}
 
 	event := resp.GetMetadataEvent()
 	if event == nil {