diff --git a/weed/mount/content_flush_condition_test.go b/weed/mount/content_flush_condition_test.go index 4eb719a253..7c7e4d3b4c 100644 --- a/weed/mount/content_flush_condition_test.go +++ b/weed/mount/content_flush_condition_test.go @@ -1,13 +1,15 @@ -//go:build !freebsd - package mount import ( "context" + "errors" + "slices" "sync" "testing" + "time" "github.com/seaweedfs/go-fuse/v2/fuse" + "github.com/seaweedfs/seaweedfs/weed/filer" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/util" "google.golang.org/grpc/codes" @@ -51,6 +53,17 @@ func (s *contentFlushConditionFiler) CreateEntry(ctx context.Context, r *filer_p }, }}, nil } +func (s *contentFlushConditionFiler) UpdateEntry(ctx context.Context, r *filer_pb.UpdateEntryRequest) (*filer_pb.UpdateEntryResponse, error) { + s.mu.Lock() + defer s.mu.Unlock() + if clauses := r.GetCondition().GetClauses(); len(clauses) > 0 && !proto.Equal(clauses[0].GetExpectedEntry(), s.entry) { + return nil, status.Error(codes.FailedPrecondition, "entry changed") + } + s.calls++ + s.entry = proto.Clone(r.Entry).(*filer_pb.Entry) + s.entry.CommitTsNs = 1000 + int64(s.calls) + return &filer_pb.UpdateEntryResponse{LogTsNs: 2000}, nil +} func (s *contentFlushConditionFiler) LookupDirectoryEntry(ctx context.Context, r *filer_pb.LookupDirectoryEntryRequest) (*filer_pb.LookupDirectoryEntryResponse, error) { s.mu.Lock() defer s.mu.Unlock() @@ -68,9 +81,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 +92,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 +148,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 +196,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 +220,182 @@ 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, -1, false) + 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, -1, false) + 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) + } + + // Truncated to 1, then written at 7: the peer chunk at 2 stays cut. + regrown := proto.Clone(truncated).(*filer_pb.Entry) + regrown.Chunks = append(regrown.Chunks, &filer_pb.FileChunk{Fid: &filer_pb.FileId{VolumeId: 1, FileKey: 7, Cookie: 1}, Offset: 7, Size: 1, ModifiedTsNs: 7}) + regrown.Attributes.FileSize = 8 + merged, ok = mergeContentEntry(stored, base, regrown, regrown, 1, false) + if !ok { + t.Fatal("truncate then write not merged") + } + if got := chunkFileKeys(merged); !slices.Equal(got, []uint64{1, 7}) || merged.Attributes.FileSize != 8 { + t.Fatalf("truncate then write kept %v, size %d", got, merged.Attributes.FileSize) + } + + // Shortening a manifest reference leaves its members whole. + peerManifest := proto.Clone(base).(*filer_pb.Entry) + peerManifest.Chunks = append(peerManifest.Chunks, &filer_pb.FileChunk{Fid: &filer_pb.FileId{VolumeId: 1, FileKey: 5, Cookie: 1}, Offset: 0, Size: 10, IsChunkManifest: true}) + if _, ok := mergeContentEntry(peerManifest, base, regrown, regrown, 1, false); ok { + t.Fatal("merged a peer manifest across the truncate") + } + + // A peer truncate stands when the handle only overwrote inside it. + peerTruncated := proto.Clone(base).(*filer_pb.Entry) + peerTruncated.Chunks, peerTruncated.Attributes.FileSize = peerTruncated.Chunks[:1], 1 + overwrote := proto.Clone(base).(*filer_pb.Entry) + overwrote.Chunks = append(overwrote.Chunks, &filer_pb.FileChunk{Fid: &filer_pb.FileId{VolumeId: 1, FileKey: 6, Cookie: 1}, Size: 1, ModifiedTsNs: 6}) + merged, ok = mergeContentEntry(peerTruncated, base, overwrote, overwrote, -1, false) + if !ok { + t.Fatal("overwrite over a peer truncate not merged") + } + if got := chunkFileKeys(merged); !slices.Equal(got, []uint64{1, 6}) || merged.Attributes.FileSize != 1 { + t.Fatalf("overwrite over a peer truncate kept %v, size %d", got, merged.Attributes.FileSize) + } + + // A rewrite folded into a manifest is still a rewrite. + rewritten := proto.Clone(base).(*filer_pb.Entry) + rewritten.Chunks = []*filer_pb.FileChunk{{Fid: &filer_pb.FileId{VolumeId: 1, FileKey: 8, Cookie: 1}, Size: 2, IsChunkManifest: true}} + if _, ok := mergeContentEntry(stored, base, rewritten, rewritten, -1, true); ok { + t.Fatal("merged a stale whole-file rewrite over a peer append") + } + if _, ok := mergeContentEntry(stored, base, rewritten, rewritten, -1, false); !ok { + t.Fatal("refused new chunks the handle did not rewrite") + } + if _, ok := mergeContentEntry(proto.Clone(base).(*filer_pb.Entry), base, rewritten, rewritten, -1, true); !ok { + t.Fatal("refused a rewrite with no peer chunk change") + } + + stored.Attributes.Inode = 99 + if _, ok := mergeContentEntry(stored, base, local, local, -1, false); ok { + t.Fatal("merged into a replaced file") + } + unstamped := flushConditionEntry(4) + unstamped.Chunks[0].Size, unstamped.Chunks[0].ModifiedTsNs = 3, 0 + shortened := proto.Clone(unstamped).(*filer_pb.Entry) + shortened.Chunks[0].Size, shortened.Attributes.FileSize = 2, 2 + peerAppend := proto.Clone(unstamped).(*filer_pb.Entry) + peerAppend.Chunks = append(peerAppend.Chunks, &filer_pb.FileChunk{Fid: &filer_pb.FileId{VolumeId: 1, FileKey: 9, Cookie: 1}, Offset: 3, Size: 1, ModifiedTsNs: 9}) + if _, ok := mergeContentEntry(peerAppend, unstamped, shortened, shortened, 2, false); !ok { + t.Fatal("refused a truncate that shortened an unstamped base chunk") + } + + noInode := proto.Clone(base).(*filer_pb.Entry) + noInode.Attributes.Inode, noInode.Attributes.Crtime = 0, 5 + sameFile := proto.Clone(noInode).(*filer_pb.Entry) + sameFile.Chunks = append(sameFile.Chunks, stored.Chunks[2]) + if _, ok := mergeContentEntry(sameFile, noInode, local, local, -1, false); !ok { + t.Fatal("refused an inode-less file with the same creation time") + } + sameFile.Attributes.Crtime = 6 + if _, ok := mergeContentEntry(sameFile, noInode, local, local, -1, false); ok { + t.Fatal("merged into an inode-less replacement") + } + + empty := flushConditionEntry() + inline := proto.Clone(empty).(*filer_pb.Entry) + inline.Content = []byte("peer") + appended := proto.Clone(empty).(*filer_pb.Entry) + appended.Chunks = flushConditionEntry(5).Chunks + if _, ok := mergeContentEntry(inline, empty, appended, appended, -1, false); ok { + t.Fatal("merged chunks next to inline content") + } + if _, ok := mergeContentEntry(appended, empty, inline, inline, -1, false); ok { + t.Fatal("merged inline content next to chunks") } } @@ -246,3 +439,59 @@ func TestContentFlushRepublishesUnchangedEntry(t *testing.T) { }) } } + +// Rebasing keeps a size the handle reached after the flush was taken, not the +// size it sent: a peer's truncate merged in must stand. +func TestRebaseEntryKeepsMergedSize(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) + sent := proto.Clone(fh.GetEntry().GetEntry()).(*filer_pb.Entry) + merged := flushConditionEntry(1) + merged.Attributes.FileSize = 1 + + fh.rebaseEntry(merged, sent) + if size := fh.GetEntry().GetEntry().Attributes.FileSize; size != 1 { + t.Fatalf("size %d after rebasing onto a truncate, want 1", size) + } + + fh.GetEntry().GetEntry().Attributes.FileSize = 5 + fh.rebaseEntry(merged, sent) + if size := fh.GetEntry().GetEntry().Attributes.FileSize; size != 5 { + t.Fatalf("size %d after a write grew the handle, want 5", size) + } +} + +// A merged entry whose manifest cannot be resolved yet leaves the read view +// marked, and the next read resolves it again instead of failing for good. +func TestRebaseEntryRetriesUnresolvedChunkView(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), 0, 0) + lookupDown := true + group, err := filer.NewChunkGroup(func(ctx context.Context, fileId string) ([]string, error) { + if lookupDown { + return nil, errors.New("volume lookup down") + } + return nil, nil + }, nil, nil, 1, nil, nil) + if err != nil { + t.Fatal(err) + } + fh.entryChunkGroup = group + + sent := proto.Clone(fh.GetEntry().GetEntry()).(*filer_pb.Entry) + merged := flushConditionEntry(1, 2) + merged.Chunks[1].IsChunkManifest = true + fh.rebaseEntry(merged, sent) + if !fh.chunkViewStale.Load() { + t.Fatal("unresolved manifest not marked") + } + + lookupDown = false + fh.UpdateEntry(func(e *filer_pb.Entry) { e.Chunks = e.Chunks[:1] }) + fh.refreshChunkView() + if fh.chunkViewStale.Load() { + t.Fatal("read view still marked after the chunks resolved") + } +} diff --git a/weed/mount/filehandle.go b/weed/mount/filehandle.go index 627453fa17..e606a0a842 100644 --- a/weed/mount/filehandle.go +++ b/weed/mount/filehandle.go @@ -1,6 +1,7 @@ package mount import ( + "bytes" "os" "sync" "sync/atomic" @@ -66,6 +67,15 @@ type FileHandle struct { // must be judged here, not against the live entry. Always store a clone. baseEntry atomic.Pointer[filer_pb.Entry] + // sinceBase marks a truncate or chunk rewrite over baseEntry that a + // merge must honor: later writes can grow the file past a truncate, and + // rewritten chunks carry the handle's view of the base. + sinceBase atomic.Pointer[baseMark] + + // chunkViewStale marks a read view whose manifest resolution failed; the + // next read or lseek resolves the entry's chunks again. + chunkViewStale atomic.Bool + // dlmLock holds the distributed lock for cross-mount write coordination. // Non-nil only when -dlm is enabled and the file was opened for writing. // Acquired in AcquireHandle, released in ReleaseHandle. @@ -141,6 +151,7 @@ func (fh *FileHandle) SetEntry(entry *filer_pb.Entry) { if resolveManifestErr != nil { glog.Warningf("failed to resolve manifest chunks in %+v", filer_pb.ScrubCipherKeys(entry)) } + fh.chunkViewStale.Store(resolveManifestErr != nil) } else { glog.Fatalf("setting file handle entry to nil") } @@ -174,6 +185,125 @@ 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 and growth writers added after sent was taken, and refreshes the +// read view. +func (fh *FileHandle) rebaseEntry(merged, sent *filer_pb.Entry) { + inSent := make(map[string]bool, len(sent.Chunks)) + for _, c := range sent.Chunks { + 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 > sent.GetAttributes().GetFileSize() && size > attrs.FileSize { + attrs.FileSize = size + } + e.Attributes, e.Extended, e.Content, e.Chunks = attrs, merged.Extended, merged.Content, chunks + err := fh.entryChunkGroup.SetChunks(chunks) + if err != nil { + glog.Warningf("rebase %s: resolve chunks: %v", fh.FullPath(), err) + } + fh.chunkViewStale.Store(err != nil) + }) +} + +// refreshChunkView retries a read view whose manifest resolution failed. +func (fh *FileHandle) refreshChunkView() { + if !fh.chunkViewStale.Load() { + return + } + fh.UpdateEntry(func(e *filer_pb.Entry) { + fh.chunkViewStale.Store(fh.entryChunkGroup.SetChunks(e.Chunks) != nil) + }) +} + +// 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) + }) + mark := fh.markSince(base) + merged, ok := mergeContentEntry(peer, base, local, local, mark.truncatedTo, mark.rewrote) + 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) +} + +// baseMark records what the handle did over base that its entry does not +// show: the lowest size a truncate set (-1 for none), and a whole-file +// rewrite of its chunks. +type baseMark struct { + base *filer_pb.Entry + truncatedTo int64 + rewrote bool +} + +func (fh *FileHandle) markBase(update func(m *baseMark)) { + for { + base, old := fh.baseEntry.Load(), fh.sinceBase.Load() + next := baseMark{base: base, truncatedTo: -1} + if old != nil && old.base == base { + next = *old + } + update(&next) + if fh.sinceBase.CompareAndSwap(old, &next) { + return + } + } +} + +// noteTruncate records a truncate to size over the current base. +func (fh *FileHandle) noteTruncate(size uint64) { + fh.markBase(func(m *baseMark) { + if m.truncatedTo < 0 || int64(size) < m.truncatedTo { + m.truncatedTo = int64(size) + } + }) +} + +// noteRewrite records that the handle's chunks were rewritten from its view +// of the current base. +func (fh *FileHandle) noteRewrite() { + fh.markBase(func(m *baseMark) { m.rewrote = true }) +} + +// adoptPeerBase moves a dirty handle's base to a peer's entry. What the +// handle has not published yet still applies over the new base. +func (fh *FileHandle) adoptPeerBase(base *filer_pb.Entry) { + old := fh.baseEntry.Swap(base) + if m := fh.sinceBase.Load(); m != nil && m.base == old { + moved := *m + moved.base = base + fh.sinceBase.CompareAndSwap(m, &moved) + } +} + +// markSince is what the handle did over base. +func (fh *FileHandle) markSince(base *filer_pb.Entry) baseMark { + if m := fh.sinceBase.Load(); m != nil && m.base == base { + return *m + } + return baseMark{base: base, truncatedTo: -1} +} + // 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/filehandle_read.go b/weed/mount/filehandle_read.go index a00e2d606f..47e785fe5b 100644 --- a/weed/mount/filehandle_read.go +++ b/weed/mount/filehandle_read.go @@ -100,6 +100,7 @@ func (fh *FileHandle) readFromChunksWithContext(ctx context.Context, buff []byte } // Fall back to normal chunk reading + fh.refreshChunkView() totalRead, ts, err := fh.entryChunkGroup.ReadDataAt(ctx, fileSize, buff, offset) if err != nil && err != io.EOF { diff --git a/weed/mount/weedfs.go b/weed/mount/weedfs.go index 5cd89e448a..9a85bf8476 100644 --- a/weed/mount/weedfs.go +++ b/weed/mount/weedfs.go @@ -1044,18 +1044,24 @@ func (wfs *WFS) invalidateOpenFileHandle(invalidation meta_cache.EntryInvalidati fh.advanceEntryVersion(candidateTsNs, 0) return } + // A dirty handle replays its own changes onto the peer's entry, keeping + // its pages. An unmergeable change or a replaced file keeps the writes; + // the flush reports the conflict. + if fh.dirtyMetadata && base != nil { + if sameContentFile(candidate, base) && fh.mergePeerContent(candidate, base) { + fh.adoptPeerBase(proto.Clone(candidate).(*filer_pb.Entry)) + fh.advanceEntryVersion(candidateTsNs, 0) + } + return + } // Dirty pages overlay content, so only a content change invalidates them: // a metadata-only event (chmod, touch, or a committed copy's own event - // against an approximate base) leaves them valid. A dirty handle likewise - // keeps its diverged entry unless foreign content supersedes it. - contentChanged := base == nil || !sameEntryContent(candidate, base) - if contentChanged { + // against an approximate base) leaves them valid. + if base == nil || !sameEntryContent(candidate, base) { fh.dirtyPages.Destroy() fh.dirtyPages = newPageWriter(fh, wfs.option.ChunkSizeLimit) } - if contentChanged || !fh.dirtyMetadata { - fh.SetEntry(candidate) - } + fh.SetEntry(candidate) fh.baseEntry.Store(proto.Clone(candidate).(*filer_pb.Entry)) fh.advanceEntryVersion(candidateTsNs, 0) return diff --git a/weed/mount/weedfs_attr.go b/weed/mount/weedfs_attr.go index b394912d9e..c3f610bf41 100644 --- a/weed/mount/weedfs_attr.go +++ b/weed/mount/weedfs_attr.go @@ -134,6 +134,9 @@ func (wfs *WFS) SetAttr(cancel <-chan struct{}, input *fuse.SetAttrIn, out *fuse wfs.invalidateOpenMtimeCache(input.NodeId) oldFileSize := filer.FileSize(entry) if size < oldFileSize { + if fh != nil { + fh.noteTruncate(size) + } // fmt.Printf("truncate %v \n", fullPath) var chunks []*filer_pb.FileChunk var truncatedChunks []*filer_pb.FileChunk diff --git a/weed/mount/weedfs_attr_cas_test.go b/weed/mount/weedfs_attr_cas_test.go index ac6a0bd491..2fdede2d2c 100644 --- a/weed/mount/weedfs_attr_cas_test.go +++ b/weed/mount/weedfs_attr_cas_test.go @@ -138,6 +138,82 @@ func attributeAppendEntry(records int) *filer_pb.Entry { return e } +// A peer's acknowledged attribute change leaves the content alone, so our +// append's conditional flush must still publish, keeping the peer's change. +func TestAppendFlushAfterPeerAttributeChange(t *testing.T) { + for _, op := range []string{"none", "chmod", "chown", "utimes", "setxattr"} { + t.Run(op, func(t *testing.T) { + writer, peer := newInvalidateTestWFS(t), newInvalidateTestWFS(t) + peer.atimeMap = make(map[uint64]time.Time) + base := attributeAppendEntry(1) + base.CommitTsNs = 100 + fake := &contentFlushConditionFiler{entry: proto.Clone(base).(*filer_pb.Entry)} + startFakeFiler(t, writer, fake) + startFakeFiler(t, peer, fake) + path := util.FullPath("/dir/file") + local := proto.Clone(base).(*filer_pb.Entry) + writer.mapPbIdFromFilerToLocal(local) + fh := newFileHandle(writer, 1, 42, local) + fh.RememberPath(path) + t.Cleanup(fh.dirtyPages.Destroy) + writer.fhMap.inode2fh[42], writer.fhMap.fh2inode[1] = fh, 42 + appended := attributeAppendEntry(2).Chunks[1] + fh.UpdateEntry(func(e *filer_pb.Entry) { + e.Chunks = append(e.Chunks, appended) + e.Attributes.FileSize = 512 + }) + fh.dirtyMetadata = true + + peer.inodeToPath.Lookup("/dir", 1, true, false, 0, false) + peer.inodeToPath.Lookup(path, 1, false, false, 42, true) + peer.inodeToPath.MarkChildrenCached("/dir") + if err := peer.metaCache.InsertEntry(context.Background(), filer.FromPbEntry("/dir", base), 1); err != nil { + t.Fatal(err) + } + code := fuse.OK + in := &fuse.SetAttrIn{} + in.NodeId = 42 + switch op { + case "chmod": + in.Valid, in.Mode = fuse.FATTR_MODE, 0600 + case "chown": + in.Valid, in.Owner.Uid, in.Owner.Gid = fuse.FATTR_UID|fuse.FATTR_GID, 1001, 21 + case "utimes": + in.Valid, in.Mtime = fuse.FATTR_MTIME, 123 + case "setxattr": + code = peer.SetXAttr(nil, &fuse.SetXAttrIn{InHeader: fuse.InHeader{NodeId: 42}}, "user.test", []byte("peer")) + } + if in.Valid != 0 { + code = peer.SetAttr(nil, in, &fuse.AttrOut{}) + } + if code != fuse.OK { + t.Fatalf("peer %s: %v", op, code) + } + + if code = writer.Fsync(nil, &fuse.FsyncIn{InHeader: fuse.InHeader{NodeId: 42}, Fh: 1}); code != fuse.OK { + t.Fatalf("append fsync after peer %s: %v", op, code) + } + fake.mu.Lock() + defer fake.mu.Unlock() + got := fake.entry + if len(got.Chunks) != 2 || got.Attributes.FileSize != 512 { + t.Fatalf("append not published after peer %s: %v", op, got) + } + a := got.Attributes + kept := map[string]bool{ + "none": true, + "chmod": a.FileMode&0777 == 0600, + "chown": a.Uid == 1001 && a.Gid == 21, + "utimes": a.Mtime == 123, + "setxattr": string(got.Extended["xattr-user.test"]) == "peer", + }[op] + if !kept { + t.Fatalf("append flush dropped the peer's %s: %v", op, got) + } + }) + } +} + func TestAttributeOnlyOpsPreserveAcknowledgedAppends(t *testing.T) { for _, op := range []string{"chmod", "chown", "utimes", "setxattr", "removexattr"} { for _, stale := range []bool{false, true} { diff --git a/weed/mount/weedfs_file_lseek.go b/weed/mount/weedfs_file_lseek.go index 06ebfc1446..685a5c7d56 100644 --- a/weed/mount/weedfs_file_lseek.go +++ b/weed/mount/weedfs_file_lseek.go @@ -70,6 +70,7 @@ func (wfs *WFS) Lseek(cancel <-chan struct{}, in *fuse.LseekIn, out *fuse.LseekO }() // search chunks for the offset + fh.refreshChunkView() found, offset, err := fh.entryChunkGroup.SearchChunks(ctx, offset, fileSize, in.Whence) if err != nil { glog.Errorf("Lseek %s: %v", fh.FullPath(), err) diff --git a/weed/mount/weedfs_file_sync.go b/weed/mount/weedfs_file_sync.go index 89f5652858..adc95e0707 100644 --- a/weed/mount/weedfs_file_sync.go +++ b/weed/mount/weedfs_file_sync.go @@ -254,6 +254,7 @@ func (wfs *WFS) flushMetadataToFiler(ctx context.Context, fh *FileHandle, dir, n } else if mergedChunks != nil { chunks = mergedChunks manifestChunks = nil + fh.noteRewrite() } // no chunk deleter here: a failed fold reports the blobs it saved @@ -281,11 +282,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, false) 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) + baseSnapshot = merged + } event := resp.GetMetadataEvent() if event == nil { @@ -312,16 +317,16 @@ 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. +// 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} if base != nil { @@ -329,13 +334,15 @@ 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 + stored = nil 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 +350,175 @@ 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 err == nil || stored == nil { + return resp, merged, err + } + mark := fh.markSince(base) + next, ok := mergeContentEntry(stored, base, local, sent, mark.truncatedTo, mark.rewrote) + 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) + 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 +// the handle's size and clips stored chunks at the lowest size truncated to +// (truncatedTo, or -1); otherwise the stored size stands unless the handle +// changed its size or wrote past it. +// A different file at the path, inline content changed on both sides, +// inline content next to chunks (reads stop at the inline bytes), a manifest +// across the truncate (its members keep their sizes), or a local +// whole-file rewrite (rewrote) over a peer's chunk change cannot be merged: +// the rewrite carries the handle's stale view. +func mergeContentEntry(stored, base, local, sent *filer_pb.Entry, truncatedTo int64, rewrote bool) (*filer_pb.Entry, bool) { + if !sameContentFile(stored, base) { + return nil, false } - return proto.Equal(current, observed) + 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 + } + localSize := filer.FileSize(local) + if truncatedTo < 0 && localSize < filer.FileSize(base) { + truncatedTo = int64(localSize) + } + peerChanged := len(stored.Chunks) != len(base.Chunks) + merged.Chunks = merged.Chunks[:0] + for _, c := range stored.Chunks { + key := contentChunkKey(c) + peerChanged = peerChanged || !inBase[key] + if inBase[key] && !inLocal[key] || truncatedTo >= 0 && c.Offset >= truncatedTo { + continue + } + c = proto.Clone(c).(*filer_pb.FileChunk) + if truncatedTo >= 0 && c.Offset+int64(c.Size) > truncatedTo { + if c.IsChunkManifest { + return nil, false + } + c.Size = uint64(truncatedTo - c.Offset) + } + merged.Chunks = append(merged.Chunks, c) + } + if rewrote && peerChanged { + return nil, false + } + size := int64(filer.FileSize(stored)) + if localSize != filer.FileSize(base) { + size = max(size, int64(localSize)) + } + for _, c := range local.Chunks { + if !inBase[contentChunkKey(c)] { + merged.Chunks = append(merged.Chunks, proto.Clone(c).(*filer_pb.FileChunk)) + size = max(size, c.Offset+int64(c.Size)) + } + } + if len(merged.Content) > 0 && len(merged.Chunks) > 0 { + return nil, false + } + + 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) + } + } + + if truncatedTo >= 0 { + size = int64(localSize) + } + merged.Attributes.FileSize = uint64(size) + return merged, true +} + +// sameContentFile reports whether stored is still the regular file base was: +// the same inode, or for an entry without one, the same creation time. A +// synthesized base with neither cannot tell. +func sameContentFile(stored, base *filer_pb.Entry) bool { + if stored == nil || base == nil || stored.IsDirectory || !bytes.Equal(stored.HardLinkId, base.HardLinkId) { + return false + } + if inode := base.GetAttributes().GetInode(); inode != 0 { + return stored.GetAttributes().GetInode() == inode + } + crtime := base.GetAttributes().GetCrtime() + return crtime == 0 || stored.GetAttributes().GetCrtime() == crtime +} + +// 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 { + return fmt.Sprintf("%s@%d+%d", contentFileIdKey(c), c.Offset, c.Size) +} + +func contentFileIdKey(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 c.FileId + } + } + return fmt.Sprintf("%d,%x,%x", fid.VolumeId, fid.FileKey, fid.Cookie) } // ackedBase is the base an acknowledged create leaves: the entry the filer diff --git a/weed/mount/weedfs_invalidate_open_handle_test.go b/weed/mount/weedfs_invalidate_open_handle_test.go index 7cddb221de..5186c731e5 100644 --- a/weed/mount/weedfs_invalidate_open_handle_test.go +++ b/weed/mount/weedfs_invalidate_open_handle_test.go @@ -20,6 +20,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/util" "google.golang.org/grpc" "google.golang.org/grpc/credentials/insecure" + "google.golang.org/protobuf/proto" ) func newInvalidateTestWFS(t *testing.T) *WFS { @@ -1241,7 +1242,8 @@ func TestCommittedCopyWithFailedReadbackAdoptsItsEvent(t *testing.T) { t.Fatalf("handle version = %d, want 1500", got) } - // The adoption is one-shot: a genuinely foreign event still invalidates. + // The adoption is one-shot: a genuinely foreign event still installs, + // merged under the handle's own writes. if err := wfs.metaCache.ApplyMetadataResponse(context.Background(), updateEventFor("file", 300, 1600), meta_cache.SubscriberMetadataResponseApplyOptions); err != nil { t.Fatalf("apply foreign event: %v", err) } @@ -1249,8 +1251,11 @@ func TestCommittedCopyWithFailedReadbackAdoptsItsEvent(t *testing.T) { if size := fh.GetEntry().GetEntry().Attributes.FileSize; size != 300 { t.Fatalf("live entry file size = %d, want 300 (foreign event after adoption must install)", size) } - if fh.dirtyPages == pagesBefore { - t.Fatal("foreign event after adoption must invalidate dirty pages") + if base := fh.baseEntry.Load(); base.GetAttributes().GetFileSize() != 300 { + t.Fatalf("base %v, want the foreign entry", base) + } + if fh.dirtyPages != pagesBefore { + t.Fatal("dirty pages destroyed by a mergeable foreign event") } } @@ -2075,3 +2080,308 @@ func TestAckFromAnotherFilerIsNotDroppedBehindForeignFence(t *testing.T) { t.Fatal("an older ack from the fencing filer was installed; within one clock domain it must be refused") } } + +// A peer's append reaching a dirty handle is merged under the handle's own +// writes: the uploaded chunk and the unflushed pages both survive. +func TestPeerAppendMergesIntoDirtyHandle(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), 0, 0) + fh.AddChunks([]*filer_pb.FileChunk{{FileId: "1,0300000001", Offset: 2, Size: 1, ModifiedTsNs: 3}}) + fh.dirtyMetadata = true + pagesBefore := fh.dirtyPages + + peer := flushConditionEntry(1, 2) + peer.Attributes.FileSize = 4 + peer.Chunks[1].Offset, peer.Chunks[1].Size = 3, 1 + appendResp := &filer_pb.SubscribeMetadataResponse{ + Directory: "/dir", + TsNs: 1500, + EventNotification: &filer_pb.EventNotification{ + OldEntry: flushConditionEntry(1), + NewEntry: peer, + NewParentPath: "/dir", + }, + } + if err := wfs.metaCache.ApplyMetadataResponse(context.Background(), appendResp, meta_cache.SubscriberMetadataResponseApplyOptions); err != nil { + t.Fatalf("apply append: %v", err) + } + wfs.metaCache.WaitForEntryInvalidations() + + if fh.dirtyPages != pagesBefore { + t.Fatal("dirty pages destroyed by a peer append") + } + entry := fh.GetEntry().GetEntry() + if got := len(entry.Chunks); got != 3 || entry.Attributes.FileSize != 4 { + t.Fatalf("handle has %d chunks, size %d; want the peer's and its own", got, entry.Attributes.FileSize) + } + if base := fh.baseEntry.Load(); len(base.GetChunks()) != 2 { + t.Fatalf("base %v, want the peer's entry", base) + } +} + +// A peer change that cannot be merged (inline content over the handle's new +// chunk) leaves the dirty handle alone, so its flush fails instead of the +// event silently dropping the writes. +func TestUnmergeablePeerChangeKeepsDirtyHandle(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), 0, 0) + fh.AddChunks([]*filer_pb.FileChunk{{FileId: "1,0300000001", Offset: 2, Size: 1, ModifiedTsNs: 3}}) + fh.dirtyMetadata = true + pagesBefore := fh.dirtyPages + + peer := flushConditionEntry() + peer.Content = []byte("peer") + peer.Attributes.FileSize = 4 + resp := &filer_pb.SubscribeMetadataResponse{ + Directory: "/dir", + TsNs: 1500, + EventNotification: &filer_pb.EventNotification{ + OldEntry: flushConditionEntry(1), + 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 an unmergeable peer change") + } + if got := len(fh.GetEntry().GetEntry().Chunks); got != 2 { + t.Fatalf("handle has %d chunks, want its own 2", got) + } + if base := fh.baseEntry.Load(); len(base.GetChunks()) != 1 || len(base.GetContent()) != 0 { + t.Fatalf("base %v moved to the peer's entry", base) + } +} + +// A peer event that moves a dirty handle's base keeps its unpublished +// truncate and chunk rewrite, so a later merge still honors them. +func TestPeerEventKeepsPendingTruncate(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 + fh.noteTruncate(1) + fh.noteRewrite() + + chmod := flushConditionEntry(1, 2) + chmod.Attributes.FileMode = 0600 + resp := &filer_pb.SubscribeMetadataResponse{ + Directory: "/dir", + TsNs: 1500, + EventNotification: &filer_pb.EventNotification{ + OldEntry: flushConditionEntry(1, 2), + NewEntry: chmod, + NewParentPath: "/dir", + }, + } + if err := wfs.metaCache.ApplyMetadataResponse(context.Background(), resp, meta_cache.SubscriberMetadataResponseApplyOptions); err != nil { + t.Fatalf("apply: %v", err) + } + wfs.metaCache.WaitForEntryInvalidations() + + base := fh.baseEntry.Load() + if base.GetAttributes().GetFileMode() != 0600 { + t.Fatalf("base %v did not move to the peer's entry", base) + } + if got := fh.markSince(base); got.truncatedTo != 1 || !got.rewrote { + t.Fatalf("pending truncate %d, rewrite %v after the base moved", got.truncatedTo, got.rewrote) + } +} + +// 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) + } +} + +// A peer replacing the file at the handle's path (another inode) leaves the +// dirty handle's writes and base alone instead of installing the newcomer, +// and the handle's flush then fails without touching the replacement. +func TestPeerReplacementKeepsDirtyHandle(t *testing.T) { + for _, name := range []string{"new-content", "same-content"} { + t.Run(name, func(t *testing.T) { + wfs := newInvalidateTestWFS(t) + + replacement := flushConditionEntry(7) + if name == "same-content" { + replacement = flushConditionEntry(1) + } + replacement.Attributes.Inode = 99 + fake := &contentFlushConditionFiler{entry: proto.Clone(replacement).(*filer_pb.Entry)} + startFakeFiler(t, wfs, fake) + + inode := wfs.inodeToPath.Lookup(util.FullPath("/dir/file"), time.Now().Unix(), false, false, 0, false) + base := flushConditionEntry(1) + wfs.mapPbIdFromFilerToLocal(base) + fh, _, _, _ := wfs.fhMap.AcquireFileHandle(wfs, inode, base, 0, 0) + t.Cleanup(func() { fh.dirtyPages.Destroy() }) + fh.AddChunks([]*filer_pb.FileChunk{{FileId: "1,0300000001", Offset: 2, Size: 1, ModifiedTsNs: 3}}) + fh.dirtyMetadata = true + pagesBefore := fh.dirtyPages + + resp := &filer_pb.SubscribeMetadataResponse{ + Directory: "/dir", + TsNs: 1500, + EventNotification: &filer_pb.EventNotification{ + OldEntry: flushConditionEntry(1), + NewEntry: replacement, + 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 replacement") + } + if got := chunkFileKeys(fh.GetEntry().GetEntry()); len(got) != 2 || got[0] != 1 { + t.Fatalf("handle chunks %v, want its own", got) + } + if base := fh.baseEntry.Load(); base.GetAttributes().GetInode() != 42 { + t.Fatalf("base %v moved to the replacement", base) + } + + if code := wfs.doFlush(context.Background(), fh, 1000, 20, false); code == fuse.OK { + t.Fatal("flush into the replacement acknowledged") + } + fake.mu.Lock() + defer fake.mu.Unlock() + if !proto.Equal(fake.entry, replacement) { + t.Fatalf("replacement changed: %v", fake.entry) + } + if !fh.dirtyMetadata { + t.Fatal("refused flush lost dirty metadata") + } + }) + } +} + +// A peer's metadata-only change reaches a dirty handle's live entry, so its +// next flush does not publish the stale value back; the handle's own edits +// stay on top. +func TestPeerChmodReachesDirtyHandle(t *testing.T) { + for _, name := range []string{"peer-only", "both-changed"} { + t.Run(name, func(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), 0, 0) + fh.AddChunks([]*filer_pb.FileChunk{{FileId: "1,0300000001", Offset: 2, Size: 1, ModifiedTsNs: 3}}) + want := uint32(0600) + if name == "both-changed" { + fh.UpdateEntry(func(e *filer_pb.Entry) { e.Attributes.FileMode = 0640 }) + want = 0640 + } + fh.dirtyMetadata = true + pagesBefore := fh.dirtyPages + + peer := flushConditionEntry(1) + peer.Attributes.FileMode = 0600 + resp := &filer_pb.SubscribeMetadataResponse{ + Directory: "/dir", + TsNs: 1500, + EventNotification: &filer_pb.EventNotification{ + OldEntry: flushConditionEntry(1), + 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 chmod") + } + entry := fh.GetEntry().GetEntry() + if got := entry.Attributes.FileMode; got != want || len(entry.Chunks) != 2 { + t.Fatalf("handle mode %o with %d chunks, want %o with its own chunk", got, len(entry.Chunks), want) + } + if base := fh.baseEntry.Load(); base.GetAttributes().GetFileMode() != 0600 { + t.Fatalf("base %v, want the peer's entry", base) + } + }) + } +} + +// 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) + } +} diff --git a/weed/mount/weedfs_metadata_flush.go b/weed/mount/weedfs_metadata_flush.go index fdcafa0ef5..7c023067d7 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, true) if err != nil { return err } + if merged != nil { + fh.rebaseEntry(merged, requestEntry) + baseSnapshot = merged + } event := resp.GetMetadataEvent() if event == nil {