perf(cache-proxy): bound, hedge, and adapt peer fetches

cmd/cache-proxy/block_serve.go

@@ -28,6 +28,10 @@ var (
 		Name: "cache_proxy_block_reads_total",
 		Help: "Blocks resolved while assembling responses, by source",
 	}, []string{"source"}) // local, peer, s3
+	peerFillHedgedTotal = promauto.NewCounter(prometheus.CounterOpts{
+		Name: "cache_proxy_peer_fill_hedged_total",
+		Help: "Block fills sent to origin because the peer wait budget expired (the peer fetch continues in background)",
+	})
 	// requestDurationSeconds is shared between the block-serve path (this
 	// file) and the forward-proxy path (proxy.go); buckets start at 1ms
 	// because a local cache hit can be sub-millisecond and top out around 8s
@@ -178,6 +182,42 @@ func (p *CacheProxy) fetchOriginSpan(r *http.Request, blockSize, firstIdx, lastI
 	return nil
 }
 
+// peerFillConcurrency bounds how many of one request's blocks are fetched
+// from peers at the same time. Requests typically span one or two blocks, so
+// this only matters for wide spans, where it keeps a single request from
+// monopolizing peer bandwidth.
+const peerFillConcurrency = 8
+
+// peerFill is the handle for one block's background peer fetch: done closes
+// when the fetch finishes, and ok reports whether a peer delivered the block.
+type peerFill struct {
+	done chan struct{}
+	ok   bool
+}
+
+// waitFill waits for a peer fill until deadline. filled reports whether the
+// block is now on local disk; hedged reports that the budget expired first —
+// the block should be fetched from origin while the fill keeps running in the
+// background (a late fill still populates the cache for future requests).
+func waitFill(f *peerFill, deadline time.Time) (filled, hedged bool) {
+	select {
+	case <-f.done:
+		return f.ok, false
+	default:
+	}
+	if wait := time.Until(deadline); wait > 0 {
+		timer := time.NewTimer(wait)
+		defer timer.Stop()
+		select {
+		case <-f.done:
+			return f.ok, false
+		case <-timer.C:
+		}
+	}
+	peerFillHedgedTotal.Inc()
+	return false, true
+}
+
 // serveBlockAligned serves a cacheable GET whose Range is an absolute
 // bytes=start-end pair from block-aligned cache entries: local disk, then
 // peers, then coalesced origin fetches for contiguous missing runs (chunked
@@ -218,9 +258,45 @@ func (p *CacheProxy) serveBlockAligned(w http.ResponseWriter, r *http.Request, r
 		return false
 	}
 
+	// Launch peer fills for every block not already on local disk, in
+	// parallel — fetching them one at a time made a request's peer wait scale
+	// linearly with its block count. Phase 1 below consumes each block's
+	// result in order, so the miss-run coalescing and single-flight keys are
+	// unchanged. All waits share one absolute deadline: the fills started
+	// together, so time spent waiting on one block has also elapsed for the
+	// rest.
+	var fills map[int64]*peerFill
+	var fillDeadline time.Time
+	if p.peers != nil {
+		fills = make(map[int64]*peerFill, blockCount)
+		sem := make(chan struct{}, peerFillConcurrency)
+		for idx := firstIdx; idx <= lastIdx; idx++ {
+			key := BlockKey(urlStr, idx, p.blockSize)
+			if p.store.Has(key) {
+				continue
+			}
+			f := &peerFill{done: make(chan struct{})}
+			fills[idx] = f
+			go func() {
+				defer close(f.done)
+				sem <- struct{}{}
+				defer func() { <-sem }()
+				_, _, ok := p.peers.FetchFromPeers(key, func(rd io.Reader) (int64, error) {
+					return p.store.PutStream(key, rd)
+				})
+				f.ok = ok
+			}()
+		}
+		budget := p.peerFillWaitBudget
+		if budget <= 0 {
+			budget = defaultPeerFillWaitBudget
+		}
+		fillDeadline = time.Now().Add(budget)
+	}
+
 	// Phase 1: ensure every block is present locally. Track sources for the
 	// hit/miss accounting and the log line.
