diff --git a/weed/mount/content_flush_condition_test.go b/weed/mount/content_flush_condition_test.go index f62d798bc7..5a4b34eda4 100644 --- a/weed/mount/content_flush_condition_test.go +++ b/weed/mount/content_flush_condition_test.go @@ -4,12 +4,14 @@ 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" @@ -418,3 +420,37 @@ func TestRebaseEntryKeepsMergedSize(t *testing.T) { 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 70d312eb44..77468adbc0 100644 --- a/weed/mount/filehandle.go +++ b/weed/mount/filehandle.go @@ -70,6 +70,10 @@ type FileHandle struct { // writes can grow the file past it; a merge still clips peer chunks there. truncatedTo atomic.Pointer[truncateMark] + // 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. @@ -145,6 +149,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") } @@ -199,7 +204,21 @@ func (fh *FileHandle) rebaseEntry(merged, sent *filer_pb.Entry) { attrs.FileSize = size } e.Attributes, e.Extended, e.Content, e.Chunks = attrs, merged.Extended, merged.Content, chunks - fh.entryChunkGroup.SetChunks(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) }) } 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_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)