diff --git a/weed/mount/content_flush_condition_test.go b/weed/mount/content_flush_condition_test.go index 4eb719a253..1862a8f75c 100644 --- a/weed/mount/content_flush_condition_test.go +++ b/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") } } diff --git a/weed/mount/filehandle.go b/weed/mount/filehandle.go index 627453fa17..5b40622502 100644 --- a/weed/mount/filehandle.go +++ b/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) diff --git a/weed/mount/weedfs_file_sync.go b/weed/mount/weedfs_file_sync.go index 89f5652858..507f1352c0 100644 --- a/weed/mount/weedfs_file_sync.go +++ b/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 diff --git a/weed/mount/weedfs_metadata_flush.go b/weed/mount/weedfs_metadata_flush.go index fdcafa0ef5..636a47c8d2 100644 --- a/weed/mount/weedfs_metadata_flush.go +++ b/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 {