fetchOrigin forwards the request verbatim (headers, Host, signature) to the real origin and streams a successful body straight to the on-disk cache, returning the stored size and Content-Type. The SigV4 signature remains valid because the URL and Host header are unchanged. Streaming to disk (rather
(cacheKey string, r *http.Request)
| 393 | } |
| 394 | if r.Header.Get(cachePassthroughHeader) == "true" { |
| 395 | p.forward(w, r) |
| 396 | return |
| 397 | } |
| 398 | |
| 399 | // Only cache URLs that look like DuckLake bucket traffic. Anything else |
| 400 | // (non-bucket HTTP) is a passthrough. |
| 401 | if !p.shouldCache(r) { |
| 402 | p.forward(w, r) |
| 403 | return |
| 404 | } |
| 405 | |
| 406 | rangeHeader := r.Header.Get("Range") |
| 407 | // The tenant scope (SigV4 access key ID) is part of every cache key, so |
| 408 | // an entry written for one tenant's credentials is never served to a |
| 409 | // request signed by another tenant. |
| 410 | scope := TenantScope(r) |
| 411 | |
| 412 | if p.blockMode && p.serveBlockAligned(w, r, rangeHeader, scope) { |
| 413 | return |
| 414 | } |
| 415 | // Legacy exact-range path (also the fallback for non-absolute ranges). |
| 416 | cacheKey := CacheKey(scope, r.URL.String(), rangeHeader) |
| 417 | |
| 418 | // Requests without a propagated parent still start a standalone trace. |
| 419 | // Thread the cache request span context into origin/peer work below. |
| 420 | ctx, span := proxyTracer.Start(r.Context(), "cache.get", trace.WithAttributes(requestSpanAttrs(r)...)) |
| 421 | defer span.End() |
| 422 | span.SetAttributes(attribute.String("duckgres.cache.key", cacheKey)) |
| 423 | r = r.WithContext(ctx) |
| 424 | |
| 425 | if reader, size, ok := p.store.Open(cacheKey); ok { |
| 426 | cacheBytesServed.WithLabelValues("local").Add(float64(size)) |
| 427 | span.SetAttributes( |
| 428 | attribute.String("duckgres.cache.source", "hit"), |
| 429 | attribute.Bool("duckgres.cache.hit", true), |
| 430 | attribute.Int64("duckgres.bytes", size), |
| 431 | ) |
| 432 | slog.Info("Served.", "source", "hit", "url", r.URL.String(), "range", rangeHeader, "bytes", size) |
| 433 | p.serveStream(w, reader, size, rangeHeader, "") |
| 434 | _ = reader.Close() |
| 435 | return |
| 436 | } |
| 437 | cacheMissesTotal.Inc() |
| 438 | span.SetAttributes(attribute.Bool("duckgres.cache.hit", false)) |
| 439 | |
| 440 | res, err := p.fetchDedup(cacheKey, r, rangeHeader) |
| 441 | if err != nil { |
| 442 | span.SetStatus(codes.Error, err.Error()) |
| 443 | // An origin that responded with a non-2xx (e.g. S3 returning a 400 with |
| 444 | // <Code>ExpiredToken</Code> in an XML envelope) is forwarded back to |
| 445 | // DuckDB verbatim — same status code, same body, same headers minus |
| 446 | // hop-by-hop. This preserves the error class so httpfs can distinguish |
| 447 | // transient (5xx) from terminal (4xx) failures, and gives DuckLake the |
| 448 | // raw S3 error body it knows how to parse. |
| 449 | var oe *originStatusError |
| 450 | if errors.As(err, &oe) { |
| 451 | span.SetAttributes(attribute.Int("http.response.status_code", oe.status)) |
| 452 | slog.Warn("Origin returned non-2xx; forwarding verbatim.", |
no test coverage detected