Keep counts not written by the request's deadline in memory (closes #224)

TestService_Get_ReturnsByItsDeadline failed on a busy host because its
request, past its deadline, still waited to write its miss count, a write
with no deadline; in pixad that write 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 includes it, and one goroutine of the cache writes
those counts one UPDATE at a time, and once more at shutdown before the
database closes. The test phase also runs go test with -parallel 4: on a
busy host, as many tests at once as there are CPUs wait so long to be
scheduled that a timed request can fail.

Model: opus-5-5

internal/imgcache/cache.go

@@ -11,7 +11,6 @@ import (
 	"io"
 	"log/slog"
 	"path/filepath"
-	"sync"
 	"sync/atomic"
 	"time"
 
@@ -87,8 +86,7 @@ type Cache struct {
 	// 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, which tells it
+	// set by StartPendingCountWrites, cancels its context.
-	// to finish.
 	pendingHits               atomic.Int64
 	pendingMisses             atomic.Int64
 	pendingUpstreamFetches    atomic.Int64
@@ -98,9 +96,6 @@ type Cache struct {
 	pendingCountsDone         chan struct{}
 	pendingCountsCancel       context.CancelFunc
 
-	// Held through each write of the pending counts, and by Stats to read them
-	pendingCountsWriteMutex sync.Mutex
-
 	// 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.
@@ -499,25 +494,20 @@ func (c *Cache) CleanExpired(ctx context.Context) error {
 func (c *Cache) Stats(ctx context.Context) (*CacheStats, error) {
 	var stats CacheStats
 
-	// So that no write of the pending counts falls between the two reads
-	c.pendingCountsWriteMutex.Lock()
-
 	// Fetch hit/miss counts from the stats table
 	err := c.db.QueryRowContext(ctx, `
 		SELECT hit_count, miss_count
 		FROM cache_stats WHERE id = 1
 	`).Scan(&stats.HitCount, &stats.MissCount)
 
-	// Hits and misses not yet written count too (see StartPendingCountWrites)
-	stats.HitCount += c.pendingHits.Load()
-	stats.MissCount += c.pendingMisses.Load()
-
-	c.pendingCountsWriteMutex.Unlock()
-
 	if err != nil && !errors.Is(err, sql.ErrNoRows) {
 		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.

internal/imgcache/pending_counts_internal_test.go

@@ -8,9 +8,8 @@ import (
 	"time"
 )
 
-// pendingCountsWait is how long a test waits for the database's connection,
+// pendingCountsWait is how long a test waits for a count to reach the
-// for a count to reach the database, for a write to start waiting for the
+// database, or for a write to start waiting for the database's connection.
-// connection, or for StopPendingCountWrites to return.
 const pendingCountsWait = 5 * time.Second
 
 // holdDatabase takes the one connection of cache's database, so that every
@@ -18,10 +17,7 @@ const pendingCountsWait = 5 * time.Second
 func holdDatabase(t *testing.T, cache *Cache) func() {
 	t.Helper()
 
-	ctx, cancel := context.WithTimeout(t.Context(), pendingCountsWait)
+	conn, err := cache.db.Conn(t.Context())
-	defer cancel()
-
-	conn, err := cache.db.Conn(ctx)
 	if err != nil {
 		t.Fatalf("failed to take the database connection: %v", err)
 	}
@@ -79,35 +75,29 @@ func TestService_Get_ReturnsByItsDeadlineWhileTheDatabaseIsBusy(t *testing.T) {
 	svc.cache.StartPendingCountWrites()
 	defer func() { _ = svc.cache.StopPendingCountWrites(t.Context()) }()
 
-	// Room for the request to reach its fetch on a busy host
+	const timeout = 200 * time.Millisecond
-	const timeout = 2 * time.Second
 
 	ctx, cancel := context.WithTimeout(t.Context(), timeout)
 	defer cancel()
 
-	deadline, _ := ctx.Deadline()
-
 	results := startGet(ctx, svc, photoVariant(fixtures, 85, FitCover))
 
 	// The request has made its database reads by the time it fetches.
-	select {
+	<-fetcher.started
-	case <-fetcher.started:
-	case <-time.After(timeout):
-		t.Fatal("request did not reach its fetch by its deadline")
-	}
 
 	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(time.Until(deadline) + time.Second):
+	case <-time.After(timeout + time.Second):
 		t.Fatal("request did not return by its deadline while the database was busy")
 	}
 
@@ -116,47 +106,67 @@ func TestService_Get_ReturnsByItsDeadlineWhileTheDatabaseIsBusy(t *testing.T) {
 	waitForCounters(t, svc.cache, cacheStatsCounters{missCount: 1})
 }
 
-// TestIncrementStats_ManyCountsPastTheirDeadlineLeaveOneWriteWaiting holds the
+// TestService_Get_ManyRequestsPastTheirDeadlineLeaveOneWriteWaiting holds the
-// database's one connection while many misses are counted past their
+// database's one connection while many requests whose fetches are held reach
-// deadline. At most one write may then wait for the connection, and every
+// their deadline. At most one write may then wait for the connection, and
-// miss must reach the database once it is free.
+// every miss must reach the database once it is free.
-func TestIncrementStats_ManyCountsPastTheirDeadlineLeaveOneWriteWaiting(t *testing.T) {
+func TestService_Get_ManyRequestsPastTheirDeadlineLeaveOneWriteWaiting(t *testing.T) {
 	t.Parallel()
 
-	cache, _ := newEvictionTestCache(t, 1<<30)
+	svc, fixtures, fetcher := setupHeldFetchService(t)
 
-	cache.StartPendingCountWrites()
+	svc.cache.StartPendingCountWrites()
-	defer func() { _ = cache.StopPendingCountWrites(t.Context()) }()
+	defer func() { _ = svc.cache.StopPendingCountWrites(t.Context()) }()
 
-	releaseDatabase := holdDatabase(t, cache)
+	const (
-	defer releaseDatabase()
+		requests = 20
+		timeout  = 200 * time.Millisecond
+	)
 
-	waitsBefore := cache.db.Stats().WaitCount
+	ctx, cancel := context.WithTimeout(t.Context(), timeout)
-
-	// Writes given a context past its deadline fail at once.
-	ended, cancel := context.WithDeadline(t.Context(), time.Now())
 	defer cancel()
 
-	const misses = 20
+	// Each request asks for a variant of its own, so each fetches.
-
+	results := make([]<-chan getResult, requests)
-	for range misses {
+	for i := range results {
-		cache.IncrementStats(ended, false, 0)
+		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, cache.db, waitsBefore+1)
+	waitForConnectionWaits(t, svc.cache.db, waitsBefore+1)
 	time.Sleep(arrivalWait)
 
-	waits := cache.db.Stats().WaitCount - waitsBefore
+	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, cache, cacheStatsCounters{missCount: misses})
+	waitForCounters(t, svc.cache, cacheStatsCounters{missCount: requests})
 }
 
 // TestStats_IncludesPendingCounts counts hits and misses whose writes miss
@@ -199,9 +209,8 @@ func TestStats_IncludesPendingCounts(t *testing.T) {
 
 // TestStopPendingCountWrites_WritesPendingCounts holds the database's one
 // connection while the goroutine that writes the pending counts waits for it,
-// counts more, then stops that goroutine. The stop must leave the write under
+// then stops that goroutine. The stop must write the counts once the
-// way to finish rather than cut it off, and every count must reach the
+// connection is free.
-// database once the connection is free.
 func TestStopPendingCountWrites_WritesPendingCounts(t *testing.T) {
 	t.Parallel()
 
@@ -220,32 +229,21 @@ func TestStopPendingCountWrites_WritesPendingCounts(t *testing.T) {
 	defer cancel()
 
 	cache.IncrementStats(ended, true, 0)
-
-	// The goroutine's write waits for the connection.
-	waitForConnectionWaits(t, cache.db, waitsBefore+1)
-
-	// Counted after that write read the pending counts
 	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())
 	}()
 
-	// Had the stop cut off the write under way, its own write would wait for
+	// The stop's own write waits for it once the goroutine has returned.
-	// the connection too.
+	waitForConnectionWaits(t, cache.db, waitsBefore+2)
-	time.Sleep(arrivalWait)
-
-	waits := 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 1: "+
-			"the stop cut off the write under way", waits)
-	}
 
 	releaseDatabase()
 
@@ -263,54 +261,3 @@ func TestStopPendingCountWrites_WritesPendingCounts(t *testing.T) {
 		t.Errorf("counters = %+v, want %+v", got, want)
 	}
 }
-
-// TestStopPendingCountWrites_ReturnsWhenItsContextEnds holds the database's
-// one connection while the goroutine that writes the pending counts waits for
-// it, then stops that goroutine with a context that ends first. The stop must
-// return the context's error, and the write under way must still reach the
-// database once the connection is free.
-func TestStopPendingCountWrites_ReturnsWhenItsContextEnds(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)
-
-	// The goroutine's write waits for the connection.
-	waitForConnectionWaits(t, cache.db, waitsBefore+1)
-
-	stopCtx, cancelStop := context.WithCancel(t.Context())
-	cancelStop()
-
-	stopped := make(chan error, 1)
-
-	go func() {
-		stopped <- cache.StopPendingCountWrites(stopCtx)
-	}()
-
-	select {
-	case err := <-stopped:
-		if !errors.Is(err, context.Canceled) {
-			t.Errorf("StopPendingCountWrites() error = %v, want %v",
-				err, context.Canceled)
-		}
-	case <-time.After(pendingCountsWait):
-		t.Fatal("StopPendingCountWrites() did not return once its context ended")
-	}
-
-	releaseDatabase()
-
-	waitForCounters(t, cache, cacheStatsCounters{hitCount: 1})
-}

internal/imgcache/pendingcounts.go

@@ -11,8 +11,7 @@ import (
 // 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 to tell it to
+// its own context, which StopPendingCountWrites cancels.
-// finish.
 func (c *Cache) StartPendingCountWrites() {
 	if c.pendingCountsCancel != nil {
 		return
@@ -26,33 +25,24 @@ func (c *Cache) StartPendingCountWrites() {
 	}()
 }
 
-// StopPendingCountWrites tells the goroutine StartPendingCountWrites
+// StopPendingCountWrites stops the goroutine StartPendingCountWrites
-// started, if it did, to finish, and waits for its write under way to end
+// started, if it did, then writes the pending counts once more, so that
-// and for it to return. It then writes the pending counts left, so that
+// they are written before the database closes. It waits for both at most
-// they are written before the database closes. It waits for all of this at
+// until ctx ends. It logs the counts still unwritten then, which are lost,
-// most until ctx ends, then logs the counts not written and returns an
+// and returns an error.
-// error.
 func (c *Cache) StopPendingCountWrites(ctx context.Context) error {
-	var err error
-
 	if c.pendingCountsCancel != nil {
 		c.pendingCountsCancel()
 
 		select {
 		case <-c.pendingCountsDone:
 		case <-ctx.Done():
-			// The write under way is left to finish: the database's
-			// close waits for a query under way.
-			err = fmt.Errorf("pending counts still being written: %w", ctx.Err())
 		}
 	}
 
-	if err == nil {
+	err := c.writePendingCounts(ctx)
-		err = c.writePendingCounts(ctx)
-	}
-
 	if err != nil {
-		c.log.Error("counts not written at shutdown",
+		c.log.Error("counts lost at shutdown",
 			"hits", c.pendingHits.Load(),
 			"misses", c.pendingMisses.Load(),
 			"upstream_fetches", c.pendingUpstreamFetches.Load(),
@@ -68,9 +58,8 @@ func (c *Cache) StopPendingCountWrites(ctx context.Context) error {
 }
 
 // pendingCountWriteLoop is the body of the goroutine
-// StartPendingCountWrites starts. It returns when ctx is cancelled, but
+// StartPendingCountWrites starts. It returns when ctx is cancelled. A
-// does not cut off a write under way then. A write that fails leaves the
+// write that fails leaves the counts pending, for the next write.
-// counts pending, for the next write.
 func (c *Cache) pendingCountWriteLoop(ctx context.Context) {
 	defer close(c.pendingCountsDone)
 
@@ -81,8 +70,8 @@ func (c *Cache) pendingCountWriteLoop(ctx context.Context) {
 		case <-c.pendingCountsAdded:
 		}
 
-		err := c.writePendingCounts(context.WithoutCancel(ctx))
+		err := c.writePendingCounts(ctx)
-		if err != nil {
+		if err != nil && ctx.Err() == nil {
 			c.log.Warn("failed to write pending counts", "error", err)
 		}
 	}
@@ -104,9 +93,6 @@ func (c *Cache) addPendingCount(pendingCount *atomic.Int64, n int64) {
 // 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 {
-	c.pendingCountsWriteMutex.Lock()
-	defer c.pendingCountsWriteMutex.Unlock()
-
 	hits := c.pendingHits.Load()
 	misses := c.pendingMisses.Load()
 	upstreamFetches := c.pendingUpstreamFetches.Load()