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>

weed/mount/filehandle.go

@@ -1,6 +1,7 @@
 package mount
 
 import (
+	"bytes"
 	"os"
 	"sync"
 	"sync/atomic"
@@ -222,23 +223,28 @@ func (fh *FileHandle) refreshChunkView() {
 	})
 }
 
-// mergePeerContent folds a peer's content change into a dirty handle, so its
-// own writes stay on top of the peer's instead of being discarded. Buffered
-// writes are not chunks yet, so a merge that would shrink the file waits for
-// the flush.
+// 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)
 	})
 	merged, ok := mergeContentEntry(peer, base, local, local, fh.truncatedSince(base))
-	ok = ok && merged.Attributes.FileSize >= filer.FileSize(local)
+	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)
+}
+
 type truncateMark struct {
 	base *filer_pb.Entry
 	size int64

weed/mount/weedfs_file_sync.go

@@ -324,8 +324,7 @@ func (wfs *WFS) flushMetadataToFiler(ctx context.Context, fh *FileHandle, dir, n
 // 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, a merge that would shrink the file is refused:
-// the buffered range is not in local's chunks yet.
+// 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}
@@ -358,7 +357,7 @@ func (wfs *WFS) createContentEntry(ctx context.Context, fh *FileHandle, request
 			return resp, merged, err
 		}
 		next, ok := mergeContentEntry(stored, base, local, sent, fh.truncatedSince(base))
-		if !ok || buffered && next.Attributes.FileSize < filer.FileSize(local) {
+		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)

weed/mount/weedfs_invalidate_open_handle_test.go

@@ -2325,3 +2325,42 @@ func TestPeerChmodReachesDirtyHandle(t *testing.T) {
 		})
 	}
 }
+
+// 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)
+	}
+}