-	var nLocal, nPeer, nOrigin int64
+	var nLocal, nPeer, nOrigin, nHedged int64
 	var missRunStart int64 = -1
 	flushRun := func(runEnd int64) bool {
 		if missRunStart < 0 {
@@ -277,32 +353,39 @@ func (p *CacheProxy) serveBlockAligned(w http.ResponseWriter, r *http.Request, r
 	}
 	for idx := firstIdx; idx <= lastIdx; idx++ {
 		key := BlockKey(urlStr, idx, p.blockSize)
-		if p.store.Has(key) {
-			if !flushRun(idx - 1) {
-				return true // error already written
-			}
-			if idx > lastIdx {
-				break
-			}
-			nLocal++
-			continue
-		}
-		if p.peers != nil {
+		// The fill check must precede the Has check: a block our own prefetch
+		// has already landed would otherwise pass Has and be counted as a
+		// local hit, corrupting the local/peer split in the log and metrics.
+		if f := fills[idx]; f != nil {
 			peerStart := time.Now()
-			_, _, ok := p.peers.FetchFromPeers(key, func(rd io.Reader) (int64, error) {
-				return p.store.PutStream(key, rd)
-			})
+			filled, hedged := waitFill(f, fillDeadline)
 			peerDur += time.Since(peerStart)
-			if ok {
+			if hedged {
+				nHedged++
+			}
+			if filled {
 				if !flushRun(idx - 1) {
-					return true
+					return true // error already written
 				}
 				if idx > lastIdx {
 					break
 				}
 				nPeer++
 				continue
 			}
+			// Fall through: a concurrent request may have landed the block
+			// since the fill failed, so the Has check below can still rescue
+			// it from an origin fetch.
+		}
+		if p.store.Has(key) {
+			if !flushRun(idx - 1) {
+				return true
+			}
+			if idx > lastIdx {
+				break
+			}
+			nLocal++
+			continue
 		}
 		if missRunStart < 0 {
 			missRunStart = idx
@@ -477,6 +560,7 @@ func (p *CacheProxy) serveBlockAligned(w http.ResponseWriter, r *http.Request, r
 	requestDurationSeconds.WithLabelValues("block", source).Observe(totalDur.Seconds())
 	slog.Info("Served.", "source", "blocks", "url", urlStr, "range", rangeHeader,
 		"bytes", served, "blocks_local", nLocal, "blocks_peer", nPeer, "blocks_s3", nOrigin,
+		"blocks_hedged", nHedged,
 		"dur_ms", totalDur.Milliseconds(), "peer_ms", peerDur.Milliseconds(),
 		"s3_ms", s3Dur.Milliseconds(), "write_ms", writeDur.Milliseconds())
 	return true

cmd/cache-proxy/block_serve_test.go

@@ -12,6 +12,7 @@ import (
 	"sync"
 	"sync/atomic"
 	"testing"
+	"time"
 
 	"github.com/prometheus/client_golang/prometheus"
 )
@@ -502,6 +503,134 @@ func TestServeBlockAlignedPeerFillCountsAsHit(t *testing.T) {
 	}
 }
 
+// TestServeBlockAlignedPeerFillsRunConcurrently locks in the parallel peer
+// fill behavior: the peer's /cache/get handlers gate on all three of the
+// request's blocks being fetched at once, so a regression to one-at-a-time
+// fills can never open the gate — the request would hedge to a closed origin
+// and fail instead of assembling the response.
+func TestServeBlockAlignedPeerFillsRunConcurrently(t *testing.T) {
+	const blockSize = 1024
+	const nBlocks = 3
+	origin := originServer(t, nBlocks*blockSize)
+	target := origin.URL + "/bucket/f.parquet"
+
+	body := make([]byte, nBlocks*blockSize)
+	for i := range body {
+		body[i] = byte(i % 251)
+	}
+	keys := make(map[string]int64, nBlocks)
+	for idx := int64(0); idx < nBlocks; idx++ {
+		keys[BlockKey(target, idx, blockSize)] = idx
+	}
+
+	var arrivals int32
+	gate := make(chan struct{})
+	mux := http.NewServeMux()
+	mux.HandleFunc("/cache/has", func(w http.ResponseWriter, r *http.Request) {
+		if _, ok := keys[r.URL.Query().Get("key")]; ok {
+			w.WriteHeader(http.StatusOK)
+			return
+		}
+		w.WriteHeader(http.StatusNotFound)
+	})
+	mux.HandleFunc("/cache/get", func(w http.ResponseWriter, r *http.Request) {
+		idx, ok := keys[r.URL.Query().Get("key")]
+		if !ok {
+			w.WriteHeader(http.StatusNotFound)
+			return
+		}
+		if atomic.AddInt32(&arrivals, 1) == nBlocks {
+			close(gate)
+		}
+		<-gate
+		block := body[idx*blockSize : (idx+1)*blockSize]
+		w.Header().Set("Content-Length", strconv.Itoa(len(block)))
+		_, _ = w.Write(block)
+	})
+	srv := httptest.NewServer(mux)
+	t.Cleanup(srv.Close)
+
+	origin.Close() // all blocks must come from the peer; a hedge to origin fails loudly
+
+	store, err := NewDiskCache(t.TempDir(), 80)
+	if err != nil {
+		t.Fatal(err)
+	}
+	p := NewCacheProxy(store, peerManagerWith([]string{strings.TrimPrefix(srv.URL, "http://")}), []string{})
+	p.blockSize = blockSize
+	p.maxSpanBlocks = 8
+
+	w := doBlockRequest(t, p, target, fmt.Sprintf("bytes=0-%d", nBlocks*blockSize-1))
+	if w.Code != http.StatusPartialContent {
+		t.Fatalf("status %d, want 206 (sequential fills would starve the gate and hedge into the closed origin)", w.Code)
+	}
+	if got := w.Body.Bytes(); string(got) != string(body) {
+		t.Fatalf("body mismatch: got %d bytes, want %d", len(got), len(body))
+	}
+	if got := atomic.LoadInt32(&arrivals); got != nBlocks {
+		t.Fatalf("peer /cache/get arrivals = %d, want %d", got, nBlocks)
+	}
+}
+
+// TestServeBlockAlignedHedgesSlowPeerToOrigin covers the wait budget: a peer
+// that claims the block but stalls the body transfer must not pin the request
+// for the full peer get timeout — after peerFillWaitBudget the block is
+// fetched from origin and the response completes.
+func TestServeBlockAlignedHedgesSlowPeerToOrigin(t *testing.T) {
+	const blockSize = 1024
+	origin := originServer(t, 4*blockSize)
+	t.Cleanup(origin.Close)
+	target := origin.URL + "/bucket/f.parquet"
+	key := BlockKey(target, 0, blockSize)
+
+	release := make(chan struct{})
+	mux := http.NewServeMux()
+	mux.HandleFunc("/cache/has", func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Query().Get("key") == key {
+			w.WriteHeader(http.StatusOK)
+			return
+		}
+		w.WriteHeader(http.StatusNotFound)
+	})
+	mux.HandleFunc("/cache/get", func(w http.ResponseWriter, r *http.Request) {
+		<-release // stall the body transfer past the wait budget
+		w.WriteHeader(http.StatusNotFound)
+	})
+	srv := httptest.NewServer(mux)
+	t.Cleanup(srv.Close)
+	t.Cleanup(func() { close(release) }) // LIFO: unblock the handler before srv.Close waits on it
+
+	store, err := NewDiskCache(t.TempDir(), 80)
+	if err != nil {
+		t.Fatal(err)
+	}
+	p := NewCacheProxy(store, peerManagerWith([]string{strings.TrimPrefix(srv.URL, "http://")}), []string{})
+	p.blockSize = blockSize
+	p.maxSpanBlocks = 8
+	p.peerFillWaitBudget = 50 * time.Millisecond
+
+	hedgedBefore := counterValue(t, peerFillHedgedTotal)
+	s3ReadsBefore := counterValue(t, blockReadsTotal.WithLabelValues("s3"))
+
+	w := doBlockRequest(t, p, target, "bytes=0-99")
+	if w.Code != http.StatusPartialContent {
+		t.Fatalf("status %d, want 206", w.Code)
+	}
+	want := make([]byte, 100)
+	for i := range want {
+		want[i] = byte(i % 251)
+	}
+	if got := w.Body.Bytes(); string(got) != string(want) {
+		t.Fatalf("body mismatch: got %d bytes", len(got))
+	}
+	if got := counterValue(t, peerFillHedgedTotal); got != hedgedBefore+1 {
+		t.Fatalf("peerFillHedgedTotal delta = %v, want 1", got-hedgedBefore)
+	}
+	if got := counterValue(t, blockReadsTotal.WithLabelValues("s3")); got != s3ReadsBefore+1 {
+		t.Fatalf("blockReadsTotal{s3} delta = %v, want 1 (hedged block must be served from origin)", got-s3ReadsBefore)
+	}
+}
+
 // TestServeBlockAlignedDoesNotReverifyPastObjectEOF covers the cold-request
 // case where the requested end lies past the true object size. The validated
 // Content-Range must clamp the response and shrink the block span immediately,

cmd/cache-proxy/peers.go

@@ -31,8 +31,16 @@ var (
 // separate: sharing one 2s budget (as the original code did) meant a slow
 // has-race ate into the body-transfer time and large peer ranges timed out
 // mid-stream, silently downgrading the hit to a full S3 fetch.
+//
+// The has budget is deliberately tight: a healthy peer answers the probe in
+// single-digit ms even under load, while an origin block fetch costs roughly
+// 200ms — so once a probe has gone unanswered for ~150ms, waiting longer only
+// delays a faster origin fallback. FetchFromPeers waits for every peer's "no"
+// before giving up, which means the slowest peer in the fleet gates every
+// cold fill; with a 1s budget, production trails showed 16% of cold-scan
+// requests burning the full second on that drain.
 const (
-	peerHasTimeout = 1 * time.Second
+	peerHasTimeout = 150 * time.Millisecond
 	peerGetTimeout = 30 * time.Second
 )
 

cmd/cache-proxy/proxy.go

@@ -63,6 +63,12 @@ type CacheProxy struct {
 	blockSize     int64
 	maxSpanBlocks int64
 
+	// peerFillWaitBudget bounds how long a block request waits on its
+	// concurrent peer fills before falling back to a coalesced origin fetch.
+	// The fills keep running past the budget and still populate the cache;
+	// see serveBlockAligned.
+	peerFillWaitBudget time.Duration
+
 	// objectSizes remembers validated complete lengths learned from origin
 	// Content-Range responses. Disk blocks remain the durable cache; this map
 	// lets subsequent requests in the same process emit precise range headers
@@ -85,6 +91,12 @@ const (
 	defaultOriginRetryMaxAttempts    = 4
 	defaultOriginRetryInitialBackoff = 100 * time.Millisecond
 	defaultOriginRetryMaxBackoff     = 1 * time.Second
+
+	// defaultPeerFillWaitBudget caps a block request's wait on peer fills. It
+	// covers the 150ms has-round plus a healthy same-VPC 1MiB body transfer
+	// with slack; a holder that can't deliver within it is slower than the
+	// ~200ms origin path the request falls back to.
+	defaultPeerFillWaitBudget = 400 * time.Millisecond
 )
 
 type singleFlight struct {
@@ -133,6 +145,7 @@ func NewCacheProxy(store *DiskCache, peers *PeerManager, cacheHostSuffixes []str
 		originRetryInitialBackoff: defaultOriginRetryInitialBackoff,
 		originRetryMaxBackoff:     defaultOriginRetryMaxBackoff,
 		cacheHostSuffixes:         cacheHostSuffixes,
+		peerFillWaitBudget:        defaultPeerFillWaitBudget,
 	}
 }
 

docs/metrics.md

@@ -223,6 +223,7 @@ These are emitted by the standalone `cache-proxy` binary itself (`cmd/cache-prox
 | `cache_proxy_request_duration_seconds` | Histogram | `path`, `source` | End-to-end duration of a served request. `path` is `block` (block-aligned cache path) or `forward` (uncached forward-proxy path); `source` is `local`, `peer`, or `s3` for `block`, and always `origin` for `forward`. |
 | `cache_proxy_forward_requests_total` | Counter | `method` | Requests handled by the uncached forward-proxy path, by HTTP method. |
 | `cache_proxy_inflight_requests` | Gauge | None | Requests currently being handled by the proxy's request entry point; the queue-depth signal. |
+| `cache_proxy_peer_fill_hedged_total` | Counter | None | Blocks fetched from origin because their peer fill did not finish within the wait budget; the peer fetch continues in the background and still populates the cache. A high rate means peers are answering slower than the origin path. |
 
 ## PromQL recipes