| 468 | reader, size, ok := p.store.openFile(cacheKey) |
| 469 | if !ok { |
| 470 | // The entry was evicted in the narrow window between commit and serve. |
| 471 | // 502 so httpfs retries rather than receiving a truncated body. |
| 472 | slog.Warn("Cached entry vanished before serve.", "url", r.URL.String(), "range", rangeHeader) |
| 473 | http.Error(w, "cache entry vanished", http.StatusBadGateway) |
| 474 | return |
| 475 | } |
| 476 | defer func() { _ = reader.Close() }() |
| 477 | span.SetAttributes( |
| 478 | attribute.String("duckgres.cache.source", res.source), |
| 479 | attribute.Int64("duckgres.bytes", size), |
| 480 | ) |
| 481 | slog.Info("Served.", "source", res.source, "url", r.URL.String(), "range", rangeHeader, "bytes", size) |
| 482 | p.serveStream(w, reader, size, rangeHeader, res.contentType) |
| 483 | } |
| 484 | |
| 485 | // fetchDedup resolves a local miss to on-disk bytes with as little origin |
| 486 | // traffic as the cluster state allows. On success the body has been committed |
| 487 | // to local disk under cacheKey; the caller serves it by streaming from the |
| 488 | // file. Nothing here holds the body in memory. |
| 489 | func (p *CacheProxy) fetchDedup(cacheKey string, r *http.Request, rangeHeader string) (fetchResult, error) { |
| 490 | return p.flights.Do(cacheKey, func() (fetchResult, error) { |
| 491 | if p.peers != nil { |
| 492 | if p.peers.lookupMode == peerLookupSummary { |
| 493 | // One logical lookup uses local Bloom tests first. Peers without a |
| 494 | // valid summary remain on the legacy probe path only while this |
| 495 | // proxy is converging after startup or membership changes. |
| 496 | peerFetchesTotal.Inc() |
| 497 | positive, uncovered := p.peers.SummaryLookup(cacheKey, time.Now()) |
| 498 | if holder, flight, ok, _ := p.peers.LocateSummaryKey(r.Context(), cacheKey, positive, uncovered, p.peers.peerMaxProbes); ok { |
| 499 | if res, ok := p.fetchFromPeer(holder, flight, cacheKey, r); ok { |
| 500 | summaryConfirmedGetsTotal.WithLabelValues("success").Inc() |
| 501 | return res, nil |
| 502 | } |
| 503 | summaryConfirmedGetsTotal.WithLabelValues("miss_or_error").Inc() |
| 504 | } |
| 505 | } else if holder, flight, ok := p.peers.LocateKey(r.Context(), cacheKey); ok { |
| 506 | res, ok := p.fetchFromPeer(holder, flight, cacheKey, r) |
| 507 | if ok { |
| 508 | return res, nil |
| 509 | } |
| 510 | } |
| 511 | } |
| 512 | originFetchInFlight.Inc() |
| 513 | defer originFetchInFlight.Dec() |
| 514 | size, ct, err := p.fetchOrigin(cacheKey, r) |
| 515 | if err != nil { |
| 516 | originFetchesTotal.WithLabelValues(originFetchOutcome(err)).Inc() |
| 517 | return fetchResult{}, err |
| 518 | } |