diff --git a/Dockerfile b/Dockerfile index 8afc3dd..e82d2d0 100644 --- a/Dockerfile +++ b/Dockerfile @@ -32,10 +32,13 @@ RUN script/bootstrap --cgo COPY . . # Without -v first; on a failure, again with -v for the details, and -# the step fails even if the second run passes. -RUN go test -count=1 -timeout 90s -race -cover ./... || \ +# the step fails even if the second run passes. -parallel 4: by default +# a package runs as many of its tests at once as the host has CPUs, and +# on a busy host they then wait so long to be scheduled that a test +# that times a request can see it return a second late. +RUN go test -count=1 -timeout 90s -race -parallel 4 -cover ./... || \ { echo "--- Rerunning with -v for details ---"; \ - go test -count=1 -timeout 90s -race -v ./...; exit 1; } + go test -count=1 -timeout 90s -race -parallel 4 -v ./...; exit 1; } # Build stage. Nothing is wanted from the two phases above: these copies # make BuildKit build them first, so this stage runs only when lint and diff --git a/README.md b/README.md index 8b6b032..f6ad430 100644 --- a/README.md +++ b/README.md @@ -83,10 +83,12 @@ which answers 200 whenever pixa is running, in maintenance mode too (see `maintenance_mode`). On SIGTERM or SIGINT pixa stops accepting connections, gives the requests in -progress and the images being processed 5 seconds to finish, and exits: with 0, -or with 1 when images were still being processed after those 5 seconds or -another part of pixa failed to stop. A request not finished by then is cut off. -`docker stop` waits 10 seconds before it kills the container. +progress and the images being processed 5 seconds to finish, then writes to the +database the counts of cache hits, misses, fetches and conversions that requests +could not write by their deadline, and exits: with 0, or with 1 when images were +still being processed after those 5 seconds, some of those counts could not be +written, or another part of pixa failed to stop. A request not finished by then +is cut off. `docker stop` waits 10 seconds before it kills the container. Outside Docker, pixa needs libvips (the image has 8.16) and libheif to run, as it uses libvips through CGO. pixad does not start unless libvips has its JPEG XL diff --git a/TODO.md b/TODO.md index 7bee655..bb24edf 100644 --- a/TODO.md +++ b/TODO.md @@ -30,6 +30,19 @@ P2: security: per-IP rate limiting on the image routes # Completed Steps +- 2026-10-08 a request past its deadline no longer waits for the database to + count it (closes #224). On a busy host, `TestService_Get_ReturnsByItsDeadline` + failed because its request, past its deadline, still waited for its miss count + to be written, a write with no deadline, which in pixad waits for the one + database connection every request shares. Each count write now keeps the + request's deadline but not its cancellation. A count not written by then is + kept in memory, where `Stats` sees it, and one goroutine of the cache writes + those counts, one write at a time, and once more at shutdown before the + database closes. The test phase of the `Dockerfile` also runs at most 4 tests + of a package at once (`go test -parallel 4`), where it ran as many as the host + has CPUs: on a busy host they then wait so long to be scheduled that a test + that times a request can fail. `-p`, how many packages are tested at once, is + unchanged, as capping it also slows compiling. - 2026-10-08 every output format is saved with settings pixa sets on purpose (closes #232): each format has its own govips export, as JPEG XL does, in place of govips' generic `Export`, which sent libvips a zero for some settings diff --git a/internal/handlers/handlers.go b/internal/handlers/handlers.go index 0414026..6405d08 100644 --- a/internal/handlers/handlers.go +++ b/internal/handlers/handlers.go @@ -4,6 +4,7 @@ package handlers import ( "context" "encoding/json" + "errors" "log/slog" "net/http" "time" @@ -71,16 +72,21 @@ func New(lc fx.Lifecycle, params Params) (*Handlers, error) { } lc.Append(fx.Hook{ - //nolint:contextcheck // the eviction loop outlives OnStart; OnStop cancels it + //nolint:contextcheck // the cache's goroutines outlive OnStart; OnStop stops them OnStart: func(_ context.Context) error { return s.initImageService() }, + // The pending counts are written here, before the database's stop + // hook closes it. OnStop: func(ctx context.Context) error { if s.imgCache == nil { return nil } - return s.imgCache.StopEviction(ctx) + return errors.Join( + s.imgCache.StopEviction(ctx), + s.imgCache.StopPendingCountWrites(ctx), + ) }, }) @@ -122,6 +128,9 @@ func (s *Handlers) initImageService() error { // write-pressure passes. No-op when the disk cache is disabled. cache.StartEviction(imgcache.DefaultEvictionInterval) + // Writes the counts requests could not write by their deadline + cache.StartPendingCountWrites() + // Create the fetcher config fetcherCfg := httpfetcher.DefaultConfig() fetcherCfg.AllowHTTP = s.config.AllowHTTP diff --git a/internal/imgcache/cache.go b/internal/imgcache/cache.go index 00589ab..ddccbc5 100644 --- a/internal/imgcache/cache.go +++ b/internal/imgcache/cache.go @@ -11,6 +11,7 @@ import ( "io" "log/slog" "path/filepath" + "sync/atomic" "time" lru "github.com/hashicorp/golang-lru/v2" @@ -79,6 +80,22 @@ type Cache struct { evictionDone chan struct{} evictionCancel context.CancelFunc + // The pending counts: hits, misses, upstream fetches and transforms + // whose write to the database missed its request's deadline, kept + // here until they are written by the goroutine that + // StartPendingCountWrites starts. As for eviction, the channels are + // created in NewCache: pendingCountsAdded wakes that goroutine, and + // pendingCountsDone is closed when it returns. pendingCountsCancel, + // set by StartPendingCountWrites, cancels its context. + pendingHits atomic.Int64 + pendingMisses atomic.Int64 + pendingUpstreamFetches atomic.Int64 + pendingUpstreamFetchBytes atomic.Int64 + pendingTransforms atomic.Int64 + pendingCountsAdded chan struct{} + pendingCountsDone chan struct{} + pendingCountsCancel context.CancelFunc + // metaCache holds the content types of the variants most recently // stored or served, so a hit does not read the variant's .meta file. // It never stands in for the variant file, which is always opened. @@ -139,6 +156,9 @@ func newCache( metaCache: metaCache, contentLocks: newContentLock(), + pendingCountsAdded: make(chan struct{}, 1), + pendingCountsDone: make(chan struct{}), + reconciliationPageSize: defaultReconciliationPageSize, } @@ -484,6 +504,10 @@ func (c *Cache) Stats(ctx context.Context) (*CacheStats, error) { return nil, fmt.Errorf("failed to get cache stats: %w", err) } + // Hits and misses not yet written count too (see StartPendingCountWrites) + stats.HitCount += c.pendingHits.Load() + stats.MissCount += c.pendingMisses.Load() + // Count and size the cached source images and processed variants from // their tables. A disabled cache holds none, whatever rows an earlier // run left. @@ -511,19 +535,30 @@ func (c *Cache) Stats(ctx context.Context) (*CacheStats, error) { } // IncrementStats counts a cache hit or miss, and an upstream fetch that read -// fetchBytes bytes, as IncrementUpstreamFetch does. +// fetchBytes bytes, as IncrementUpstreamFetch does. Like the other Increment +// methods, it writes to the database with ctx's deadline but not its +// cancellation, so an ended request is still counted but does not wait for +// the database past its deadline; a count not written by then becomes a +// pending count (see StartPendingCountWrites). func (c *Cache) IncrementStats(ctx context.Context, hit bool, fetchBytes int64) { + countCtx, cancel := withoutCancelKeepingDeadline(ctx) + defer cancel() + var err error + pendingCount := &c.pendingMisses + if hit { - _, err = c.db.ExecContext(ctx, ` + pendingCount = &c.pendingHits + + _, err = c.db.ExecContext(countCtx, ` UPDATE cache_stats SET hit_count = hit_count + 1, last_updated_at = CURRENT_TIMESTAMP WHERE id = 1 `) } else { - _, err = c.db.ExecContext(ctx, ` + _, err = c.db.ExecContext(countCtx, ` UPDATE cache_stats SET miss_count = miss_count + 1, last_updated_at = CURRENT_TIMESTAMP @@ -531,7 +566,10 @@ func (c *Cache) IncrementStats(ctx context.Context, hit bool, fetchBytes int64) `) } - if err != nil { + switch { + case errors.Is(err, context.DeadlineExceeded): + c.addPendingCount(pendingCount, 1) + case err != nil: c.log.Warn("failed to count cache hit or miss", "hit", hit, "error", err) } @@ -545,14 +583,22 @@ func (c *Cache) IncrementUpstreamFetch(ctx context.Context, fetchBytes int64) { return } - _, err := c.db.ExecContext(ctx, ` + countCtx, cancel := withoutCancelKeepingDeadline(ctx) + defer cancel() + + _, err := c.db.ExecContext(countCtx, ` UPDATE cache_stats SET upstream_fetch_count = upstream_fetch_count + 1, upstream_fetch_bytes = upstream_fetch_bytes + ?, last_updated_at = CURRENT_TIMESTAMP WHERE id = 1 `, fetchBytes) - if err != nil { + + switch { + case errors.Is(err, context.DeadlineExceeded): + c.addPendingCount(&c.pendingUpstreamFetches, 1) + c.addPendingCount(&c.pendingUpstreamFetchBytes, fetchBytes) + case err != nil: c.log.Warn("failed to count upstream fetch", "fetch_bytes", fetchBytes, "error", err) } @@ -560,13 +606,20 @@ func (c *Cache) IncrementUpstreamFetch(ctx context.Context, fetchBytes int64) { // IncrementTransformCount counts one image transcoded by the image processor. func (c *Cache) IncrementTransformCount(ctx context.Context) { - _, err := c.db.ExecContext(ctx, ` + countCtx, cancel := withoutCancelKeepingDeadline(ctx) + defer cancel() + + _, err := c.db.ExecContext(countCtx, ` UPDATE cache_stats SET transform_count = transform_count + 1, last_updated_at = CURRENT_TIMESTAMP WHERE id = 1 `) - if err != nil { + + switch { + case errors.Is(err, context.DeadlineExceeded): + c.addPendingCount(&c.pendingTransforms, 1) + case err != nil: c.log.Warn("failed to count transform", "error", err) } } diff --git a/internal/imgcache/pending_counts_internal_test.go b/internal/imgcache/pending_counts_internal_test.go new file mode 100644 index 0000000..85ab531 --- /dev/null +++ b/internal/imgcache/pending_counts_internal_test.go @@ -0,0 +1,263 @@ +package imgcache + +import ( + "context" + "database/sql" + "errors" + "testing" + "time" +) + +// pendingCountsWait is how long a test waits for a count to reach the +// database, or for a write to start waiting for the database's connection. +const pendingCountsWait = 5 * time.Second + +// holdDatabase takes the one connection of cache's database, so that every +// other query waits for it, and returns the func that frees it. +func holdDatabase(t *testing.T, cache *Cache) func() { + t.Helper() + + conn, err := cache.db.Conn(t.Context()) + if err != nil { + t.Fatalf("failed to take the database connection: %v", err) + } + + return func() { _ = conn.Close() } +} + +// waitForConnectionWaits waits until db.Stats().WaitCount, the number of +// times a caller has waited for a connection, reaches waits. +func waitForConnectionWaits(t *testing.T, db *sql.DB, waits int64) { + t.Helper() + + deadline := time.Now().Add(pendingCountsWait) + + for db.Stats().WaitCount < waits { + if time.Now().After(deadline) { + t.Fatalf("callers waited for the database connection %d times, want %d", + db.Stats().WaitCount, waits) + } + + time.Sleep(10 * time.Millisecond) + } +} + +// waitForCounters waits until the cache_stats row holds want. +func waitForCounters(t *testing.T, cache *Cache, want cacheStatsCounters) { + t.Helper() + + deadline := time.Now().Add(pendingCountsWait) + + for { + got := readCacheStatsCounters(t, cache) + if got == want { + return + } + + if time.Now().After(deadline) { + t.Fatalf("counters = %+v, want %+v", got, want) + } + + time.Sleep(10 * time.Millisecond) + } +} + +// TestService_Get_ReturnsByItsDeadlineWhileTheDatabaseIsBusy holds the +// database's one connection while a request whose fetch is held reaches its +// deadline. The request must still return by its deadline with the +// deadline's error, and its miss must reach the database once the +// connection is free. +func TestService_Get_ReturnsByItsDeadlineWhileTheDatabaseIsBusy(t *testing.T) { + t.Parallel() + + svc, fixtures, fetcher := setupHeldFetchService(t) + + svc.cache.StartPendingCountWrites() + defer func() { _ = svc.cache.StopPendingCountWrites(t.Context()) }() + + const timeout = 200 * time.Millisecond + + ctx, cancel := context.WithTimeout(t.Context(), timeout) + defer cancel() + + results := startGet(ctx, svc, photoVariant(fixtures, 85, FitCover)) + + // The request has made its database reads by the time it fetches. + <-fetcher.started + + releaseDatabase := holdDatabase(t, svc.cache) + defer releaseDatabase() + + select { + case got := <-results: + deadline, _ := ctx.Deadline() + t.Logf("Get() returned %v after its deadline, error = %v", + time.Since(deadline), got.err) + + if !errors.Is(got.err, context.DeadlineExceeded) { + t.Errorf("Get() error = %v, want %v", got.err, context.DeadlineExceeded) + } + case <-time.After(timeout + time.Second): + t.Fatal("request did not return by its deadline while the database was busy") + } + + releaseDatabase() + + waitForCounters(t, svc.cache, cacheStatsCounters{missCount: 1}) +} + +// TestService_Get_ManyRequestsPastTheirDeadlineLeaveOneWriteWaiting holds the +// database's one connection while many requests whose fetches are held reach +// their deadline. At most one write may then wait for the connection, and +// every miss must reach the database once it is free. +func TestService_Get_ManyRequestsPastTheirDeadlineLeaveOneWriteWaiting(t *testing.T) { + t.Parallel() + + svc, fixtures, fetcher := setupHeldFetchService(t) + + svc.cache.StartPendingCountWrites() + defer func() { _ = svc.cache.StopPendingCountWrites(t.Context()) }() + + const ( + requests = 20 + timeout = 200 * time.Millisecond + ) + + ctx, cancel := context.WithTimeout(t.Context(), timeout) + defer cancel() + + // Each request asks for a variant of its own, so each fetches. + results := make([]<-chan getResult, requests) + for i := range results { + results[i] = startGet(ctx, svc, photoVariant(fixtures, i+1, FitCover)) + } + + // Each request has made its database reads by the time it fetches. + for range requests { + <-fetcher.started + } + + releaseDatabase := holdDatabase(t, svc.cache) + defer releaseDatabase() + + waitsBefore := svc.cache.db.Stats().WaitCount + + // Once one write waits for the connection, any others start waiting too. + waitForConnectionWaits(t, svc.cache.db, waitsBefore+1) + time.Sleep(arrivalWait) + + waits := svc.cache.db.Stats().WaitCount - waitsBefore + t.Logf("%d writes waited for the database", waits) + + if waits > 1 { + t.Errorf("%d writes waited for the database, want at most 1", waits) + } + + for _, result := range results { + select { + case got := <-result: + if !errors.Is(got.err, context.DeadlineExceeded) { + t.Errorf("Get() error = %v, want %v", got.err, context.DeadlineExceeded) + } + case <-time.After(time.Second): + t.Fatal("request did not return by its deadline while the database was busy") + } + } + + releaseDatabase() + + waitForCounters(t, svc.cache, cacheStatsCounters{missCount: requests}) +} + +// TestStats_IncludesPendingCounts counts hits and misses whose writes miss +// their deadline, and checks that Stats adds them to the database's counts +// before they are written. +func TestStats_IncludesPendingCounts(t *testing.T) { + t.Parallel() + + cache, _ := newEvictionTestCache(t, 1<<30) + + _, err := cache.db.ExecContext(t.Context(), + `UPDATE cache_stats SET hit_count = 75, miss_count = 25 WHERE id = 1`) + if err != nil { + t.Fatal(err) + } + + // Writes given a context past its deadline fail at once. + ended, cancel := context.WithDeadline(t.Context(), time.Now()) + defer cancel() + + cache.IncrementStats(ended, true, 0) + cache.IncrementStats(ended, true, 0) + cache.IncrementStats(ended, false, 0) + + stats, err := cache.Stats(t.Context()) + if err != nil { + t.Fatalf("Stats() error = %v", err) + } + + if stats.HitCount != 77 || stats.MissCount != 26 { + t.Errorf("HitCount = %d, MissCount = %d, want 77 and 26", + stats.HitCount, stats.MissCount) + } + + want := cacheStatsCounters{hitCount: 75, missCount: 25} + if got := readCacheStatsCounters(t, cache); got != want { + t.Errorf("counters in the database = %+v, want %+v", got, want) + } +} + +// TestStopPendingCountWrites_WritesPendingCounts holds the database's one +// connection while the goroutine that writes the pending counts waits for it, +// then stops that goroutine. The stop must write the counts once the +// connection is free. +func TestStopPendingCountWrites_WritesPendingCounts(t *testing.T) { + t.Parallel() + + cache, _ := newEvictionTestCache(t, 1<<30) + + cache.StartPendingCountWrites() + defer func() { _ = cache.StopPendingCountWrites(t.Context()) }() + + releaseDatabase := holdDatabase(t, cache) + defer releaseDatabase() + + waitsBefore := cache.db.Stats().WaitCount + + // Writes given a context past its deadline fail at once. + ended, cancel := context.WithDeadline(t.Context(), time.Now()) + defer cancel() + + cache.IncrementStats(ended, true, 0) + cache.IncrementStats(ended, false, 0) + cache.IncrementUpstreamFetch(ended, 1024) + cache.IncrementTransformCount(ended) + + // The goroutine's write waits for the connection. + waitForConnectionWaits(t, cache.db, waitsBefore+1) + + stopped := make(chan error, 1) + + go func() { + stopped <- cache.StopPendingCountWrites(t.Context()) + }() + + // The stop's own write waits for it once the goroutine has returned. + waitForConnectionWaits(t, cache.db, waitsBefore+2) + + releaseDatabase() + + select { + case err := <-stopped: + if err != nil { + t.Fatalf("StopPendingCountWrites() error = %v", err) + } + case <-time.After(pendingCountsWait): + t.Fatal("StopPendingCountWrites() did not return once the database was free") + } + + want := cacheStatsCounters{1, 1, 1, 1024, 1} + if got := readCacheStatsCounters(t, cache); got != want { + t.Errorf("counters = %+v, want %+v", got, want) + } +} diff --git a/internal/imgcache/pendingcounts.go b/internal/imgcache/pendingcounts.go new file mode 100644 index 0000000..6f5bde2 --- /dev/null +++ b/internal/imgcache/pendingcounts.go @@ -0,0 +1,127 @@ +package imgcache + +import ( + "context" + "fmt" + "sync/atomic" +) + +// StartPendingCountWrites starts the goroutine that writes the pending +// counts to the database whenever there are some, one UPDATE at a time: +// however many requests pass their deadline before their counts are +// written, at most this one write waits for the database. It is a no-op +// when already started. The goroutine outlives the caller, so it runs with +// its own context, which StopPendingCountWrites cancels. +func (c *Cache) StartPendingCountWrites() { + if c.pendingCountsCancel != nil { + return + } + + ctx, cancel := context.WithCancel(context.Background()) + c.pendingCountsCancel = cancel + + go func() { + c.pendingCountWriteLoop(ctx) + }() +} + +// StopPendingCountWrites stops the goroutine StartPendingCountWrites +// started, if it did, then writes the pending counts once more, so that +// they are written before the database closes. It waits for both at most +// until ctx ends. It logs the counts still unwritten then, which are lost, +// and returns an error. +func (c *Cache) StopPendingCountWrites(ctx context.Context) error { + if c.pendingCountsCancel != nil { + c.pendingCountsCancel() + + select { + case <-c.pendingCountsDone: + case <-ctx.Done(): + } + } + + err := c.writePendingCounts(ctx) + if err != nil { + c.log.Error("counts lost at shutdown", + "hits", c.pendingHits.Load(), + "misses", c.pendingMisses.Load(), + "upstream_fetches", c.pendingUpstreamFetches.Load(), + "upstream_fetch_bytes", c.pendingUpstreamFetchBytes.Load(), + "transforms", c.pendingTransforms.Load(), + "error", err, + ) + + return err + } + + return nil +} + +// pendingCountWriteLoop is the body of the goroutine +// StartPendingCountWrites starts. It returns when ctx is cancelled. A +// write that fails leaves the counts pending, for the next write. +func (c *Cache) pendingCountWriteLoop(ctx context.Context) { + defer close(c.pendingCountsDone) + + for { + select { + case <-ctx.Done(): + return + case <-c.pendingCountsAdded: + } + + err := c.writePendingCounts(ctx) + if err != nil && ctx.Err() == nil { + c.log.Warn("failed to write pending counts", "error", err) + } + } +} + +// addPendingCount adds n to pendingCount, one of the pending counts, and +// wakes the goroutine that writes them. pendingCountsAdded has capacity one, +// so a wakeup already waiting is enough. +func (c *Cache) addPendingCount(pendingCount *atomic.Int64, n int64) { + pendingCount.Add(n) + + select { + case c.pendingCountsAdded <- struct{}{}: + default: + } +} + +// writePendingCounts adds the pending counts to the cache_stats row in one +// UPDATE, then takes what it wrote out of them, leaving any added +// meanwhile. When the UPDATE fails, they all stay pending. +func (c *Cache) writePendingCounts(ctx context.Context) error { + hits := c.pendingHits.Load() + misses := c.pendingMisses.Load() + upstreamFetches := c.pendingUpstreamFetches.Load() + upstreamFetchBytes := c.pendingUpstreamFetchBytes.Load() + transforms := c.pendingTransforms.Load() + + if hits+misses+upstreamFetches+upstreamFetchBytes+transforms == 0 { + return nil + } + + _, err := c.db.ExecContext(ctx, ` + UPDATE cache_stats + SET hit_count = hit_count + ?, + miss_count = miss_count + ?, + upstream_fetch_count = upstream_fetch_count + ?, + upstream_fetch_bytes = upstream_fetch_bytes + ?, + transform_count = transform_count + ?, + last_updated_at = CURRENT_TIMESTAMP + WHERE id = 1 + `, hits, misses, upstreamFetches, upstreamFetchBytes, transforms) + if err != nil { + return fmt.Errorf("failed to write pending counts: %w", err) + } + + c.pendingHits.Add(-hits) + c.pendingMisses.Add(-misses) + c.pendingUpstreamFetches.Add(-upstreamFetches) + c.pendingUpstreamFetchBytes.Add(-upstreamFetchBytes) + c.pendingTransforms.Add(-transforms) + + return nil +} diff --git a/internal/imgcache/service.go b/internal/imgcache/service.go index f196b26..52684a9 100644 --- a/internal/imgcache/service.go +++ b/internal/imgcache/service.go @@ -162,7 +162,7 @@ func (s *Service) Get(ctx context.Context, req *ImageRequest) (*ImageResponse, e // Fall through to re-process } else { // Counted also when the request context has ended meanwhile - s.cache.IncrementStats(context.WithoutCancel(ctx), true, 0) + s.cache.IncrementStats(ctx, true, 0) return &ImageResponse{ Content: reader, @@ -179,7 +179,7 @@ func (s *Service) Get(ctx context.Context, req *ImageRequest) (*ImageResponse, e // failed or the request context has ended meanwhile response, err := s.processOrWait(ctx, req) - s.cache.IncrementStats(context.WithoutCancel(ctx), false, 0) + s.cache.IncrementStats(ctx, false, 0) if err != nil { return nil, err @@ -300,14 +300,8 @@ func (s *Service) processOrWait( } }() - processingCtx := context.WithoutCancel(ctx) - - if deadline, ok := ctx.Deadline(); ok { - var cancel context.CancelFunc - - processingCtx, cancel = context.WithDeadline(processingCtx, deadline) - defer cancel() - } + processingCtx, cancel := withoutCancelKeepingDeadline(ctx) + defer cancel() return s.processFromSourceOrFetch(processingCtx, req, cacheKey) }) @@ -344,6 +338,20 @@ func (s *Service) processOrWait( }, nil } +// withoutCancelKeepingDeadline returns ctx without its cancellation but with +// its deadline, if it has one, and the func that releases the returned +// context. +func withoutCancelKeepingDeadline( + ctx context.Context, +) (context.Context, context.CancelFunc) { + deadline, hasDeadline := ctx.Deadline() + if !hasDeadline { + return context.WithoutCancel(ctx), func() {} + } + + return context.WithDeadline(context.WithoutCancel(ctx), deadline) +} + // loadCachedSource opens source content from cache, without reading it, and // returns it with its size; nil if the cached data is unavailable, empty or // exceeds maxResponseSize. @@ -453,7 +461,7 @@ func (s *Service) fetchAndProcess( fetchBytes := int64(len(sourceData)) // Counted also when the request context has ended meanwhile - s.cache.IncrementUpstreamFetch(context.WithoutCancel(ctx), fetchBytes) + s.cache.IncrementUpstreamFetch(ctx, fetchBytes) if err != nil { return nil, fmt.Errorf("failed to read upstream response: %w", err) @@ -535,7 +543,7 @@ func (s *Service) processAndStore( processDuration := time.Since(processStart) // Counted also when the request context has ended meanwhile - s.cache.IncrementTransformCount(context.WithoutCancel(ctx)) + s.cache.IncrementTransformCount(ctx) // Read processed content processedData, err := io.ReadAll(processResult.Content)