diff --git a/weed/mount/filehandle.go b/weed/mount/filehandle.go index 6293797254..722df908b1 100644 --- a/weed/mount/filehandle.go +++ b/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) } diff --git a/weed/mount/weedfs_file_sync.go b/weed/mount/weedfs_file_sync.go index a6c21d6f69..aa01f71c72 100644 --- a/weed/mount/weedfs_file_sync.go +++ b/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) diff --git a/weed/mount/weedfs_invalidate_open_handle_test.go b/weed/mount/weedfs_invalidate_open_handle_test.go index 7aa6c154b3..3275d8712c 100644 --- a/weed/mount/weedfs_invalidate_open_handle_test.go +++ b/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) + } +} diff --git a/weed/mount/weedfs_metadata_flush.go b/weed/mount/weedfs_metadata_flush.go index 1b5fe38cd9..7c023067d7 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, 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 }