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>

weed/mount/filehandle.go

@@ -223,13 +223,16 @@ 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.
+// 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.
 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)
 	if ok {
 		fh.rebaseEntry(merged, local)
 	}

weed/mount/weedfs_file_sync.go

@@ -281,7 +281,7 @@ func (wfs *WFS) flushMetadataToFiler(ctx context.Context, fh *FileHandle, dir, n
 	baseSnapshot := proto.Clone(requestEntry).(*filer_pb.Entry)
 	wfs.mapPbIdFromLocalToFiler(request.Entry)
 
-	resp, merged, err := wfs.createContentEntry(ctx, fh, request, baseSnapshot)
+	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)
@@ -324,7 +324,9 @@ 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.
-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) {
+// With writes still buffered, a merge that would shrink the file is refused:
+// the buffered range is not in local's chunks yet.
+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 {
@@ -356,7 +358,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 {
+		if !ok || buffered && next.Attributes.FileSize < filer.FileSize(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

@@ -2194,3 +2194,40 @@ func TestPeerEventKeepsPendingTruncate(t *testing.T) {
 		t.Fatalf("pending truncate %d after the base moved, want 1", got)
 	}
 }
+
+// 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)
+	}
+}

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, merged, err := wfs.createContentEntry(context.Background(), fh, request, baseSnapshot)
+	resp, merged, err := wfs.createContentEntry(context.Background(), fh, request, baseSnapshot, true)
 	if err != nil {
 		return err
 	}