[Filer] Parallelize Chunk Manifest Resolution to Reduce Large File Read Latency (#11215)

* fix issue-11214

* fix(filer): cancel sibling manifest reads on failure

* fix(filer): scope manifest cancellation to read batch and propagate context to encrypted reads

Address PR review comments on #11215:

- Scope cancellation to each parallel read batch instead of the resolver-wide
  context, so a later manifest failure does not cancel recursive work for an
  earlier successful manifest (CodeRabbit #3952446554).
- Propagate the resolver context through GetAuthenticatedWithContext so
  encrypted sibling reads observe cancellation and stop promptly when another
  manifest fails (Greptile #3952422343).
- Use net.ListenConfig.Listen with an explicit context in the test fixture to
  satisfy the noctx linter (CodeRabbit #3952111696).
- Add regression tests for encrypted sibling cancellation and for preserving
  earlier manifest children on later failure.

* fix(filer): return known manifest errors without blocking on earlier children

Address Greptile review comment on #11215:

After all parallel reads complete, pre-scan slots for the first real
(non-internal-cancel) error before recursing into earlier manifests'
children. If a later manifest already failed, return its error promptly
with data chunks already in hand, instead of blocking on recursive
network reads of earlier manifests' children.

Updated the regression test to verify the error returns within 1 second
when an earlier manifest's child has a 2-second delay, and that the
child is never loaded.

* fix(filer): filter partial child chunks by requested range on error path

Address CodeRabbit review comment on #11215:

The pre-scan error path appended non-manifest child chunks from earlier
manifests without applying the [startOffset, stopOffset) overlap check
used for top-level chunks. A child outside the requested range could be
returned in dataChunks alongside the later manifest's error.

Apply the same range predicate before appending. Add regression test
with an out-of-range child chunk.

* fix(filer): buffer job channel and abort submission on batch cancellation

Address Greptile review comment on #11215:

- Use a buffered job channel (capacity 128) so submission does not block
  when all workers are busy. This ensures a promptly-failing manifest is
  always queued and can cancel stalled sibling reads once a worker picks
  it up, instead of blocking the caller on the unbuffered channel send.
- Add batchCtx.Done() to the submit select so submission aborts promptly
  when the batch is already cancelled by a sibling failure.
- Add regression test with 5 manifests (4 stalled + 1 failing) verifying
  the failing job is queued and picked up after a stalled worker is freed.

* fix(filer): avoid double WaitGroup decrement on batch cancellation in submit

Address Devin review comment on #11215:

When batchCtx.Done() fired in submit, it called job.done.Done() and
returned false. The caller in resolve also called reads.Done() on the
same WaitGroup, causing a double decrement that would panic with a
negative counter.

Fix: submit sets the result error but does not decrement the WaitGroup.
The caller always owns the decrement and skips overwriting the result
when submit already set it.

* fix(filer): overflow execution when job queue buffer is full

Address Greptile follow-up review comment on #11215:

With a 128-entry buffer, if more than 132 in-range manifests (4 workers +
128 buffer) stall at one level, a promptly-failing manifest beyond the
buffer cannot be submitted and cannot cancel the stalled reads.

Fix: when the buffer is full, run the job directly in a goroutine instead
of blocking on the channel send. This only triggers for >132 manifests at
one level (exceedingly rare), so the bounded concurrency guarantee (4
workers) holds for all normal workloads. Extracted executeJob method
shared by both workers and overflow goroutines.

* fix(filer): bound overflow execution with a semaphore

Address Devin review comment on #11215:

The unbounded overflow goroutines could create thousands of concurrent
reads for large files, defeating the four-worker resource bound.

Fix: add a semaphore (capacity = maxChunkManifestResolveWorkers) that
overflow goroutines must acquire before doing the read. While waiting for
the semaphore, they also watch batchCtx and r.ctx so they exit promptly
on cancellation. Total concurrency is now bounded to 2 * workers (4
workers + 4 overflow) in the degenerate case.

* refactor(http): add ctx to GetAuthenticated signature instead of new function

Reuse the existing GetAuthenticated name by adding ctx as the first
parameter, matching the pattern of ReadUrl, ReadUrlAsStream, and
RetriedFetchChunkData. Removes the GetAuthenticatedWithContext wrapper.

---------

Co-authored-by: Chris Lu <chris.lu@gmail.com>

weed/filer/filechunk_manifest.go

@@ -3,6 +3,7 @@ package filer
 import (
 	"bytes"
 	"context"
+	"errors"
 	"fmt"
 	"io"
 	"math"
@@ -25,6 +26,14 @@ var bytesBufferPool = sync.Pool{
 	},
 }
 
+// Keep Manifest reads bounded across all recursion levels of one resolution.
+const maxChunkManifestResolveWorkers = 4
+
+// Size of the job queue buffer. Large enough that submission does not block
+// under normal chunk counts, so a promptly-failing manifest is always queued
+// and can cancel stalled sibling reads once a worker picks it up.
+const chunkManifestResolveJobBufferSize = 128
+
 func HasChunkManifest(chunks []*filer_pb.FileChunk) bool {
 	for _, chunk := range chunks {
 		if chunk.IsChunkManifest {
@@ -46,26 +55,229 @@ func SeparateManifestChunks(chunks []*filer_pb.FileChunk) (manifestChunks, nonMa
 }
 
 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 {
+	resolver := newChunkManifestResolver(ctx, lookupFileIdFn, invalidator)
+	defer resolver.close()
+	return resolver.resolve(chunks, startOffset, stopOffset)
+}
 
+type chunkManifestResolveJob struct {
+	chunk       *filer_pb.FileChunk
+	result      *chunkManifestResolveResult
+	done        *sync.WaitGroup
+	batchCtx    context.Context
+	batchCancel context.CancelFunc
+	batchOnce   *sync.Once
+}
+
+type chunkManifestResolveResult struct {
+	chunks         []*filer_pb.FileChunk
+	err            error
+	internalCancel bool
+}
+
+type chunkManifestResolver struct {
+	ctx            context.Context
+	parentCtx      context.Context
+	cancel         context.CancelFunc
+	lookupFileIdFn wdclient.LookupFileIdFunctionType
+	invalidator    CacheInvalidator
+	jobs           chan chunkManifestResolveJob
+	overflowSem    chan struct{}
+	workers        sync.WaitGroup
+	startOnce      sync.Once
+	started        bool
+}
+
+func newChunkManifestResolver(ctx context.Context, lookupFileIdFn wdclient.LookupFileIdFunctionType, invalidator CacheInvalidator) *chunkManifestResolver {
+	workCtx, cancel := context.WithCancel(ctx)
+	resolver := &chunkManifestResolver{
+		ctx:            workCtx,
+		parentCtx:      ctx,
+		cancel:         cancel,
+		lookupFileIdFn: lookupFileIdFn,
+		invalidator:    invalidator,
+		jobs:           make(chan chunkManifestResolveJob, chunkManifestResolveJobBufferSize),
+		overflowSem:    make(chan struct{}, maxChunkManifestResolveWorkers),
+	}
+	return resolver
+}
+
+func (r *chunkManifestResolver) executeJob(job chunkManifestResolveJob) {
+	job.result.chunks, job.result.err = ResolveOneChunkManifest(job.batchCtx, r.lookupFileIdFn, job.chunk, r.invalidator)
+	if job.result.err != nil && r.parentCtx.Err() == nil {
+		if job.batchCtx.Err() != nil && errors.Is(job.result.err, context.Canceled) {
+			job.result.internalCancel = true
+		} else if job.batchCtx.Err() == nil {
+			job.batchOnce.Do(job.batchCancel)
+		}
+	}
+	job.done.Done()
+}
+
+func (r *chunkManifestResolver) executeOverflowJob(job chunkManifestResolveJob) {
+	select {
+	case r.overflowSem <- struct{}{}:
+		defer func() { <-r.overflowSem }()
+		r.executeJob(job)
+	case <-job.batchCtx.Done():
+		if job.result.err == nil {
+			job.result.err = job.batchCtx.Err()
+			job.result.internalCancel = r.parentCtx.Err() == nil && job.result.err != nil
+		}
+		job.done.Done()
+	case <-r.ctx.Done():
+		if job.result.err == nil {
+			job.result.err = r.ctx.Err()
+			job.result.internalCancel = r.parentCtx.Err() == nil && job.result.err != nil
+		}
+		job.done.Done()
+	}
+}
+
+func (r *chunkManifestResolver) worker() {
+	defer r.workers.Done()
+	for job := range r.jobs {
+		r.executeJob(job)
+	}
+}
+
+func (r *chunkManifestResolver) close() {
+	r.cancel()
+	close(r.jobs)
+	if r.started {
+		r.workers.Wait()
+	}
+}
+
+func (r *chunkManifestResolver) submit(job chunkManifestResolveJob) bool {
+	r.startOnce.Do(func() {
+		r.workers.Add(maxChunkManifestResolveWorkers)
+		for i := 0; i < maxChunkManifestResolveWorkers; i++ {
+			go r.worker()
+		}
+		r.started = true
+	})
+	select {
+	case r.jobs <- job:
+		return true
+	case <-r.ctx.Done():
+		return false
+	case <-job.batchCtx.Done():
+		// Batch was cancelled by a sibling failure; don't queue this job.
+		// Set the result so the caller doesn't overwrite it; the caller
+		// owns the WaitGroup decrement.
+		job.result.err = job.batchCtx.Err()
+		job.result.internalCancel = r.parentCtx.Err() == nil && job.result.err != nil
+		return false
+	default:
+		// Buffer is full (exceedingly rare: more than chunkManifestResolveJobBufferSize
+		// in-range manifests at one level). Run the job in a bounded overflow goroutine
+		// so a promptly-failing manifest can still cancel the batch without waiting for
+		// a worker slot, while keeping total concurrency bounded.
+		go r.executeOverflowJob(job)
+		return true
+	}
+}
+
+func (r *chunkManifestResolver) resolve(chunks []*filer_pb.FileChunk, startOffset, stopOffset int64) (dataChunks, manifestChunks []*filer_pb.FileChunk, manifestResolveErr error) {
+	type resolveSlot struct {
+		chunk  *filer_pb.FileChunk
+		result chunkManifestResolveResult
+	}
+
+	// Cancellation is scoped to this parallel read batch so a failure in a
+	// later manifest does not cancel recursive work for an earlier manifest
+	// that already completed. Recursion creates its own batch context derived
+	// from the still-alive resolver context.
+	batchCtx, batchCancel := context.WithCancel(r.ctx)
+	defer batchCancel()
+	var batchOnce sync.Once
+
+	slots := make([]resolveSlot, len(chunks))
+	var reads sync.WaitGroup
+	for i, chunk := range chunks {
 		if max(chunk.Offset, startOffset) >= min(chunk.Offset+int64(chunk.Size), stopOffset) {
 			continue
 		}
 
+		slots[i].chunk = chunk
 		if !chunk.IsChunkManifest {
-			dataChunks = append(dataChunks, chunk)
 			continue
 		}
 
-		resolvedChunks, err := ResolveOneChunkManifest(ctx, lookupFileIdFn, chunk, invalidator)
-		if err != nil {
+		reads.Add(1)
+		if !r.submit(chunkManifestResolveJob{
+			chunk:       chunk,
+			result:      &slots[i].result,
+			done:        &reads,
+			batchCtx:    batchCtx,
+			batchCancel: batchCancel,
+			batchOnce:   &batchOnce,
+		}) {
+			// submit may have already set the result (batch cancellation).
+			// Only set it here for the resolver-cancellation case.
+			if slots[i].result.err == nil {
+				slots[i].result.err = r.ctx.Err()
+				slots[i].result.internalCancel = r.parentCtx.Err() == nil && slots[i].result.err != nil
+			}
+			reads.Done()
+		}
+	}
+	reads.Wait()
+	if err := r.parentCtx.Err(); err != nil {
 		return dataChunks, nil, err
 	}
 
-		manifestChunks = append(manifestChunks, chunk)
-		// recursive
-		subDataChunks, subManifestChunks, subErr := ResolveChunkManifest(ctx, lookupFileIdFn, resolvedChunks, startOffset, stopOffset, invalidator)
+	// Recurse only after this level's reads release their worker slots. A
+	// worker must never wait for a child manifest while holding a slot.
+	//
+	// Pre-scan for the first real (non-internal-cancel) error before
+	// recursing. If a later manifest already failed, return its error
+	// promptly with data chunks that are already in hand, instead of
+	// blocking on recursive reads of earlier manifests' children.
+	for i, slot := range slots {
+		if slot.chunk == nil {
+			continue
+		}
+		if slot.result.err != nil {
+			if slot.result.internalCancel {
+				continue
+			}
+			for j := 0; j < i; j++ {
+				if slots[j].chunk == nil {
+					continue
+				}
+				if !slots[j].chunk.IsChunkManifest {
+					dataChunks = append(dataChunks, slots[j].chunk)
+					continue
+				}
+				for _, c := range slots[j].result.chunks {
+					if !c.IsChunkManifest && max(c.Offset, startOffset) < min(c.Offset+int64(c.Size), stopOffset) {
+						dataChunks = append(dataChunks, c)
+					}
+				}
+			}
+			return dataChunks, nil, slot.result.err
+		}
+	}
+
+	for _, slot := range slots {
+		if slot.chunk == nil {
+			continue
+		}
+		if slot.result.err != nil {
+			if slot.result.internalCancel {
+				continue
+			}
+			return dataChunks, nil, slot.result.err
+		}
+		if !slot.chunk.IsChunkManifest {
+			dataChunks = append(dataChunks, slot.chunk)
+			continue
+		}
+
+		manifestChunks = append(manifestChunks, slot.chunk)
+		subDataChunks, subManifestChunks, subErr := r.resolve(slot.result.chunks, startOffset, stopOffset)
 		if subErr != nil {
 			return dataChunks, nil, subErr
 		}

weed/filer/filechunk_manifest_resolve_test.go

@@ -0,0 +1,595 @@
+package filer
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/proto"
+)
+
+type manifestReadFixture struct {
+	server    *httptest.Server
+	mu        sync.Mutex
+	started   []string
+	completed []string
+	active    atomic.Int32
+	maxActive atomic.Int32
+	loads     atomic.Int32
+	manifests map[string][]byte
+	delays    map[string]time.Duration
+	stopReads chan struct{}
+}
+
+func newManifestReadFixture(t testing.TB, manifests map[string][]*filer_pb.FileChunk, delays map[string]time.Duration) *manifestReadFixture {
+	t.Helper()
+	encoded := make(map[string][]byte, len(manifests))
+	for id, chunks := range manifests {
+		data, err := proto.Marshal(&filer_pb.FileChunkManifest{Chunks: chunks})
+		require.NoError(t, err)
+		encoded[id] = data
+	}
+
+	fixture := &manifestReadFixture{manifests: encoded, delays: delays, stopReads: make(chan struct{})}
+	listener, err := (&net.ListenConfig{}).Listen(context.Background(), "tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+	fixture.server = httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		id := strings.TrimPrefix(r.URL.Path, "/")
+		fixture.loads.Add(1)
+		fixture.active.Add(1)
+		fixture.updateMaxActive()
+		fixture.mu.Lock()
+		fixture.started = append(fixture.started, id)
+		fixture.mu.Unlock()
+		finishRead := func() {
+			fixture.active.Add(-1)
+			fixture.mu.Lock()
+			fixture.completed = append(fixture.completed, id)
+			fixture.mu.Unlock()
+		}
+
+		if delay := fixture.delays[id]; delay > 0 {
+			timer := time.NewTimer(delay)
+			select {
+			case <-r.Context().Done():
+				timer.Stop()
+				finishRead()
+				return
+			case <-fixture.stopReads:
+				timer.Stop()
+				finishRead()
+				return
+			case <-timer.C:
+			}
+		}
+		data, ok := fixture.manifests[id]
+		if !ok {
+			http.NotFound(w, r)
+			finishRead()
+			return
+		}
+		w.Header().Set("Content-Length", fmt.Sprintf("%d", len(data)))
+		w.WriteHeader(http.StatusOK)
+		_, _ = w.Write(data)
+		finishRead()
+	}))
+	fixture.server.Listener = listener
+	fixture.server.Start()
+	t.Cleanup(fixture.server.Close)
+	return fixture
+}
+
+func (f *manifestReadFixture) updateMaxActive() {
+	for {
+		current := f.active.Load()
+		maximum := f.maxActive.Load()
+		if current <= maximum || f.maxActive.CompareAndSwap(maximum, current) {
+			return
+		}
+	}
+}
+
+func (f *manifestReadFixture) lookup(ctx context.Context, fileID string) ([]string, error) {
+	if err := ctx.Err(); err != nil {
+		return nil, err
+	}
+	return []string{f.server.URL + "/" + fileID}, nil
+}
+
+func (f *manifestReadFixture) completionOrder() []string {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return append([]string(nil), f.completed...)
+}
+
+func resolveTestManifest(id string, offset int64) *filer_pb.FileChunk {
+	return &filer_pb.FileChunk{FileId: id, IsChunkManifest: true, Offset: offset, Size: 100}
+}
+
+func resolveTestData(id string, offset int64) *filer_pb.FileChunk {
+	return &filer_pb.FileChunk{FileId: id, Offset: offset, Size: 10}
+}
+
+func TestResolveChunkManifestParallelReadsAreBounded(t *testing.T) {
+	manifests := make(map[string][]*filer_pb.FileChunk)
+	inputs := make([]*filer_pb.FileChunk, 0, 8)
+	delays := make(map[string]time.Duration)
+	for i := 0; i < 8; i++ {
+		id := fmt.Sprintf("m%d", i)
+		manifests[id] = []*filer_pb.FileChunk{resolveTestData(fmt.Sprintf("d%d", i), int64(i*10))}
+		inputs = append(inputs, resolveTestManifest(id, int64(i*100)))
+		delays[id] = 50 * time.Millisecond
+	}
+	fixture := newManifestReadFixture(t, manifests, delays)
+
+	data, meta, err := ResolveChunkManifest(context.Background(), fixture.lookup, inputs, 0, 1000, nil)
+	require.NoError(t, err)
+	require.Len(t, data, len(inputs))
+	require.Len(t, meta, len(inputs))
+	require.GreaterOrEqual(t, fixture.maxActive.Load(), int32(2), "independent manifests should overlap")
+	require.LessOrEqual(t, fixture.maxActive.Load(), int32(4), "manifest reads must be bounded")
+}
+
+func TestResolveChunkManifestPreservesInputOrderWhenReadsCompleteOutOfOrder(t *testing.T) {
+	fixture := newManifestReadFixture(t,
+		map[string][]*filer_pb.FileChunk{
+			"slow": {resolveTestData("slow-data", 0)},
+			"fast": {resolveTestData("fast-data", 100)},
+		},
+		map[string]time.Duration{"slow": 80 * time.Millisecond, "fast": 5 * time.Millisecond},
+	)
+
+	data, meta, err := ResolveChunkManifest(context.Background(), fixture.lookup, []*filer_pb.FileChunk{
+		resolveTestManifest("slow", 0),
+		resolveTestManifest("fast", 100),
+	}, 0, 200, nil)
+	require.NoError(t, err)
+	require.Equal(t, []string{"slow-data", "fast-data"}, fileIDs(data))
+	require.Equal(t, []string{"slow", "fast"}, fileIDs(meta))
+	require.Equal(t, []string{"fast", "slow"}, fixture.completionOrder())
+}
+
+func TestResolveChunkManifestNestedManifestsKeepDepthFirstOrder(t *testing.T) {
+	fixture := newManifestReadFixture(t,
+		map[string][]*filer_pb.FileChunk{
+			"parent":  {resolveTestData("parent-data", 0), resolveTestManifest("nested", 10)},
+			"sibling": {resolveTestData("sibling-data", 20)},
+			"nested":  {resolveTestData("nested-data", 10)},
+		},
+		map[string]time.Duration{"parent": 60 * time.Millisecond, "sibling": 5 * time.Millisecond},
+	)
+
+	data, meta, err := ResolveChunkManifest(context.Background(), fixture.lookup, []*filer_pb.FileChunk{
+		resolveTestManifest("parent", 0),
+		resolveTestManifest("sibling", 20),
+	}, 0, 100, nil)
+	require.NoError(t, err)
+	require.Equal(t, []string{"parent-data", "nested-data", "sibling-data"}, fileIDs(data))
+	require.Equal(t, []string{"parent", "nested", "sibling"}, fileIDs(meta))
+	require.Equal(t, []string{"sibling", "parent", "nested"}, fixture.completionOrder())
+}
+
+func TestResolveChunkManifestFiltersBeforeReadingAndRecursively(t *testing.T) {
+	fixture := newManifestReadFixture(t,
+		map[string][]*filer_pb.FileChunk{
+			"outside": {resolveTestData("never-read", 0)},
+			"inside":  {resolveTestData("before", 0), resolveTestData("selected", 100), resolveTestData("after", 200)},
+		}, nil)
+
+	data, meta, err := ResolveChunkManifest(context.Background(), fixture.lookup, []*filer_pb.FileChunk{
+		resolveTestManifest("outside", 0),
+		resolveTestManifest("inside", 90),
+	}, 100, 200, nil)
+	require.NoError(t, err)
+	require.Equal(t, []string{"selected"}, fileIDs(data))
+	require.Equal(t, []string{"inside"}, fileIDs(meta))
+	require.Equal(t, int32(1), fixture.loads.Load(), "the non-overlapping parent manifest must not be read")
+}
+
+func TestResolveChunkManifestPropagatesReadAndFormatErrorsInInputOrder(t *testing.T) {
+	fixture := newManifestReadFixture(t,
+		map[string][]*filer_pb.FileChunk{
+			"valid":  {resolveTestData("valid-data", 0)},
+			"broken": nil,
+		}, map[string]time.Duration{"broken": 50 * time.Millisecond})
+	fixture.manifests["broken"] = []byte("not a protobuf manifest")
+
+	data, meta, err := ResolveChunkManifest(context.Background(), fixture.lookup, []*filer_pb.FileChunk{
+		resolveTestData("plain", 0),
+		resolveTestManifest("valid", 10),
+		resolveTestManifest("broken", 20),
+	}, 0, 100, nil)
+	require.Error(t, err)
+	require.Nil(t, meta)
+	require.Equal(t, []string{"plain", "valid-data"}, fileIDs(data))
+	require.Contains(t, err.Error(), "fail to unmarshal manifest broken")
+
+	lookupErr := errors.New("lookup failed")
+	lookup := func(context.Context, string) ([]string, error) { return nil, lookupErr }
+	_, _, err = ResolveChunkManifest(context.Background(), lookup, []*filer_pb.FileChunk{resolveTestManifest("lookup-error", 0)}, 0, 100, nil)
+	require.ErrorIs(t, err, lookupErr)
+}
+
+func TestResolveChunkManifestCancellationStopsAllReads(t *testing.T) {
+	fixture := newManifestReadFixture(t,
+		map[string][]*filer_pb.FileChunk{
+			"one": {resolveTestData("one-data", 0)},
+			"two": {resolveTestData("two-data", 10)},
+		},
+		map[string]time.Duration{"one": time.Second, "two": time.Second},
+	)
+	ctx, cancel := context.WithCancel(context.Background())
+	result := make(chan error, 1)
+	go func() {
+		_, _, err := ResolveChunkManifest(ctx, fixture.lookup, []*filer_pb.FileChunk{
+			resolveTestManifest("one", 0),
+			resolveTestManifest("two", 10),
+		}, 0, 100, nil)
+		result <- err
+	}()
+
+	deadline := time.Now().Add(time.Second)
+	for fixture.loads.Load() < 1 && time.Now().Before(deadline) {
+		time.Sleep(time.Millisecond)
+	}
+	close(fixture.stopReads)
+	cancel()
+	err := <-result
+	require.ErrorIs(t, err, context.Canceled)
+	require.Eventually(t, func() bool {
+		return fixture.active.Load() == 0
+	}, time.Second, time.Millisecond, "all manifest readers must exit after ResolveChunkManifest returns")
+}
+
+type manifestFailureReadFixture struct {
+	server       *httptest.Server
+	fastRelease  chan struct{}
+	slowStarted  chan struct{}
+	fastStarted  chan struct{}
+	slowCanceled chan struct{}
+}
+
+func newManifestFailureReadFixture(t testing.TB) *manifestFailureReadFixture {
+	t.Helper()
+	fixture := &manifestFailureReadFixture{
+		fastRelease:  make(chan struct{}),
+		slowStarted:  make(chan struct{}),
+		fastStarted:  make(chan struct{}),
+		slowCanceled: make(chan struct{}),
+	}
+	fixture.server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		switch strings.TrimPrefix(r.URL.Path, "/") {
+		case "fast":
+			close(fixture.fastStarted)
+			<-fixture.fastRelease
+			w.Header().Set("Content-Length", "7")
+			w.WriteHeader(http.StatusOK)
+			_, _ = w.Write([]byte("invalid"))
+		case "slow":
+			close(fixture.slowStarted)
+			<-r.Context().Done()
+			close(fixture.slowCanceled)
+		}
+	}))
+	t.Cleanup(fixture.server.Close)
+	return fixture
+}
+
+func (f *manifestFailureReadFixture) lookup(ctx context.Context, fileID string) ([]string, error) {
+	if err := ctx.Err(); err != nil {
+		return nil, err
+	}
+	return []string{f.server.URL + "/" + fileID}, nil
+}
+
+func waitForManifestFailureSignal(t *testing.T, signal <-chan struct{}) {
+	t.Helper()
+	select {
+	case <-signal:
+	case <-time.After(time.Second):
+		t.Fatal("timed out waiting for manifest request")
+	}
+}
+
+func TestResolveChunkManifestFastFailureCancelsSlowSibling(t *testing.T) {
+	testCases := []struct {
+		name       string
+		inputIDs   []string
+		expectedID string
+	}{
+		{name: "fast then slow returns quickly", inputIDs: []string{"fast", "slow"}, expectedID: "fast"},
+		{name: "slow then fast keeps real error", inputIDs: []string{"slow", "fast"}, expectedID: "fast"},
+	}
+
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+			fixture := newManifestFailureReadFixture(t)
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+
+			chunks := make([]*filer_pb.FileChunk, 0, len(testCase.inputIDs))
+			for i, id := range testCase.inputIDs {
+				chunks = append(chunks, resolveTestManifest(id, int64(i*100)))
+			}
+			result := make(chan error, 1)
+			go func() {
+				_, _, err := ResolveChunkManifest(ctx, fixture.lookup, chunks, 0, 200, nil)
+				result <- err
+			}()
+
+			waitForManifestFailureSignal(t, fixture.fastStarted)
+			waitForManifestFailureSignal(t, fixture.slowStarted)
+			close(fixture.fastRelease)
+
+			var err error
+			select {
+			case err = <-result:
+			case <-time.After(time.Second):
+				cancel()
+				select {
+				case <-result:
+				case <-time.After(time.Second):
+					t.Fatal("ResolveChunkManifest did not finish after caller cancellation")
+				}
+				t.Fatal("fast manifest failure waited for the slow sibling")
+			}
+			require.Error(t, err)
+			require.Contains(t, err.Error(), "fail to unmarshal manifest "+testCase.expectedID)
+			require.NotErrorIs(t, err, context.Canceled)
+			waitForManifestFailureSignal(t, fixture.slowCanceled)
+		})
+	}
+}
+
+func TestResolveChunkManifestFastFailureCancelsEncryptedSibling(t *testing.T) {
+	fixture := newManifestFailureReadFixture(t)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	slowChunk := resolveTestManifest("slow", 0)
+	slowChunk.CipherKey = []byte("0123456789abcdef")
+	chunks := []*filer_pb.FileChunk{
+		slowChunk,
+		resolveTestManifest("fast", 100),
+	}
+	result := make(chan error, 1)
+	go func() {
+		_, _, err := ResolveChunkManifest(ctx, fixture.lookup, chunks, 0, 200, nil)
+		result <- err
+	}()
+
+	waitForManifestFailureSignal(t, fixture.fastStarted)
+	waitForManifestFailureSignal(t, fixture.slowStarted)
+	close(fixture.fastRelease)
+
+	var err error
+	select {
+	case err = <-result:
+	case <-time.After(time.Second):
+		cancel()
+		select {
+		case <-result:
+		case <-time.After(time.Second):
+			t.Fatal("ResolveChunkManifest did not finish after caller cancellation")
+		}
+		t.Fatal("fast manifest failure waited for the encrypted slow sibling")
+	}
+	require.Error(t, err)
+	require.Contains(t, err.Error(), "fail to unmarshal manifest fast")
+	require.NotErrorIs(t, err, context.Canceled)
+	waitForManifestFailureSignal(t, fixture.slowCanceled)
+}
+
+func TestResolveChunkManifestKeepsRealErrorAfterInternalCancellation(t *testing.T) {
+	earlyStarted := make(chan struct{})
+	lateStarted := make(chan struct{})
+	earlyErr := errors.New("early lookup failed")
+	lateErr := errors.New("late lookup failed")
+	lookup := func(ctx context.Context, fileID string) ([]string, error) {
+		switch fileID {
+		case "early":
+			close(earlyStarted)
+			<-ctx.Done()
+			return nil, earlyErr
+		case "late":
+			close(lateStarted)
+			return nil, lateErr
+		default:
+			return nil, fmt.Errorf("unexpected manifest %s", fileID)
+		}
+	}
+
+	result := make(chan error, 1)
+	go func() {
+		_, _, err := ResolveChunkManifest(context.Background(), lookup, []*filer_pb.FileChunk{
+			resolveTestManifest("early", 0),
+			resolveTestManifest("late", 100),
+		}, 0, 200, nil)
+		result <- err
+	}()
+	waitForManifestFailureSignal(t, earlyStarted)
+	waitForManifestFailureSignal(t, lateStarted)
+
+	select {
+	case err := <-result:
+		require.ErrorIs(t, err, earlyErr)
+		require.NotErrorIs(t, err, lateErr)
+	case <-time.After(time.Second):
+		t.Fatal("ResolveChunkManifest did not return the input-order error")
+	}
+}
+
+func TestResolveChunkManifestReturnsLaterFailureWithoutRecursingEarlierChildren(t *testing.T) {
+	fixture := newManifestReadFixture(t,
+		map[string][]*filer_pb.FileChunk{
+			"a":       {resolveTestManifest("a-child", 0)},
+			"a-child": {resolveTestData("a-child-data", 0)},
+			"b":       nil,
+		},
+		map[string]time.Duration{"a": 5 * time.Millisecond, "b": 50 * time.Millisecond, "a-child": 2 * time.Second},
+	)
+	fixture.manifests["b"] = []byte("not a protobuf manifest")
+
+	start := time.Now()
+	data, meta, err := ResolveChunkManifest(context.Background(), fixture.lookup, []*filer_pb.FileChunk{
+		resolveTestManifest("a", 0),
+		resolveTestManifest("b", 100),
+	}, 0, 200, nil)
+	elapsed := time.Since(start)
+
+	require.Error(t, err)
+	require.Contains(t, err.Error(), "fail to unmarshal manifest b")
+	require.Empty(t, data, "earlier manifest's children must not be recursed when a later manifest already failed")
+	require.Nil(t, meta)
+	require.Equal(t, int32(2), fixture.loads.Load(), "only the two top-level manifests must be read, not a-child")
+	require.Less(t, elapsed, time.Second, "must return promptly without waiting for a-child's slow read")
+}
+
+func TestResolveChunkManifestExcludesOutOfRangeChildrenOnLaterFailure(t *testing.T) {
+	fixture := newManifestReadFixture(t,
+		map[string][]*filer_pb.FileChunk{
+			"a": {
+				resolveTestData("a-in-range", 0),
+				resolveTestData("a-out-of-range", 500),
+			},
+			"b": nil,
+		},
+		map[string]time.Duration{"b": 50 * time.Millisecond},
+	)
+	fixture.manifests["b"] = []byte("not a protobuf manifest")
+
+	data, _, err := ResolveChunkManifest(context.Background(), fixture.lookup, []*filer_pb.FileChunk{
+		resolveTestManifest("a", 0),
+		resolveTestManifest("b", 100),
+	}, 0, 200, nil)
+	require.Error(t, err)
+	require.Contains(t, err.Error(), "fail to unmarshal manifest b")
+	require.Equal(t, []string{"a-in-range"}, fileIDs(data), "out-of-range child chunks must be excluded from partial results")
+}
+
+func TestResolveChunkManifestQueuesJobsBeyondWorkerCount(t *testing.T) {
+	// More manifests than workers. The first four stall; the fifth fails
+	// promptly. With a buffered job channel, the fifth job is queued without
+	// blocking submission. When a stalled read is released, a worker picks up
+	// the failing job, which cancels the batch and unblocks the rest.
+	stallRelease := make(chan struct{})
+	stallStarted := make(chan struct{}, 4)
+	lookup := func(ctx context.Context, fileID string) ([]string, error) {
+		switch fileID {
+		case "stall-0", "stall-1", "stall-2", "stall-3":
+			stallStarted <- struct{}{}
+			select {
+			case <-stallRelease:
+				return nil, nil
+			case <-ctx.Done():
+				return nil, ctx.Err()
+			}
+		case "fail":
+			return nil, errors.New("prompt lookup failure")
+		default:
+			return nil, fmt.Errorf("unexpected manifest %s", fileID)
+		}
+	}
+
+	result := make(chan error, 1)
+	go func() {
+		_, _, err := ResolveChunkManifest(context.Background(), lookup, []*filer_pb.FileChunk{
+			resolveTestManifest("stall-0", 0),
+			resolveTestManifest("stall-1", 100),
+			resolveTestManifest("stall-2", 200),
+			resolveTestManifest("stall-3", 300),
+			resolveTestManifest("fail", 400),
+		}, 0, 500, nil)
+		result <- err
+	}()
+
+	// Wait until all 4 workers are stalled.
+	for i := 0; i < 4; i++ {
+		waitForManifestFailureSignal(t, stallStarted)
+	}
+
+	// Release the stalled reads so workers can pick up the queued failing job.
+	close(stallRelease)
+
+	select {
+	case err := <-result:
+		require.Error(t, err)
+		require.Contains(t, err.Error(), "fail to read manifest fail")
+	case <-time.After(2 * time.Second):
+		t.Fatal("failing job was not picked up after releasing stalled workers")
+	}
+}
+
+func fileIDs(chunks []*filer_pb.FileChunk) []string {
+	ids := make([]string, 0, len(chunks))
+	for _, chunk := range chunks {
+		ids = append(ids, chunk.GetFileIdString())
+	}
+	return ids
+}
+
+func resolveChunkManifestSerialForBenchmark(ctx context.Context, lookupFileIdFn func(context.Context, string) ([]string, error), chunks []*filer_pb.FileChunk, startOffset, stopOffset int64) (dataChunks, manifestChunks []*filer_pb.FileChunk, err error) {
+	for _, chunk := range chunks {
+		if max(chunk.Offset, startOffset) >= min(chunk.Offset+int64(chunk.Size), stopOffset) {
+			continue
+		}
+		if !chunk.IsChunkManifest {
+			dataChunks = append(dataChunks, chunk)
+			continue
+		}
+
+		resolvedChunks, resolveErr := ResolveOneChunkManifest(ctx, lookupFileIdFn, chunk, nil)
+		if resolveErr != nil {
+			return dataChunks, nil, resolveErr
+		}
+		manifestChunks = append(manifestChunks, chunk)
+		subDataChunks, subManifestChunks, subErr := resolveChunkManifestSerialForBenchmark(ctx, lookupFileIdFn, resolvedChunks, startOffset, stopOffset)
+		if subErr != nil {
+			return dataChunks, nil, subErr
+		}
+		dataChunks = append(dataChunks, subDataChunks...)
+		manifestChunks = append(manifestChunks, subManifestChunks...)
+	}
+	return dataChunks, manifestChunks, nil
+}
+
+func BenchmarkResolveChunkManifestSerialVsParallel(b *testing.B) {
+	manifests := make(map[string][]*filer_pb.FileChunk)
+	inputs := make([]*filer_pb.FileChunk, 0, 8)
+	delays := make(map[string]time.Duration)
+	for i := 0; i < 8; i++ {
+		id := fmt.Sprintf("m%d", i)
+		manifests[id] = []*filer_pb.FileChunk{resolveTestData(fmt.Sprintf("d%d", i), int64(i*10))}
+		inputs = append(inputs, resolveTestManifest(id, int64(i*10)))
+		delays[id] = time.Millisecond
+	}
+	fixture := newManifestReadFixture(b, manifests, delays)
+	b.ReportAllocs()
+
+	b.Run("serial", func(b *testing.B) {
+		for i := 0; i < b.N; i++ {
+			_, _, err := resolveChunkManifestSerialForBenchmark(context.Background(), fixture.lookup, inputs, 0, 1000)
+			if err != nil {
+				b.Fatal(err)
+			}
+		}
+	})
+	b.Run("parallel", func(b *testing.B) {
+		for i := 0; i < b.N; i++ {
+			_, _, err := ResolveChunkManifest(context.Background(), fixture.lookup, inputs, 0, 1000, nil)
+			if err != nil {
+				b.Fatal(err)
+			}
+		}
+	})
+}

weed/util/http/http_global_client_util.go

@@ -98,11 +98,11 @@ func Post(url string, values url.Values) ([]byte, error) {
 // github.com/seaweedfs/seaweedfs/unmaintained/repeated_vacuum/repeated_vacuum.go
 // may need increasing http.Client.Timeout
 func Get(url string) ([]byte, bool, error) {
-	return GetAuthenticated(url, "")
+	return GetAuthenticated(context.Background(), url, "")
 }
 
-func GetAuthenticated(url, jwt string) ([]byte, bool, error) {
-	request, err := http.NewRequest(http.MethodGet, url, nil)
+func GetAuthenticated(ctx context.Context, url, jwt string) ([]byte, bool, error) {
+	request, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
 	if err != nil {
 		return nil, true, err
 	}
@@ -111,7 +111,9 @@ func GetAuthenticated(url, jwt string) ([]byte, bool, error) {
 
 	response, err := GetGlobalHttpClient().Do(request)
 	if err != nil {
+		if ctx.Err() == nil {
 			recordUnreachable(request.URL.Host)
+		}
 		return nil, true, err
 	}
 	recordReachable(request.URL.Host)
@@ -135,6 +137,9 @@ func GetAuthenticated(url, jwt string) ([]byte, bool, error) {
 		return nil, retryable, fmt.Errorf("%s: %s", url, response.Status)
 	}
 	if err != nil {
+		if ctxErr := ctx.Err(); ctxErr != nil {
+			return nil, false, ctxErr
+		}
 		return nil, false, err
 	}
 	return b, false, nil
@@ -454,7 +459,7 @@ func ReadUrlAsStream(ctx context.Context, fileUrl, jwt string, cipherKey []byte,
 }
 
 func readEncryptedUrl(ctx context.Context, fileUrl, jwt string, cipherKey []byte, isContentCompressed bool, isFullChunk bool, offset int64, size int, fn func(data []byte)) (bool, error) {
-	encryptedData, retryable, err := GetAuthenticated(fileUrl, jwt)
+	encryptedData, retryable, err := GetAuthenticated(ctx, fileUrl, jwt)
 	if err != nil {
 		return retryable, fmt.Errorf("fetch %s: %v", fileUrl, err)
 	}