diff --git a/README.md b/README.md index c8d6fdb7..5fa65076 100644 --- a/README.md +++ b/README.md @@ -93,7 +93,7 @@ When configuring S3 clients, ensure path-style addressing is enabled. See [docs/ | Header | Description | | --------- | ---------------------------------------------------- | -| `X-Cache` | Cache status: `HIT`, `MISS`, `BYPASS`, or `DISABLED` | +| `X-Cache` | Cache decision at response commitment: `HIT`, `MISS`, `BYPASS`, or `DISABLED`; a block-mode `HIT` may include a verified upstream remainder | ### Cache Behavior @@ -101,6 +101,9 @@ When configuring S3 clients, ensure path-style addressing is enabled. See [docs/ - Objects with `Cache-Control: no-store` or `private` are not cached - Among cacheable objects (within `size_threshold`), those at or above `block_size` are cached as fixed-size blocks (block caching, on by default), so a range read fetches only the covering blocks; smaller objects are whole-cached - Set `block_caching_enabled: false` to cache whole objects instead — a range request then triggers a background fetch of the full object (if within threshold) +- Cached range reads are checked against the requested byte count. A committed block GET can recover an unwritten suffix only from an upstream response with the same ETag +- `X-Cache: HIT` identifies a cache-backed response selected at commitment; it does not promise that every byte came from cache or that the transfer completed +- If a committed block GET has a changed-ETag suffix, or a whole-object Range read is short, TAG keeps the sent status and Content-Length, returns a handler error, and may end with only the cached prefix; client write failures stop without an upstream retry - PUT/DELETE operations invalidate the cache entry See [docs/cache-control.md](docs/cache-control.md) for detailed cache control and revalidation documentation. For tuning and sizing block caching in production, see [docs/deploy.md](docs/deploy.md#block-aligned-caching). @@ -141,7 +144,7 @@ For production, TAG ships manifests, Compose files, and guides under [`deploy/`] ### Request Flow 1. **Cache Check**: TAG first checks if the object exists in its embedded cache -2. **Cache Hit**: Returns cached object with `X-Cache: HIT` header +2. **Cache Hit**: Serves using a cache entry with `X-Cache: HIT`; a degraded block stream may append a same-ETag upstream remainder 3. **Cache Miss**: Forwards request to upstream Tigris, caches response, returns with `X-Cache: MISS` ### Request Coalescing diff --git a/cache/cache.go b/cache/cache.go index d95d0cf5..1d5d5dcf 100644 --- a/cache/cache.go +++ b/cache/cache.go @@ -471,7 +471,8 @@ func (c *Cache) IsBlockLocal(bucket, key, etag string, blockSize, blockIdx int64 // Uses ocache's GetRangeStream for efficient partial reads from disk. // start and end are inclusive byte positions (HTTP Range semantics). // Pass the meta's ETag so the range resolves to the exact cached version. -// Returns ErrNotFound if the object is not in cache. +// Returns ErrNotFound if the object is not in cache. A positive short backend read returns an +// error after writing its available bytes to w. func (c *Cache) GetRangeStream(ctx context.Context, bucket, key, etag string, start, end int64, w io.Writer) error { if !c.IsEnabled() { return ErrCacheDisabled @@ -483,7 +484,8 @@ func (c *Cache) GetRangeStream(ctx context.Context, bucket, key, etag string, st // block of a block-mode object to w. blockIdx identifies the block; start and end are // offsets WITHIN the block (0 = first byte of the block), not within the object. Pass the // meta's ETag so the block resolves to the exact cached version. Returns ErrNotFound if the -// block is not in cache. See RFC 0001. +// block is not in cache. A positive short backend read returns an error after writing its +// available bytes to w. See RFC 0001. func (c *Cache) GetBlockRangeStream(ctx context.Context, bucket, key, etag string, blockSize, blockIdx, start, end int64, w io.Writer) error { if !c.IsEnabled() { return ErrCacheDisabled @@ -505,8 +507,9 @@ func (cw *countingWriter) Write(p []byte) (int, error) { } // getRangeStreamByKey streams an inclusive byte range [start,end] of the blob at cacheKey to -// w, mapping ocache's not-found to ErrNotFound and handling ocache's read-byte-0 quirk. -// bucket/key are used for logging only. +// w, mapping ocache's not-found to ErrNotFound, handling ocache's read-byte-0 quirk, and +// rejecting successful reads that do not deliver the requested range in full. bucket/key are +// used for logging only. func (c *Cache) getRangeStreamByKey(ctx context.Context, cacheKey, bucket, key string, start, end int64, w io.Writer) error { // Handle ocache quirk: reading byte 0 alone requires reading 2 bytes // and discarding the last byte @@ -562,6 +565,13 @@ func (c *Cache) getRangeStreamByKey(ctx context.Context, cacheKey, bucket, key s log.Debug().Str("bucket", bucket).Str("key", key).Msg("Cache miss (range)") return ErrNotFound } + want := end - start + 1 + if cw.written < want { + return fmt.Errorf("cache range short read: got %d of %d bytes: %w", cw.written, want, io.ErrUnexpectedEOF) + } + if cw.written > want { + return fmt.Errorf("cache range over-read: got %d bytes, want %d", cw.written, want) + } c.recordServeLocality(cacheKey) log.Debug(). diff --git a/docs/cache-control.md b/docs/cache-control.md index 7bb2dfea..3c82aa02 100644 --- a/docs/cache-control.md +++ b/docs/cache-control.md @@ -6,12 +6,14 @@ TAG supports RFC 7234-compliant cache revalidation. Clients can control caching | Value | Meaning | | ------------- | -------------------------------------------------------------------------------- | -| `HIT` | Served from cache (includes revalidation that confirmed the object is unchanged) | +| `HIT` | A cache entry was selected to serve the response (including a revalidation that confirmed the object is unchanged) | | `MISS` | Not in cache, fetched from upstream and now cached | | `REVALIDATED` | Revalidated with upstream, object changed, new content returned | | `BYPASS` | Cache bypassed entirely (client requested `no-store`) | | `DISABLED` | Caching is disabled server-side (`TAG_CACHE_DISABLED=true`) | +In block mode, a committed stream can recover missing bytes from an upstream range only after validating the cached ETag and requested range. `HIT` is committed before the body streams, so it describes the cache-backed request, not each byte's source or whether the transfer completed. See [S3 compatibility](s3-compatibility.md#caching-behavior) for incomplete cache-range behavior. + ## Force Revalidation Send `Cache-Control: no-cache` or `Cache-Control: max-age=0` to force TAG to check with upstream before serving a cached object. TAG sends a conditional request using the cached ETag. If the object hasn't changed, upstream returns 304 and TAG serves from cache (`X-Cache: HIT`). If changed, TAG streams the new content (`X-Cache: REVALIDATED`). diff --git a/docs/metrics.md b/docs/metrics.md index 3d563f7a..63eeef38 100644 --- a/docs/metrics.md +++ b/docs/metrics.md @@ -28,7 +28,7 @@ Total number of requests processed by TAG. | `operation` | S3 operation: `GetObject`, `PutObject`, `DeleteObject`, `HeadObject` | | `status` | Result: `success`, `error`, `auth_error`, `range_not_satisfiable` | | `mode` | The process's operating mode (`transparent`, `signing`, `tiered`) — one constant value per instance, for fleet-wide splits | -| `source` | Where the response was produced: `local` (TAG's own store or knowledge — cache hits, revalidated-304 serves, authoritative misses, auth errors) or `upstream` (proxied) | +| `source` | Request handling path, not per-byte provenance: `local` for TAG's cache/local-answer path (including verified block-suffix recovery), or `upstream` for a directly proxied response | **Example queries:** @@ -39,7 +39,7 @@ rate(tag_requests_total[5m]) # Error rate sum(rate(tag_requests_total{status="error"}[5m])) / sum(rate(tag_requests_total[5m])) -# Share of traffic answered without touching upstream (per mode) +# Share of requests handled by TAG's local/cache path (per mode; may include upstream recovery) sum(rate(tag_requests_total{source="local"}[5m])) by (mode) / sum(rate(tag_requests_total[5m])) by (mode) # GetObject success rate @@ -57,7 +57,7 @@ Request duration in seconds. | ----------- | ------------ | | `operation` | S3 operation | | `mode` | Operating mode (constant per instance) | -| `source` | `local` or `upstream` — where the response was produced | +| `source` | `local` or `upstream` — request handling path, not every byte's origin | **Buckets:** 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 1.5, 2, 2.5, 3, 4, 5, 7.5, 10 (the default Prometheus buckets plus extra tail resolution at 1.5, 2, 3, 4 and 7.5 seconds) @@ -80,9 +80,12 @@ rate(tag_request_duration_seconds_sum[5m]) / rate(tag_request_duration_seconds_c **Type:** Counter -Total number of cache hits — every request served from cache, including -range-from-cache hits and conditional 304 (Not Modified) responses. It is -recorded in lockstep with the `X-Cache: HIT` response header. +Total number of requests whose cache status was committed as `X-Cache: HIT`, +including range-from-cache hits and conditional 304 (Not Modified) responses. It +is recorded in lockstep with that header. This is a request-level cache decision, +not a per-byte source or successful-transfer count: a degraded block response may +append a verified upstream suffix or later finish with an error, which is recorded +separately in `tag_requests_total`. `tag_cache_hits_total` and `tag_cache_misses_total` count only `HIT` and `MISS`. `REVALIDATED` responses (object changed on upstream) are neither — they are tracked diff --git a/docs/rfcs/0001-block-aligned-caching.md b/docs/rfcs/0001-block-aligned-caching.md index 876c74f5..3a42e4ca 100644 --- a/docs/rfcs/0001-block-aligned-caching.md +++ b/docs/rfcs/0001-block-aligned-caching.md @@ -119,7 +119,7 @@ Rare for a range-read-dominated workload but must be correct — and fast, since So a block-eligible object is **always** block-mode; a full read never produces a competing whole-body entry. -**Bounded degraded serves.** Because `blocks_complete` is a hint and blocks evict independently of the meta, a committed serve (headers already sent) recovers at most `maxInlineFetchesPerServe` (2) absent blocks via individual aligned upstream fetches. Past that cap, on a populate-budget decline, or on a transient cache-read failure, the committed response is completed **byte-exact from a single uncached upstream remainder stream** — validated against the entry's ETag and the exact remaining byte range — instead of truncating, and a cap-tripped (mass-evicted) entry is invalidated so the next GET re-establishes it with one streaming re-split. Only three things still truncate a committed body: a stale signal (a different object version must never be mixed into a committed response), a client write failure (the client is broken, not the cache — tracked explicitly so it is never "salvaged" with a wasted upstream round trip), and a failed remainder stream. +**Bounded degraded serves.** Because `blocks_complete` is a hint and blocks evict independently of the meta, a committed serve (headers already sent) recovers at most `maxInlineFetchesPerServe` (2) absent blocks via individual aligned upstream fetches. Past that cap, on a populate-budget decline, or on a transient or short cache-read failure, the committed response is completed **byte-exact from a single uncached upstream remainder stream** — validated against the entry's ETag and the exact remaining byte range — instead of truncating, and a cap-tripped (mass-evicted) entry is invalidated so the next GET re-establishes it with one streaming re-split. Only three things still truncate a committed body: a stale signal (a different object version must never be mixed into a committed response), a client write failure (the client is broken, not the cache — tracked explicitly so it is never "salvaged" with a wasted upstream round trip), and a failed remainder stream. The request-level `X-Cache: HIT` is committed before streaming; it identifies the cache-backed response selection, not every byte's source or successful completion. A client-forced revalidation (`Cache-Control: no-cache`) of a block-mode entry likewise invalidates it and lets the miss path re-establish the current version. diff --git a/docs/s3-compatibility.md b/docs/s3-compatibility.md index 4ec6c94d..0b172f4f 100644 --- a/docs/s3-compatibility.md +++ b/docs/s3-compatibility.md @@ -71,7 +71,7 @@ TAG adds a caching layer for read operations. The `X-Cache` response header indi | Value | Meaning | | ---------- | -------------------------------------------------------- | -| `HIT` | Served entirely from cache | +| `HIT` | A cache entry was selected to serve the request; block streams may also recover a verified upstream remainder | | `MISS` | Fetched from Tigris and cached | | `BYPASS` | Caching skipped (e.g., conditional request returned 304) | | `DISABLED` | Caching is turned off | @@ -80,6 +80,7 @@ Key caching behaviors: - **Request coalescing** — Multiple concurrent GETs for the same object are coalesced into a single upstream fetch using a broadcast/subscriber pattern. - **Range requests** — If the full object is cached, range requests are served from cache. On a range cache miss, the range is served directly from Tigris while a background fetch populates the cache with the full object. +- **Incomplete cache ranges** — TAG checks that a cached range supplies the requested byte count. A committed block GET can recover an unwritten suffix only when the upstream ETag and range match the cached object. `X-Cache: HIT` is committed before the stream; it identifies the cache-backed request, not per-byte provenance or successful completion. If the ETag changes, TAG rejects the suffix without appending it; the already-sent 200 status and Content-Length remain, so the client can receive only the cached prefix while the handler reports an error. A whole-object cached Range response commits after its first byte and cannot be restarted; a later short read returns a handler error, and the 206 body may end before Content-Length. Client write failures are not retried upstream. - **Conditional requests** — `If-None-Match` and `If-Modified-Since` are evaluated against cached metadata and can return 304 without hitting Tigris. - **Write-through invalidation** — PUT, DELETE, CopyObject, and DeleteObjects invalidate the cache _before_ forwarding to Tigris to prevent stale reads. - **Tombstone protection** — A short-lived tombstone is written on DELETE to prevent in-flight background cache writes from resurrecting deleted objects. diff --git a/metrics/metrics.go b/metrics/metrics.go index 0670551c..221cda06 100644 --- a/metrics/metrics.go +++ b/metrics/metrics.go @@ -458,10 +458,11 @@ var ( ) // RecordRequest records a request with its duration and status. -// Source label values for RecordRequest: where the response was produced. +// Source label values identify the request path that owns response accounting, +// not the origin of every byte in a streamed response. const ( - SourceLocal = "local" // answered from TAG's own store or knowledge - SourceUpstream = "upstream" // proxied from upstream + SourceLocal = "local" // handled through TAG's local/cache path, including verified block-suffix recovery + SourceUpstream = "upstream" // proxied directly from upstream ) // processMode is the operating mode stamped on every request metric. One @@ -482,6 +483,17 @@ func RecordRequest(operation, status, source string, durationSeconds float64) { RequestDuration.WithLabelValues(operation, processMode, source).Observe(durationSeconds) } +// RecordRequestResult maps an operation result to the shared success/error status labels. +// Callers that own specialized outcomes such as authentication or range validation use +// RecordRequest with their explicit status instead. +func RecordRequestResult(operation, source string, err error, durationSeconds float64) { + status := "success" + if err != nil { + status = "error" + } + RecordRequest(operation, status, source, durationSeconds) +} + // RecordCacheHit records a cache hit. func RecordCacheHit() { CacheHits.Inc() diff --git a/proxy/block_cache.go b/proxy/block_cache.go index 0581d568..efa75ea7 100644 --- a/proxy/block_cache.go +++ b/proxy/block_cache.go @@ -206,10 +206,11 @@ func (s *Service) serveRangeFromBlockCache( w.WriteHeader(http.StatusPartialContent) if _, berr := s.streamBlockRange(ctx, w, bucket, key, accessKey, secretKey, meta, rng.start, rng.end); berr != nil { + metrics.RecordRequestResult("GetObject", metrics.SourceLocal, berr, time.Since(startTime).Seconds()) return true, berr } metrics.RecordRangeFromCacheHit() - metrics.RecordRequest("GetObject", "success", metrics.SourceLocal, time.Since(startTime).Seconds()) + metrics.RecordRequestResult("GetObject", metrics.SourceLocal, nil, time.Since(startTime).Seconds()) return true, nil } @@ -371,10 +372,11 @@ func (s *Service) serveAssembledRange( metrics.BytesTransferred.WithLabelValues("out").Add(float64(n)) } if werr != nil { + metrics.RecordRequestResult("GetObject", metrics.SourceLocal, werr, time.Since(startTime).Seconds()) return true, werr } metrics.RecordRangeFromCacheHit() - metrics.RecordRequest("GetObject", "success", metrics.SourceLocal, time.Since(startTime).Seconds()) + metrics.RecordRequestResult("GetObject", metrics.SourceLocal, nil, time.Since(startTime).Seconds()) // A parquet reader's trailer probe is a few bytes, so it is served here rather // than by streamBlockRange — this, not the streaming path, is where the footer // signal actually arrives for a reader opening a file. @@ -872,9 +874,10 @@ func (s *Service) serveFullObjectFromBlockCache( w.WriteHeader(meta.StatusCode) if _, berr := s.streamBlockRange(ctx, w, bucket, key, accessKey, secretKey, meta, 0, meta.ContentLength-1); berr != nil { + metrics.RecordRequestResult("GetObject", metrics.SourceLocal, berr, time.Since(startTime).Seconds()) return true, berr } - metrics.RecordRequest("GetObject", "success", metrics.SourceLocal, time.Since(startTime).Seconds()) + metrics.RecordRequestResult("GetObject", metrics.SourceLocal, nil, time.Since(startTime).Seconds()) return true, nil } @@ -897,9 +900,9 @@ func (s *Service) remainingMetaTTL(meta *cache.CachedObjectMeta) int { // commit: if it is absent (evicted since populate) it is fetched pre-commit, and if that fails // the response is untouched — (false, nil) lets the caller fall through to the miss path, and a // definitive stale signal invalidates the entry first. After commit, the remaining blocks -// stream with an inline fetch recovering any absent one; only a fetch failure there truncates -// (headers are already sent), which is the same terminal behavior the probe path has for a -// block evicted between probe and stream. Hit/miss metrics count inline-fetched blocks as +// stream with inline fetch recovering any absent one; a short cache read or failed inline fetch +// can use the shared same-ETag remainder path. If recovery fails, the caller can only return the +// post-commit error — it cannot fall through. Hit/miss metrics count inline-fetched blocks as // misses, recorded once the serve outcome is known. func (s *Service) serveCompleteFromBlocks( ctx context.Context, @@ -962,9 +965,10 @@ func (s *Service) serveCompleteFromBlocks( } } if berr != nil { + metrics.RecordRequestResult("GetObject", metrics.SourceLocal, berr, time.Since(startTime).Seconds()) return true, berr } - metrics.RecordRequest("GetObject", "success", metrics.SourceLocal, time.Since(startTime).Seconds()) + metrics.RecordRequestResult("GetObject", metrics.SourceLocal, nil, time.Since(startTime).Seconds()) return true, nil } diff --git a/proxy/block_cache_integration_test.go b/proxy/block_cache_integration_test.go index 0928c678..484b00b5 100644 --- a/proxy/block_cache_integration_test.go +++ b/proxy/block_cache_integration_test.go @@ -1,6 +1,7 @@ package proxy import ( + "bytes" "context" "errors" "fmt" @@ -14,6 +15,7 @@ import ( "github.com/prometheus/client_golang/prometheus/testutil" cacheclient "github.com/tigrisdata/ocache/client" + "github.com/tigrisdata/tag/auth" "github.com/tigrisdata/tag/cache" "github.com/tigrisdata/tag/config" "github.com/tigrisdata/tag/metrics" @@ -157,6 +159,67 @@ func newBlockServiceWithBudget(t *testing.T, mock *blockMockForwarder, blockSize return NewService(mock, c, cfg), c } +type rangeTestUpstream struct { + requests atomic.Int32 + rangeCalls atomic.Int32 + lastRange atomic.Value + invalid atomic.Bool +} + +func newRangeTestUpstream(t *testing.T, bucket, key string, object []byte, etag string) (*httptest.Server, *rangeTestUpstream) { + t.Helper() + state := &rangeTestUpstream{} + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + state.requests.Add(1) + if r.Method != http.MethodGet || r.URL.Path != "/"+bucket+"/"+key { + state.invalid.Store(true) + http.Error(w, "unexpected upstream request", http.StatusBadRequest) + return + } + if rangeHeader := r.Header.Get("Range"); rangeHeader != "" { + state.rangeCalls.Add(1) + state.lastRange.Store(rangeHeader) + var start, end int64 + if count, err := fmt.Sscanf(rangeHeader, "bytes=%d-%d", &start, &end); err != nil || count != 2 || start < 0 || end < start || end >= int64(len(object)) { + state.invalid.Store(true) + http.Error(w, "invalid range", http.StatusRequestedRangeNotSatisfiable) + return + } + body := object[start : end+1] + w.Header().Set("ETag", etag) + w.Header().Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", start, end, len(object))) + w.Header().Set("Content-Length", fmt.Sprintf("%d", len(body))) + w.WriteHeader(http.StatusPartialContent) + if _, err := w.Write(body); err != nil { + state.invalid.Store(true) + } + return + } + w.Header().Set("ETag", etag) + w.Header().Set("Content-Length", fmt.Sprintf("%d", len(object))) + w.WriteHeader(http.StatusOK) + if _, err := w.Write(object); err != nil { + state.invalid.Store(true) + } + })) + t.Cleanup(server.Close) + return server, state +} + +func newHTTPRangeService(t *testing.T, endpoint string, blockCaching bool) (*Service, *cache.Cache) { + t.Helper() + cfg := config.NewDefault() + cfg.Upstream.Endpoint = endpoint + cfg.Cache.SetBlockCachingEnabled(blockCaching) + if blockCaching { + cfg.Cache.BlockSize = 4 + cfg.Cache.SizeThreshold = 1 << 20 + } + store := cache.NewCacheWithClient(cacheclient.NewMemoryCache(), &cfg.Cache) + forwarder := NewForwarder(nil, endpoint, cfg.Upstream.Region, cfg.Upstream.MaxIdleConnsPerHost, auth.NewProxySigner("test-proxy-access", "test-proxy-secret"), nil) + return NewService(forwarder, store, cfg), store +} + // Cold range miss forwards from upstream, then block-populates in the background; a // subsequent range read within a populated block is served from cache with no re-fetch. func TestBlockCache_ColdMissPopulatesThenServesFromBlocks(t *testing.T) { @@ -1605,6 +1668,213 @@ func TestBlockCache_CompleteEntryFetchDeclineSalvagesViaRemainder(t *testing.T) } } +func seedTwoBlockCompleteEntryWithShortTail(t *testing.T, c *cache.Cache) { + t.Helper() + ctx := context.Background() + const blockSize = int64(4) + for idx, data := range [][]byte{[]byte("ABCD"), []byte("EF")} { + if err := c.PutBlockStream(ctx, wowBucket, wowKey, `"v1"`, blockSize, int64(idx), bytes.NewReader(data), 60); err != nil { + t.Fatalf("seed block %d: %v", idx, err) + } + } + meta := &cache.CachedObjectMeta{ + Bucket: wowBucket, + Key: wowKey, + ETag: `"v1"`, + ContentLength: 8, + StatusCode: http.StatusOK, + BlockSize: blockSize, + BlocksComplete: true, + } + if wrote, err := c.PutMetaIfVersion(ctx, wowBucket, wowKey, meta, 60, cache.VersionAny); err != nil || !wrote { + t.Fatalf("seed BlocksComplete metadata = (wrote=%t, err=%v)", wrote, err) + } +} + +// A positive short nil-error cache read is not a complete block hit: after the +// first block commits, the stream must request exactly the unwritten suffix and +// complete it only when the upstream ETag still matches. X-Cache is already +// committed as a request-level cache decision before this recovery. +func TestBlockCache_ShortPositiveReadRecoversSameETagRemainder(t *testing.T) { + object := []byte("ABCDEFGH") + server, upstream := newRangeTestUpstream(t, wowBucket, wowKey, object, `"v1"`) + svc, c := newHTTPRangeService(t, server.URL, true) + seedTwoBlockCompleteEntryWithShortTail(t, c) + + beforeSuccesses := requestCount(t, "success") + beforeErrors := requestCount(t, "error") + w := httptest.NewRecorder() + err := svc.HandleGetObject(w, fullGet(wowBucket, wowKey)) + if err != nil { + t.Fatalf("HandleGetObject() error = %v, want same-ETag suffix recovery", err) + } + if w.Code != http.StatusOK || w.Body.String() != string(object) { + t.Errorf("response = (status=%d, body=%q), want (200, %q)", w.Code, w.Body.String(), object) + } + if got := w.Header().Get("Content-Length"); got != "8" { + t.Errorf("Content-Length = %q, want 8", got) + } + if got := w.Result().Header.Get(XCacheHeader); got != XCacheHit { + t.Errorf("committed %s = %q, want %q", XCacheHeader, got, XCacheHit) + } + if got := upstream.rangeCalls.Load(); got != 1 { + t.Errorf("same-ETag suffix HTTP requests = %d, want exactly 1", got) + } else if got := upstream.lastRange.Load(); got != "bytes=6-7" { + t.Errorf("same-ETag upstream Range = %v, want bytes=6-7", got) + } + if got := upstream.requests.Load(); got != 1 || upstream.invalid.Load() { + t.Errorf("upstream requests=%d invalid=%t, want one valid suffix request", got, upstream.invalid.Load()) + } + if got := requestCount(t, "success"); got != beforeSuccesses+1 { + t.Errorf("success requests = %v, want %v", got, beforeSuccesses+1) + } + if got := requestCount(t, "error"); got != beforeErrors { + t.Errorf("error requests = %v, want %v", got, beforeErrors) + } +} + +// A short cache read may be followed only by bytes verified under the cached +// ETag. A changed upstream version is surfaced as an error and its body is never +// appended to the already-committed response. +func TestBlockCache_ShortPositiveReadRejectsChangedETagRemainder(t *testing.T) { + server, upstream := newRangeTestUpstream(t, wowBucket, wowKey, []byte("abcdefgh"), `"v2"`) + svc, c := newHTTPRangeService(t, server.URL, true) + seedTwoBlockCompleteEntryWithShortTail(t, c) + + beforeSuccesses := requestCount(t, "success") + beforeErrors := requestCount(t, "error") + w := httptest.NewRecorder() + err := svc.HandleGetObject(w, fullGet(wowBucket, wowKey)) + if !errors.Is(err, errBlockETagMismatch) { + t.Errorf("HandleGetObject() error = %v, want changed-ETag error", err) + } + if w.Code != http.StatusOK || w.Body.String() != "ABCDEF" { + t.Errorf("response = (status=%d, body=%q), want only cached prefix ABCDEF", w.Code, w.Body.String()) + } + if got := w.Header().Get("Content-Length"); got != "8" { + t.Errorf("Content-Length = %q, want 8", got) + } + if got := upstream.rangeCalls.Load(); got != 1 { + t.Errorf("changed-ETag suffix HTTP requests = %d, want exactly 1", got) + } else if got := upstream.lastRange.Load(); got != "bytes=6-7" { + t.Errorf("changed-ETag upstream Range = %v, want bytes=6-7", got) + } + if got := upstream.requests.Load(); got != 1 || upstream.invalid.Load() { + t.Errorf("upstream requests=%d invalid=%t, want one valid suffix request", got, upstream.invalid.Load()) + } + if got := requestCount(t, "success"); got != beforeSuccesses { + t.Errorf("success requests = %v, want unchanged %v", got, beforeSuccesses) + } + if got := requestCount(t, "error"); got != beforeErrors+1 { + t.Errorf("error requests = %v, want %v", got, beforeErrors+1) + } +} + +// Whole-body Range GETs use the same cache range wrapper. If the stored body is +// shorter than metadata says, the committed 206 must return an error and must +// not be accounted as a complete cache hit. +func TestWholeBodyRange_ShortPositiveReadReturnsError(t *testing.T) { + server, upstream := newRangeTestUpstream(t, wowBucket, wowKey, []byte("ABCDEFGH"), `"v1"`) + svc, c := newHTTPRangeService(t, server.URL, false) + ctx := context.Background() + meta := &cache.CachedObjectMeta{ + Bucket: wowBucket, + Key: wowKey, + ETag: `"v1"`, + ContentLength: 8, + StatusCode: http.StatusOK, + } + if err := c.PutBodyStream(ctx, wowBucket, wowKey, meta.ETag, bytes.NewReader([]byte("ABCD")), 60); err != nil { + t.Fatalf("seed short body: %v", err) + } + if wrote, err := c.PutMetaIfVersion(ctx, wowBucket, wowKey, meta, 60, cache.VersionAny); err != nil || !wrote { + t.Fatalf("seed whole-object metadata = (wrote=%t, err=%v)", wrote, err) + } + + beforeSuccesses := requestCount(t, "success") + beforeErrors := requestCount(t, "error") + w := httptest.NewRecorder() + err := svc.HandleGetObject(w, blockGet(wowBucket, wowKey, "bytes=0-7")) + if !errors.Is(err, io.ErrUnexpectedEOF) { + t.Errorf("HandleGetObject() error = %v, want short-range error", err) + } + if w.Code != http.StatusPartialContent || w.Body.String() != "ABCD" { + t.Errorf("response = (status=%d, body=%q), want committed 206 with prefix ABCD", w.Code, w.Body.String()) + } + if got := w.Header().Get("Content-Length"); got != "8" { + t.Errorf("Content-Length = %q, want 8", got) + } + if got := upstream.requests.Load(); got != 0 { + t.Errorf("upstream HTTP requests = %d, want 0 after response commitment", got) + } + if got := requestCount(t, "success"); got != beforeSuccesses { + t.Errorf("success requests = %v, want unchanged %v", got, beforeSuccesses) + } + if got := requestCount(t, "error"); got != beforeErrors+1 { + t.Errorf("error requests = %v, want %v", got, beforeErrors+1) + } +} + +type failFirstResponseWrite struct { + http.ResponseWriter + writes int +} + +func (w *failFirstResponseWrite) Write(p []byte) (int, error) { + w.writes++ + if w.writes == 1 { + return 0, errors.New("client write failed") + } + return w.ResponseWriter.Write(p) +} + +// A client error on the committed first byte of a whole-object Range GET must +// stop the cache copy instead of dropping that byte and counting later writes as +// a successful response. +func TestWholeBodyRange_FirstClientWriteFailureIsError(t *testing.T) { + body := []byte("ABCDEFGH") + server, upstream := newRangeTestUpstream(t, wowBucket, wowKey, body, `"v1"`) + svc, c := newHTTPRangeService(t, server.URL, false) + ctx := context.Background() + meta := &cache.CachedObjectMeta{ + Bucket: wowBucket, + Key: wowKey, + ETag: `"v1"`, + ContentLength: int64(len(body)), + StatusCode: http.StatusOK, + } + if err := c.PutBodyStream(ctx, wowBucket, wowKey, meta.ETag, bytes.NewReader(body), 60); err != nil { + t.Fatalf("seed body: %v", err) + } + if wrote, err := c.PutMetaIfVersion(ctx, wowBucket, wowKey, meta, 60, cache.VersionAny); err != nil || !wrote { + t.Fatalf("seed whole-object metadata = (wrote=%t, err=%v)", wrote, err) + } + + beforeSuccesses := requestCount(t, "success") + beforeErrors := requestCount(t, "error") + recorder := httptest.NewRecorder() + w := &failFirstResponseWrite{ResponseWriter: recorder} + err := svc.HandleGetObject(w, blockGet(wowBucket, wowKey, "bytes=0-7")) + if err == nil { + t.Fatal("HandleGetObject() returned nil after the client write failed") + } + if w.writes != 1 || recorder.Code != http.StatusPartialContent || recorder.Body.Len() != 0 { + t.Errorf("response writes=%d status=%d body=%q, want one failed write after committing an empty 206", w.writes, recorder.Code, recorder.Body.Bytes()) + } + if got := recorder.Header().Get("Content-Length"); got != "8" { + t.Errorf("Content-Length = %q, want 8", got) + } + if got := upstream.requests.Load(); got != 0 { + t.Errorf("client write failure triggered %d upstream HTTP requests, want 0 after commitment", got) + } + if got := requestCount(t, "success"); got != beforeSuccesses { + t.Errorf("success requests = %v, want unchanged %v", got, beforeSuccesses) + } + if got := requestCount(t, "error"); got != beforeErrors+1 { + t.Errorf("error requests = %v, want %v", got, beforeErrors+1) + } +} + // A BlocksComplete meta surviving mass block eviction must not turn one full GET into N // serial upstream fetches: after maxInlineFetchesPerServe recoveries the serve switches to a // single uncached upstream remainder stream, and the degenerate entry is invalidated so the @@ -1824,11 +2094,26 @@ func TestBlockCache_SequentialClientWriteFailureDoesNotFetchRemainder(t *testing } before := mock.blockGets.Load() - fw := &failAfterWriter{ResponseWriter: httptest.NewRecorder(), remaining: 9} - _ = svc.HandleGetObject(fw, fullGet(wowBucket, wowKey)) + beforeErrors := requestCount(t, "error") + beforeSuccesses := requestCount(t, "success") + recorder := httptest.NewRecorder() + fw := &failAfterWriter{ResponseWriter: recorder, remaining: 9} + err := svc.HandleGetObject(fw, fullGet(wowBucket, wowKey)) + if err == nil { + t.Fatal("HandleGetObject() returned nil after the client write failed") + } + if recorder.Code != http.StatusOK || !bytes.Equal(recorder.Body.Bytes(), object[:9]) { + t.Errorf("client response = (status=%d, body=%q), want status 200 with the accepted 9-byte prefix", recorder.Code, recorder.Body.Bytes()) + } if got := mock.blockGets.Load() - before; got != 0 { t.Errorf("client write failure triggered %d upstream requests, want 0 (no remainder salvage)", got) } + if got := requestCount(t, "success"); got != beforeSuccesses { + t.Errorf("success requests = %v, want unchanged %v", got, beforeSuccesses) + } + if got := requestCount(t, "error"); got != beforeErrors+1 { + t.Errorf("error requests = %v, want %v", got, beforeErrors+1) + } } // clientForwards reports how many client-range forwards have hit upstream. diff --git a/proxy/get_object.go b/proxy/get_object.go index ed3c1731..9f6108bc 100644 --- a/proxy/get_object.go +++ b/proxy/get_object.go @@ -725,10 +725,13 @@ func (s *Service) serveRangeFromCache( writeCacheStatus(w, XCacheHit) w.WriteHeader(http.StatusPartialContent) - // Stream range from cache using counting writer to track actual bytes + // Stream range from cache using counting writer to track actual bytes. Stop immediately + // on a client write failure; continuing could drop the first byte and still look successful. cw := &countingWriter{w: w} - cw.Write(firstByte[:n]) - _, copyErr := io.Copy(cw, pr) + _, copyErr := cw.Write(firstByte[:n]) + if copyErr == nil { + _, copyErr = io.Copy(cw, pr) + } pr.Close() // Track bytes out (even on error, some bytes may have been written) @@ -737,15 +740,16 @@ func (s *Service) serveRangeFromCache( } if copyErr != nil { + metrics.RecordRequestResult("GetObject", metrics.SourceLocal, copyErr, time.Since(startTime).Seconds()) log.Warn().Err(copyErr).Str("bucket", bucket).Str("key", key). Int64("start", rng.start).Int64("end", rng.end). Msg("Failed to stream range from cache") - // Headers already sent, can't return error to client + // Headers already sent, can't replace the partial response with an error response. return true, copyErr } metrics.RecordRangeFromCacheHit() - metrics.RecordRequest("GetObject", "success", metrics.SourceLocal, time.Since(startTime).Seconds()) + metrics.RecordRequestResult("GetObject", metrics.SourceLocal, nil, time.Since(startTime).Seconds()) return true, nil } diff --git a/proxy/revalidation.go b/proxy/revalidation.go index 245baed6..32f6ba85 100644 --- a/proxy/revalidation.go +++ b/proxy/revalidation.go @@ -356,7 +356,6 @@ func (s *Service) serveFromCache( // pipe and a second copy. cw := &lazyCommitWriter{w: w, meta: meta} bodyErr := s.cache.GetBodyStream(ctx, bucket, key, meta.BodyDiscriminator(), cw) - status := "success" if bodyErr != nil { if !cw.committed { return fmt.Errorf("cache body unavailable: %w", bodyErr) @@ -367,14 +366,13 @@ func (s *Service) serveFromCache( // however, so record the request as an error. log.Warn().Err(bodyErr).Str("bucket", bucket).Str("key", key). Msg("Failed to stream cache body after headers committed") - status = "error" } if !cw.committed { return fmt.Errorf("cache body empty for %s/%s", bucket, key) } metrics.BytesTransferred.WithLabelValues("out").Add(float64(cw.written)) - metrics.RecordRequest("GetObject", status, metrics.SourceLocal, time.Since(start).Seconds()) + metrics.RecordRequestResult("GetObject", metrics.SourceLocal, bodyErr, time.Since(start).Seconds()) return nil } diff --git a/proxy/service.go b/proxy/service.go index eb9e9958..19eba16e 100644 --- a/proxy/service.go +++ b/proxy/service.go @@ -32,11 +32,13 @@ const ( // lockstep. Call it once per response, where the status is committed — never pair // it with a separate RecordCacheHit/RecordCacheMiss for the same response. // -// HIT increments hits; MISS increments misses. REVALIDATED, BYPASS, and DISABLED -// set the header but are neither hits nor misses: REVALIDATED (object changed on -// upstream) is tracked by the tag_revalidations_* metrics, and a bypassed/disabled -// request made no cache decision. Range-from-cache hits additionally call -// RecordRangeFromCacheHit for the specialized counter. +// HIT identifies a response committed using a cache entry, not per-byte provenance +// or transfer completion: a block stream can recover a suffix from upstream after +// commitment or later return an error. HIT increments hits; MISS increments misses. +// REVALIDATED, BYPASS, and DISABLED set the header but are neither hits nor misses: +// REVALIDATED (object changed on upstream) is tracked by the tag_revalidations_* metrics, +// and a bypassed/disabled request made no cache decision. Range-from-cache hits additionally +// call RecordRangeFromCacheHit for the specialized counter. func writeCacheStatus(w http.ResponseWriter, status string) { w.Header().Set(XCacheHeader, status) switch status {