From 0f05957bc4d82fd1d855f13349965fa3cc3f21d2 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Wed, 2 Sep 2026 17:43:46 -0700 Subject: [PATCH] filer: self-heal chunk manifest reads when volume locations go stale (#11107) * filer: self-heal fetchWholeChunk on stale volume locations Upstream #10156/#10800 wired cache invalidation into the buffer-based read paths, but manifest resolution still goes through fetchWholeChunk, which returns the raw error on failure. When cached volume locations are stale (volume tiered to remote storage, server rolled), resolving a large multipart file fails permanently even though other locations are healthy. Thread the ChunkGroup's cacheInvalidator through ResolveChunkManifest / ResolveOneChunkManifest / fetchWholeChunk, and on failure invalidate, re-lookup and retry once via the existing retryFetchWithFreshLocations helper. The streaming bytesBuffer is reset before the retry so partial bytes from the failed attempt cannot corrupt the manifest proto.Unmarshal. Non-mount callers pass nil and keep their semantics. * filer: move the manifest self-heal tests in with the other manifest tests Also make the stale server stream a prefix and then abort mid-body, which is what actually leaves partial bytes in the buffer: an HTTP error status returns before ReadUrlAsStream ever calls the writer, so a 500 never exercised the Reset the tests claimed to cover. Claude-Session: https://claude.ai/code/session_01FK3oGC5ZVeJYvNBWgb9JUD * filer: keep the cached volume locations when a manifest read is cancelled A cancelled or timed-out read says nothing about where the volume lives, so dropping the location and going back to the master only costs the next reader a round trip. PrepareStreamContentWithThrottler already guards its self-heal this way. The guard also goes inside retryFetchWithFreshLocations, since the caller can be cancelled between its own check and the invalidation, and that covers the reader cache and prefetch paths too. fetchWholeChunk returns the context error rather than the stream failure it provoked, and ResolveOneChunkManifest wraps with %w so errors.Is still sees it. That matters even where no invalidator is passed: volume.fsck resolves manifests with nil and tells its own abort from a corrupt manifest that way, so the cancellation check sits ahead of the nil-invalidator return. Claude-Session: https://claude.ai/code/session_01FK3oGC5ZVeJYvNBWgb9JUD * filer: self-heal manifest reads on the filer and s3 paths too Every caller that already holds the location cache backing its lookup function can hand it over: the filer's read, copy and deletion paths and the log cache have the MasterClient right there, and s3api has the FilerClient. MinusChunks takes one for the same reason, since the deletion path resolves manifests through it. Only the shell tools and the replication sinks, whose lookup functions cache privately with nothing to invalidate, keep passing nil. Claude-Session: https://claude.ai/code/session_01FK3oGC5ZVeJYvNBWgb9JUD --------- Co-authored-by: bruce-zzz --- weed/filer/filechunk_group.go | 5 +- weed/filer/filechunk_manifest.go | 33 ++-- weed/filer/filechunk_manifest_test.go | 156 ++++++++++++++++++ weed/filer/filechunks.go | 8 +- weed/filer/filer_deletion.go | 4 +- weed/filer/persisted_log_cache.go | 2 +- weed/filer/stream.go | 6 + weed/mount/peer_fetcher.go | 2 +- .../replication/sink/filersink/fetch_write.go | 2 +- weed/replication/sink/filersink/filer_sink.go | 4 +- weed/s3api/s3api_chunk_manifest.go | 2 +- weed/s3api/s3api_object_handlers.go | 2 +- weed/server/filer_grpc_server.go | 2 +- weed/server/filer_grpc_server_rename.go | 2 +- weed/server/filer_server_handlers_copy.go | 2 +- weed/server/filer_server_handlers_read.go | 2 +- .../filer_server_handlers_write_merge.go | 2 +- weed/shell/command_fs_distribute_chunks.go | 2 +- weed/shell/command_fs_merge_volumes.go | 2 +- weed/shell/command_fs_verify.go | 2 +- weed/shell/command_volume_fsck.go | 2 +- 21 files changed, 209 insertions(+), 35 deletions(-) diff --git a/weed/filer/filechunk_group.go b/weed/filer/filechunk_group.go index 87c700847..0052ba707 100644 --- a/weed/filer/filechunk_group.go +++ b/weed/filer/filechunk_group.go @@ -18,6 +18,8 @@ type ChunkGroup struct { sectionsLock sync.RWMutex readerCache *ReaderCache concurrentReaders int + // cacheInvalidator lets manifest resolution drop stale volume locations, as ReaderCache does for chunk reads + cacheInvalidator CacheInvalidator } // NewChunkGroup creates a ChunkGroup with configurable concurrency. @@ -43,6 +45,7 @@ func NewChunkGroup(lookupFn wdclient.LookupFileIdFunctionType, chunkCache chunk_ sections: make(map[SectionIndex]*FileChunkSection), readerCache: NewReaderCache(readerCacheLimit, chunkCache, lookupFn, cacheInvalidator), concurrentReaders: concurrentReaders, + cacheInvalidator: cacheInvalidator, } err := group.SetChunks(chunks) @@ -227,7 +230,7 @@ func (group *ChunkGroup) SetChunks(chunks []*filer_pb.FileChunk) error { continue } - resolvedChunks, err := ResolveOneChunkManifest(context.Background(), group.lookupFn, chunk) + resolvedChunks, err := ResolveOneChunkManifest(context.Background(), group.lookupFn, chunk, group.cacheInvalidator) if err != nil { return err } diff --git a/weed/filer/filechunk_manifest.go b/weed/filer/filechunk_manifest.go index f29cbf2b2..57cef8fa7 100644 --- a/weed/filer/filechunk_manifest.go +++ b/weed/filer/filechunk_manifest.go @@ -49,7 +49,7 @@ func SeparateManifestChunks(chunks []*filer_pb.FileChunk) (manifestChunks, nonMa return } -func ResolveChunkManifest(ctx context.Context, lookupFileIdFn wdclient.LookupFileIdFunctionType, chunks []*filer_pb.FileChunk, startOffset, stopOffset int64) (dataChunks, manifestChunks []*filer_pb.FileChunk, manifestResolveErr error) { +func ResolveChunkManifest(ctx context.Context, lookupFileIdFn wdclient.LookupFileIdFunctionType, chunks []*filer_pb.FileChunk, startOffset, stopOffset int64, invalidator CacheInvalidator) (dataChunks, manifestChunks []*filer_pb.FileChunk, manifestResolveErr error) { // TODO maybe parallel this for _, chunk := range chunks { @@ -62,14 +62,14 @@ func ResolveChunkManifest(ctx context.Context, lookupFileIdFn wdclient.LookupFil continue } - resolvedChunks, err := ResolveOneChunkManifest(ctx, lookupFileIdFn, chunk) + resolvedChunks, err := ResolveOneChunkManifest(ctx, lookupFileIdFn, chunk, invalidator) if err != nil { return dataChunks, nil, err } manifestChunks = append(manifestChunks, chunk) // recursive - subDataChunks, subManifestChunks, subErr := ResolveChunkManifest(ctx, lookupFileIdFn, resolvedChunks, startOffset, stopOffset) + subDataChunks, subManifestChunks, subErr := ResolveChunkManifest(ctx, lookupFileIdFn, resolvedChunks, startOffset, stopOffset, invalidator) if subErr != nil { return dataChunks, nil, subErr } @@ -79,7 +79,7 @@ func ResolveChunkManifest(ctx context.Context, lookupFileIdFn wdclient.LookupFil return } -func ResolveOneChunkManifest(ctx context.Context, lookupFileIdFn wdclient.LookupFileIdFunctionType, chunk *filer_pb.FileChunk) (dataChunks []*filer_pb.FileChunk, manifestResolveErr error) { +func ResolveOneChunkManifest(ctx context.Context, lookupFileIdFn wdclient.LookupFileIdFunctionType, chunk *filer_pb.FileChunk, invalidator CacheInvalidator) (dataChunks []*filer_pb.FileChunk, manifestResolveErr error) { if !chunk.IsChunkManifest { return } @@ -88,13 +88,13 @@ func ResolveOneChunkManifest(ctx context.Context, lookupFileIdFn wdclient.Lookup bytesBuffer := bytesBufferPool.Get().(*bytes.Buffer) bytesBuffer.Reset() defer bytesBufferPool.Put(bytesBuffer) - err := fetchWholeChunk(ctx, bytesBuffer, lookupFileIdFn, chunk.GetFileIdString(), chunk.CipherKey, chunk.IsCompressed) + err := fetchWholeChunk(ctx, bytesBuffer, lookupFileIdFn, chunk.GetFileIdString(), chunk.CipherKey, chunk.IsCompressed, invalidator) if err != nil { - return nil, fmt.Errorf("fail to read manifest %s: %v", chunk.GetFileIdString(), err) + return nil, fmt.Errorf("fail to read manifest %s: %w", chunk.GetFileIdString(), err) } m := &filer_pb.FileChunkManifest{} if err := proto.Unmarshal(bytesBuffer.Bytes(), m); err != nil { - return nil, fmt.Errorf("fail to unmarshal manifest %s: %v", chunk.GetFileIdString(), err) + return nil, fmt.Errorf("fail to unmarshal manifest %s: %w", chunk.GetFileIdString(), err) } // recursive @@ -103,18 +103,27 @@ func ResolveOneChunkManifest(ctx context.Context, lookupFileIdFn wdclient.Lookup } // TODO fetch from cache for weed mount? -func fetchWholeChunk(ctx context.Context, bytesBuffer *bytes.Buffer, lookupFileIdFn wdclient.LookupFileIdFunctionType, fileId string, cipherKey []byte, isGzipped bool) error { +func fetchWholeChunk(ctx context.Context, bytesBuffer *bytes.Buffer, lookupFileIdFn wdclient.LookupFileIdFunctionType, fileId string, cipherKey []byte, isGzipped bool, invalidator CacheInvalidator) error { urlStrings, err := lookupFileIdFn(ctx, fileId) if err != nil { glog.ErrorfCtx(ctx, "operation LookupFileId %s failed, err: %v", fileId, err) return err } jwt := JwtForVolumeServer(fileId) - _, err = retriedStreamFetchChunkData(ctx, bytesBuffer, urlStrings, jwt, cipherKey, isGzipped, true, 0, 0) - if err != nil { - return err + if _, err = retriedStreamFetchChunkData(ctx, bytesBuffer, urlStrings, jwt, cipherKey, isGzipped, true, 0, 0); err == nil { + return nil } - return nil + if ctxErr := ctx.Err(); ctxErr != nil { + // a cancelled read says nothing about where the volume lives, and the + // stream error it provoked is a symptom, not the cause + return ctxErr + } + return retryFetchWithFreshLocations(ctx, invalidator, lookupFileIdFn, fileId, urlStrings, err, func(newUrls []string) error { + // the failed attempt may have streamed a partial prefix into the buffer + bytesBuffer.Reset() + _, retryErr := retriedStreamFetchChunkData(ctx, bytesBuffer, newUrls, jwt, cipherKey, isGzipped, true, 0, 0) + return retryErr + }) } func fetchChunkRange(ctx context.Context, buffer []byte, lookupFileIdFn wdclient.LookupFileIdFunctionType, fileId string, cipherKey []byte, isGzipped bool, offset int64, refreshUrls util_http.RefreshUrlsFunc) (int, error) { diff --git a/weed/filer/filechunk_manifest_test.go b/weed/filer/filechunk_manifest_test.go index 4411b01fb..e9243be69 100644 --- a/weed/filer/filechunk_manifest_test.go +++ b/weed/filer/filechunk_manifest_test.go @@ -6,6 +6,10 @@ import ( "fmt" "io" "math" + "net/http" + "net/http/httptest" + "strconv" + "sync/atomic" "testing" "github.com/stretchr/testify/assert" @@ -504,3 +508,155 @@ func TestManifestBloatDetection(t *testing.T) { } } } + +// countingInvalidator stands in for the vidMap cache a mount or filer hands to +// manifest resolution, recording how often the locations were dropped. +type countingInvalidator struct { + invalidations atomic.Int32 +} + +func (inv *countingInvalidator) InvalidateCache(fileId string) { + inv.invalidations.Add(1) +} + +// stagedLookup returns staleUrls once and freshUrls afterwards, the way a +// vidMapClient re-reads the master once InvalidateCache has dropped the entry. +type stagedLookup struct { + staleUrls []string + freshUrls []string + calls atomic.Int32 +} + +func (l *stagedLookup) lookup(ctx context.Context, fileId string) ([]string, error) { + if l.calls.Add(1) == 1 { + return l.staleUrls, nil + } + return l.freshUrls, nil +} + +func manifestServer(t *testing.T, manifestBytes []byte) *httptest.Server { + t.Helper() + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Length", strconv.Itoa(len(manifestBytes))) + w.Write(manifestBytes) + })) + t.Cleanup(srv.Close) + return srv +} + +func fetchManifestBuffer(t *testing.T) *bytes.Buffer { + t.Helper() + bytesBuffer := bytesBufferPool.Get().(*bytes.Buffer) + bytesBuffer.Reset() + t.Cleanup(func() { bytesBufferPool.Put(bytesBuffer) }) + return bytesBuffer +} + +// TestFetchWholeChunkRetriesFreshLocations covers the mount reading a multipart +// file whose manifest volume has moved: the cached location is dead, so the +// fetch has to drop it and come back with what the master knows now. The stale +// server streams a prefix before dying, which the retry must not keep. +func TestFetchWholeChunkRetriesFreshLocations(t *testing.T) { + manifestBytes, err := proto.Marshal(&filer_pb.FileChunkManifest{ + Chunks: []*filer_pb.FileChunk{ + {FileId: "100,abc", Offset: 0, Size: 8}, + {FileId: "101,def", Offset: 8, Size: 8}, + }, + }) + assert.NoError(t, err) + + staleSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Length", strconv.Itoa(len(manifestBytes)+64)) + w.Write([]byte("garbage-from-stale-server")) + w.(http.Flusher).Flush() + panic(http.ErrAbortHandler) + })) + defer staleSrv.Close() + + lookup := &stagedLookup{ + staleUrls: []string{staleSrv.URL + "/5,stale"}, + freshUrls: []string{manifestServer(t, manifestBytes).URL + "/5,stale"}, + } + inv := &countingInvalidator{} + bytesBuffer := fetchManifestBuffer(t) + + assert.NoError(t, fetchWholeChunk(context.Background(), bytesBuffer, lookup.lookup, "5,stale", nil, false, inv)) + assert.Equal(t, int32(1), inv.invalidations.Load()) + assert.Equal(t, int32(2), lookup.calls.Load()) + + // the buffer must hold the fresh manifest alone, not the stale prefix ahead of it + decoded := &filer_pb.FileChunkManifest{} + assert.NoError(t, proto.Unmarshal(bytesBuffer.Bytes(), decoded)) + assertEqualChunks(t, []*filer_pb.FileChunk{ + {FileId: "100,abc", Offset: 0, Size: 8}, + {FileId: "101,def", Offset: 8, Size: 8}, + }, decoded.Chunks) +} + +// TestFetchWholeChunkWithoutInvalidator keeps the old behavior for callers whose +// lookup function has no cache to drop: the fetch error surfaces as it is. +func TestFetchWholeChunkWithoutInvalidator(t *testing.T) { + failSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusInternalServerError) + })) + defer failSrv.Close() + + lookup := &stagedLookup{ + staleUrls: []string{failSrv.URL + "/5,abc"}, + freshUrls: []string{"http://unused:8080/5,abc"}, + } + + assert.Error(t, fetchWholeChunk(context.Background(), fetchManifestBuffer(t), lookup.lookup, "5,abc", nil, false, nil)) + assert.Equal(t, int32(1), lookup.calls.Load()) +} + +// TestFetchWholeChunkUnchangedLocations guards against looping: when the master +// still reports the servers that just failed, there is nothing new to try. +func TestFetchWholeChunkUnchangedLocations(t *testing.T) { + failSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusInternalServerError) + })) + defer failSrv.Close() + + lookup := &stagedLookup{ + staleUrls: []string{failSrv.URL + "/5,abc"}, + freshUrls: []string{failSrv.URL + "/5,abc"}, + } + inv := &countingInvalidator{} + + assert.Error(t, fetchWholeChunk(context.Background(), fetchManifestBuffer(t), lookup.lookup, "5,abc", nil, false, inv)) + assert.Equal(t, int32(2), lookup.calls.Load()) + assert.Equal(t, int32(1), inv.invalidations.Load()) +} + +// TestFetchWholeChunkCancelledKeepsLocations checks that a caller walking away +// mid-read does not cost every other reader a master round trip. +func TestFetchWholeChunkCancelledKeepsLocations(t *testing.T) { + lookup := &stagedLookup{ + staleUrls: []string{"http://unused:8080/5,abc"}, + freshUrls: []string{"http://elsewhere:8080/5,abc"}, + } + inv := &countingInvalidator{} + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + err := fetchWholeChunk(ctx, fetchManifestBuffer(t), lookup.lookup, "5,abc", nil, false, inv) + assert.ErrorIs(t, err, context.Canceled) + assert.Equal(t, int32(0), inv.invalidations.Load()) + assert.Equal(t, int32(1), lookup.calls.Load()) + + // the cancellation must survive the wrapping ResolveOneChunkManifest does, + // or callers cannot tell a dead volume server from their own abort + manifestChunk := &filer_pb.FileChunk{FileId: "5,abc", IsChunkManifest: true} + _, resolveErr := ResolveOneChunkManifest(ctx, lookup.lookup, manifestChunk, inv) + assert.ErrorIs(t, resolveErr, context.Canceled) + + // volume.fsck resolves manifests with no invalidator and still has to tell + // its own abort from a corrupt manifest, so the nil path keeps the identity + noInvalidator := retryFetchWithFreshLocations(ctx, nil, lookup.lookup, "5,abc", nil, fmt.Errorf("stale server said no"), func([]string) error { + t.Fatal("refetch must not run on a cancelled read") + return nil + }) + assert.ErrorIs(t, noInvalidator, context.Canceled) +} diff --git a/weed/filer/filechunks.go b/weed/filer/filechunks.go index 6e9ec3f21..8cd8bb843 100644 --- a/weed/filer/filechunks.go +++ b/weed/filer/filechunks.go @@ -100,13 +100,13 @@ func FindGarbageChunks(visibles *IntervalList[*VisibleInterval], start int64, st return } -func MinusChunks(ctx context.Context, lookupFileIdFn wdclient.LookupFileIdFunctionType, as, bs []*filer_pb.FileChunk) (delta []*filer_pb.FileChunk, err error) { +func MinusChunks(ctx context.Context, lookupFileIdFn wdclient.LookupFileIdFunctionType, as, bs []*filer_pb.FileChunk, invalidator CacheInvalidator) (delta []*filer_pb.FileChunk, err error) { - aData, aMeta, aErr := ResolveChunkManifest(ctx, lookupFileIdFn, as, 0, math.MaxInt64) + aData, aMeta, aErr := ResolveChunkManifest(ctx, lookupFileIdFn, as, 0, math.MaxInt64, invalidator) if aErr != nil { return nil, aErr } - bData, bMeta, bErr := ResolveChunkManifest(ctx, lookupFileIdFn, bs, 0, math.MaxInt64) + bData, bMeta, bErr := ResolveChunkManifest(ctx, lookupFileIdFn, bs, 0, math.MaxInt64, invalidator) if bErr != nil { return nil, bErr } @@ -271,7 +271,7 @@ func MergeIntoChunkViews(chunkViews *IntervalList[*ChunkView], start int64, stop // If the file chunk content is a chunk manifest func NonOverlappingVisibleIntervals(ctx context.Context, lookupFileIdFn wdclient.LookupFileIdFunctionType, chunks []*filer_pb.FileChunk, startOffset int64, stopOffset int64) (visibles *IntervalList[*VisibleInterval], err error) { - chunks, _, err = ResolveChunkManifest(ctx, lookupFileIdFn, chunks, startOffset, stopOffset) + chunks, _, err = ResolveChunkManifest(ctx, lookupFileIdFn, chunks, startOffset, stopOffset, nil) if err != nil { return } diff --git a/weed/filer/filer_deletion.go b/weed/filer/filer_deletion.go index 63a20251d..f32b97a8b 100644 --- a/weed/filer/filer_deletion.go +++ b/weed/filer/filer_deletion.go @@ -602,7 +602,7 @@ func (f *Filer) doDeleteChunks(ctx context.Context, chunks []*filer_pb.FileChunk f.FileIdDeletionQueue.EnQueue(chunk.GetFileIdString()) continue } - dataChunks, manifestResolveErr := ResolveOneChunkManifest(ctx, f.MasterClient.LookupFileId, chunk) + dataChunks, manifestResolveErr := ResolveOneChunkManifest(ctx, f.MasterClient.LookupFileId, chunk, f.MasterClient) if manifestResolveErr != nil { glog.V(0).InfofCtx(ctx, "failed to resolve manifest %s: %v", chunk.FileId, manifestResolveErr) } @@ -628,7 +628,7 @@ func (f *Filer) deleteChunksIfNotNew(ctx context.Context, oldEntry, newEntry *En newChunks = newEntry.GetChunks() } - toDelete, err := MinusChunks(ctx, f.MasterClient.GetLookupFileIdFunction(), oldChunks, newChunks) + toDelete, err := MinusChunks(ctx, f.MasterClient.GetLookupFileIdFunction(), oldChunks, newChunks, f.MasterClient) if err != nil { glog.ErrorfCtx(ctx, "Failed to resolve old entry chunks when delete old entry chunks. new: %s, old: %s", newChunks, oldChunks) return diff --git a/weed/filer/persisted_log_cache.go b/weed/filer/persisted_log_cache.go index 6f33a8c44..430582c21 100644 --- a/weed/filer/persisted_log_cache.go +++ b/weed/filer/persisted_log_cache.go @@ -190,7 +190,7 @@ func loadLogFileEntries(masterClient *wdclient.MasterClient, chunk *filer_pb.Fil lookupFileIdFn := func(ctx context.Context, fileId string) (targetUrls []string, err error) { return masterClient.LookupFileId(ctx, fileId) } - if fetchErr := fetchWholeChunk(context.Background(), bytesBuffer, lookupFileIdFn, chunk.GetFileIdString(), chunk.CipherKey, chunk.IsCompressed); fetchErr != nil { + if fetchErr := fetchWholeChunk(context.Background(), bytesBuffer, lookupFileIdFn, chunk.GetFileIdString(), chunk.CipherKey, chunk.IsCompressed, masterClient); fetchErr != nil { return nil, false, fetchErr } return decodeLogRecords(bytesBuffer.Bytes()) diff --git a/weed/filer/stream.go b/weed/filer/stream.go index b06e4a55a..5b4d4cbd5 100644 --- a/weed/filer/stream.go +++ b/weed/filer/stream.go @@ -107,6 +107,12 @@ type VolumeServerJwtFunction func(fileId string) string // resolved locations actually changed (so we never retry against the same servers). originalErr // is returned unchanged when no retry is attempted, so callers surface the real fetch failure. func retryFetchWithFreshLocations(ctx context.Context, invalidator CacheInvalidator, lookupFn wdclient.LookupFileIdFunctionType, fileId string, oldUrls []string, originalErr error, refetch func(newUrls []string) error) error { + // the caller may have gone away between its own check and this one; a + // cancelled read is no evidence the locations are wrong, and callers such + // as volume.fsck tell an abort from real corruption with errors.Is + if ctxErr := ctx.Err(); ctxErr != nil { + return ctxErr + } if invalidator == nil { return originalErr } diff --git a/weed/mount/peer_fetcher.go b/weed/mount/peer_fetcher.go index 87d816c1f..ad95e42fa 100644 --- a/weed/mount/peer_fetcher.go +++ b/weed/mount/peer_fetcher.go @@ -65,7 +65,7 @@ func (fh *FileHandle) tryPeerRead(ctx context.Context, fileSize int64, buff []by if readStop > fileSize { readStop = fileSize } - dataChunks, _, err := filer.ResolveChunkManifest(ctx, fh.wfs.LookupFn(), chunks, offset, readStop) + dataChunks, _, err := filer.ResolveChunkManifest(ctx, fh.wfs.LookupFn(), chunks, offset, readStop, fh.wfs.CacheInvalidator()) if err != nil { return 0, 0, fmt.Errorf("resolve manifest: %w", err) } diff --git a/weed/replication/sink/filersink/fetch_write.go b/weed/replication/sink/filersink/fetch_write.go index b05a926b6..08a79bffc 100644 --- a/weed/replication/sink/filersink/fetch_write.go +++ b/weed/replication/sink/filersink/fetch_write.go @@ -195,7 +195,7 @@ func (fs *FilerSink) replicateOneManifestChunk(ctx context.Context, sourceChunk resolveName := fmt.Sprintf("resolve manifest %s", sourceChunk.GetFileIdString()) missingGate := fs.newMissingSourceChunkGate(sourceChunk.GetFileIdString()) err := util.RetryUntil(resolveName, func() error { - rc, e := filer.ResolveOneChunkManifest(ctx, fs.filerSource.LookupFileId, sourceChunk) + rc, e := filer.ResolveOneChunkManifest(ctx, fs.filerSource.LookupFileId, sourceChunk, nil) if e != nil { return e } diff --git a/weed/replication/sink/filersink/filer_sink.go b/weed/replication/sink/filersink/filer_sink.go index 4552312be..0f7a42e10 100644 --- a/weed/replication/sink/filersink/filer_sink.go +++ b/weed/replication/sink/filersink/filer_sink.go @@ -405,11 +405,11 @@ func (fs *FilerSink) UpdateEntry(key string, oldEntry *filer_pb.Entry, newParent } func compareChunks(ctx context.Context, lookupFileIdFn wdclient.LookupFileIdFunctionType, oldEntry, newEntry *filer_pb.Entry) (deletedChunks, newChunks []*filer_pb.FileChunk, err error) { - aData, aMeta, aErr := filer.ResolveChunkManifest(ctx, lookupFileIdFn, oldEntry.GetChunks(), 0, math.MaxInt64) + aData, aMeta, aErr := filer.ResolveChunkManifest(ctx, lookupFileIdFn, oldEntry.GetChunks(), 0, math.MaxInt64, nil) if aErr != nil { return nil, nil, aErr } - bData, bMeta, bErr := filer.ResolveChunkManifest(ctx, lookupFileIdFn, newEntry.GetChunks(), 0, math.MaxInt64) + bData, bMeta, bErr := filer.ResolveChunkManifest(ctx, lookupFileIdFn, newEntry.GetChunks(), 0, math.MaxInt64, nil) if bErr != nil { return nil, nil, bErr } diff --git a/weed/s3api/s3api_chunk_manifest.go b/weed/s3api/s3api_chunk_manifest.go index c25a36ca2..d8eaaa9e8 100644 --- a/weed/s3api/s3api_chunk_manifest.go +++ b/weed/s3api/s3api_chunk_manifest.go @@ -65,7 +65,7 @@ func (s3a *S3ApiServer) flattenManifestChunks(ctx context.Context, entry *filer_ if entry == nil || !filer.HasChunkManifest(entry.GetChunks()) { return nil, nil } - dataChunks, manifestChunks, err := filer.ResolveChunkManifest(ctx, s3a.createLookupFileIdFunction(), entry.GetChunks(), 0, math.MaxInt64) + dataChunks, manifestChunks, err := filer.ResolveChunkManifest(ctx, s3a.createLookupFileIdFunction(), entry.GetChunks(), 0, math.MaxInt64, s3a.filerClient) if err != nil { return nil, err } diff --git a/weed/s3api/s3api_object_handlers.go b/weed/s3api/s3api_object_handlers.go index 99b80d652..902d983a8 100644 --- a/weed/s3api/s3api_object_handlers.go +++ b/weed/s3api/s3api_object_handlers.go @@ -2140,7 +2140,7 @@ func (s3a *S3ApiServer) getEncryptedStreamFromVolumes(ctx context.Context, entry lookupFileIdFn := s3a.createLookupFileIdFunction() // Resolve chunks - resolvedChunks, _, err := filer.ResolveChunkManifest(ctx, lookupFileIdFn, chunks, offset, offset+size) + resolvedChunks, _, err := filer.ResolveChunkManifest(ctx, lookupFileIdFn, chunks, offset, offset+size, s3a.filerClient) if err != nil { return nil, err } diff --git a/weed/server/filer_grpc_server.go b/weed/server/filer_grpc_server.go index 0168a9279..c2d21922c 100644 --- a/weed/server/filer_grpc_server.go +++ b/weed/server/filer_grpc_server.go @@ -715,7 +715,7 @@ func (fs *FilerServer) cleanupChunks(ctx context.Context, fullpath string, exist // remove old chunks if not included in the new ones if existingEntry != nil { - garbage, err = filer.MinusChunks(ctx, fs.lookupFileId, existingEntry.GetChunks(), newEntry.GetChunks()) + garbage, err = filer.MinusChunks(ctx, fs.lookupFileId, existingEntry.GetChunks(), newEntry.GetChunks(), fs.filer.MasterClient) if err != nil { return newEntry.GetChunks(), nil, fmt.Errorf("MinusChunks: %w", err) } diff --git a/weed/server/filer_grpc_server_rename.go b/weed/server/filer_grpc_server_rename.go index d44bad283..020a59c71 100644 --- a/weed/server/filer_grpc_server_rename.go +++ b/weed/server/filer_grpc_server_rename.go @@ -274,7 +274,7 @@ func (fs *FilerServer) moveSelfEntry(ctx context.Context, stream filer_pb.Seawee } } if existingTarget != nil { - toDelete, err := filer.MinusChunks(ctx, fs.filer.MasterClient.GetLookupFileIdFunction(), existingTarget.GetChunks(), newEntry.GetChunks()) + toDelete, err := filer.MinusChunks(ctx, fs.filer.MasterClient.GetLookupFileIdFunction(), existingTarget.GetChunks(), newEntry.GetChunks(), fs.filer.MasterClient) if err != nil { glog.ErrorfCtx(ctx, "Failed to resolve overwrite target chunks during rename. new: %v, old: %v", newEntry.GetChunks(), existingTarget.GetChunks()) } else if len(toDelete) > 0 { diff --git a/weed/server/filer_server_handlers_copy.go b/weed/server/filer_server_handlers_copy.go index 2a6364b76..9db334b1d 100644 --- a/weed/server/filer_server_handlers_copy.go +++ b/weed/server/filer_server_handlers_copy.go @@ -600,7 +600,7 @@ func (fs *FilerServer) copyChunksWithManifest(ctx context.Context, srcChunks []* return fs.filer.MasterClient.GetLookupFileIdFunction()(ctx, fileId) } - resolvedChunks, err := filer.ResolveOneChunkManifest(ctx, lookupFileIdFn, manifestChunk) + resolvedChunks, err := filer.ResolveOneChunkManifest(ctx, lookupFileIdFn, manifestChunk, fs.filer.MasterClient) if err != nil { return nil, fmt.Errorf("failed to resolve manifest chunk %s: %w", manifestChunk.GetFileIdString(), err) } diff --git a/weed/server/filer_server_handlers_read.go b/weed/server/filer_server_handlers_read.go index 2992f3132..8de704aaa 100644 --- a/weed/server/filer_server_handlers_read.go +++ b/weed/server/filer_server_handlers_read.go @@ -132,7 +132,7 @@ func (fs *FilerServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request) if entry.Chunks, _, err = filer.ResolveChunkManifest( ctx, fs.filer.MasterClient.GetLookupFileIdFunction(), - entry.GetChunks(), 0, math.MaxInt64); err != nil { + entry.GetChunks(), 0, math.MaxInt64, fs.filer.MasterClient); err != nil { err = fmt.Errorf("failed to resolve chunk manifest, err: %s", err.Error()) writeJsonError(w, r, http.StatusInternalServerError, err) return diff --git a/weed/server/filer_server_handlers_write_merge.go b/weed/server/filer_server_handlers_write_merge.go index 24e642bd6..157d96c79 100644 --- a/weed/server/filer_server_handlers_write_merge.go +++ b/weed/server/filer_server_handlers_write_merge.go @@ -64,7 +64,7 @@ func (fs *FilerServer) mergeChunks(ctx context.Context, so *operation.StorageOpt } } - garbage, err := filer.MinusChunks(ctx, fs.lookupFileId, inputChunks, mergedChunks) + garbage, err := filer.MinusChunks(ctx, fs.lookupFileId, inputChunks, mergedChunks, fs.filer.MasterClient) if err != nil { glog.ErrorfCtx(ctx, "Failed to resolve old entry chunks when delete old entry chunks. new: %s, old: %s", mergedChunks, inputChunks) diff --git a/weed/shell/command_fs_distribute_chunks.go b/weed/shell/command_fs_distribute_chunks.go index b66df41dc..96baaa740 100644 --- a/weed/shell/command_fs_distribute_chunks.go +++ b/weed/shell/command_fs_distribute_chunks.go @@ -255,7 +255,7 @@ func resolveEntryDataChunks(commandEnv *CommandEnv, entry *filer_pb.Entry) (data if !filer.HasChunkManifest(chunks) { return chunks, false, nil } - dataChunks, _, err = filer.ResolveChunkManifest(context.Background(), filer.LookupFn(commandEnv), chunks, 0, math.MaxInt64) + dataChunks, _, err = filer.ResolveChunkManifest(context.Background(), filer.LookupFn(commandEnv), chunks, 0, math.MaxInt64, nil) if err != nil { return nil, true, fmt.Errorf("resolve chunk manifest: %v", err) } diff --git a/weed/shell/command_fs_merge_volumes.go b/weed/shell/command_fs_merge_volumes.go index 393cfd84b..488006771 100644 --- a/weed/shell/command_fs_merge_volumes.go +++ b/weed/shell/command_fs_merge_volumes.go @@ -676,7 +676,7 @@ func (c *commandFsMergeVolumes) rewriteManifestChunk( return chunk, false, rewrittenNeedles{}, fmt.Errorf("not a manifest chunk: %s", chunk.GetFileIdString()) } - subChunks, err := filer.ResolveOneChunkManifest(ctx, lookupFn, chunk) + subChunks, err := filer.ResolveOneChunkManifest(ctx, lookupFn, chunk, nil) if err != nil { return chunk, false, rewrittenNeedles{}, err } diff --git a/weed/shell/command_fs_verify.go b/weed/shell/command_fs_verify.go index a96fc09d1..92aed4169 100644 --- a/weed/shell/command_fs_verify.go +++ b/weed/shell/command_fs_verify.go @@ -287,7 +287,7 @@ func (c *commandFsVerify) verifyTraverseBfs(path string) (fileCount uint64, errC return nil } } - dataChunks, manifestChunks, resolveErr := filer.ResolveChunkManifest(context.Background(), filer.LookupFn(c.env), entry.Entry.GetChunks(), 0, math.MaxInt64) + dataChunks, manifestChunks, resolveErr := filer.ResolveChunkManifest(context.Background(), filer.LookupFn(c.env), entry.Entry.GetChunks(), 0, math.MaxInt64, nil) if resolveErr != nil { return fmt.Errorf("failed to ResolveChunkManifest: %+v", resolveErr) } diff --git a/weed/shell/command_volume_fsck.go b/weed/shell/command_volume_fsck.go index 76d344f82..22d96b7b3 100644 --- a/weed/shell/command_volume_fsck.go +++ b/weed/shell/command_volume_fsck.go @@ -310,7 +310,7 @@ func (c *commandVolumeFsck) collectFilerFileIdAndPaths(dataNodeVolumeIdToVInfo m if *c.verbose && entry.Entry.IsDirectory { fmt.Fprintf(c.writer, "checking directory %s\n", util.NewFullPath(entry.Dir, entry.Entry.Name)) } - dataChunks, manifestChunks, resolveErr := filer.ResolveChunkManifest(ctx, filer.LookupFn(c.env), entry.Entry.GetChunks(), 0, math.MaxInt64) + dataChunks, manifestChunks, resolveErr := filer.ResolveChunkManifest(ctx, filer.LookupFn(c.env), entry.Entry.GetChunks(), 0, math.MaxInt64, nil) if resolveErr != nil { // Cancellation/deadline isn't manifest corruption; surface it // so the BFS bails out cleanly without polluting the