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,6 +11,7 @@ import (
 	"io"
 	"log/slog"
 	"path/filepath"
+	"sync"
 	"sync/atomic"
 	"time"
 
@@ -86,7 +87,8 @@ 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.
+	// set by StartPendingCountWrites, cancels its context, which tells it
+	// to finish.
 	pendingHits               atomic.Int64
 	pendingMisses             atomic.Int64
 	pendingUpstreamFetches    atomic.Int64
@@ -96,6 +98,9 @@ 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.
@@ -494,20 +499,25 @@ 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)
 
-	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()
 
+	c.pendingCountsWriteMutex.Unlock()
+
+	if err != nil && !errors.Is(err, sql.ErrNoRows) {
+		return nil, fmt.Errorf("failed to get cache stats: %w", err)
+	}
+
 	// 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/pendingcounts.go

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

internal/imgcache/pendingcounts_internal_test.go

@@ -8,8 +8,9 @@ import (
 	"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.
+// pendingCountsWait is how long a test waits for the database's connection,
+// for a count to reach the database, for a write to start waiting for the
+// connection, or for StopPendingCountWrites to return.
 const pendingCountsWait = 5 * time.Second
 
 // holdDatabase takes the one connection of cache's database, so that every
@@ -17,7 +18,10 @@ const pendingCountsWait = 5 * time.Second
 func holdDatabase(t *testing.T, cache *Cache) func() {
 	t.Helper()
 
-	conn, err := cache.db.Conn(t.Context())
+	ctx, cancel := context.WithTimeout(t.Context(), pendingCountsWait)
+	defer cancel()
+
+	conn, err := cache.db.Conn(ctx)
 	if err != nil {
 		t.Fatalf("failed to take the database connection: %v", err)
 	}
@@ -75,29 +79,35 @@ func TestService_Get_ReturnsByItsDeadlineWhileTheDatabaseIsBusy(t *testing.T) {
 	svc.cache.StartPendingCountWrites()
 	defer func() { _ = svc.cache.StopPendingCountWrites(t.Context()) }()
 
-	const timeout = 200 * time.Millisecond
+	// Room for the request to reach its fetch on a busy host
+	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.
-	<-fetcher.started
+	select {
+	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(timeout + time.Second):
+	case <-time.After(time.Until(deadline) + time.Second):
 		t.Fatal("request did not return by its deadline while the database was busy")
 	}
 
@@ -106,67 +116,47 @@ func TestService_Get_ReturnsByItsDeadlineWhileTheDatabaseIsBusy(t *testing.T) {
 	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) {
+// TestIncrementStats_ManyCountsPastTheirDeadlineLeaveOneWriteWaiting holds the
+// database's one connection while many misses are counted past their
+// deadline. At most one write may then wait for the connection, and every
+// miss must reach the database once it is free.
+func TestIncrementStats_ManyCountsPastTheirDeadlineLeaveOneWriteWaiting(t *testing.T) {
 	t.Parallel()
 
-	svc, fixtures, fetcher := setupHeldFetchService(t)
+	cache, _ := newEvictionTestCache(t, 1<<30)
 
-	svc.cache.StartPendingCountWrites()
-	defer func() { _ = svc.cache.StopPendingCountWrites(t.Context()) }()
+	cache.StartPendingCountWrites()
+	defer func() { _ = 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)
+	releaseDatabase := holdDatabase(t, cache)
 	defer releaseDatabase()
 
-	waitsBefore := svc.cache.db.Stats().WaitCount
+	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()
+
+	const misses = 20
+
+	for range misses {
+		cache.IncrementStats(ended, false, 0)
+	}
 
 	// Once one write waits for the connection, any others start waiting too.
-	waitForConnectionWaits(t, svc.cache.db, waitsBefore+1)
+	waitForConnectionWaits(t, cache.db, waitsBefore+1)
 	time.Sleep(arrivalWait)
 
-	waits := svc.cache.db.Stats().WaitCount - waitsBefore
+	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 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})
+	waitForCounters(t, cache, cacheStatsCounters{missCount: misses})
 }
 
 // TestStats_IncludesPendingCounts counts hits and misses whose writes miss
@@ -209,8 +199,9 @@ 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,
-// then stops that goroutine. The stop must write the counts once the
-// connection is free.
+// counts more, then stops that goroutine. The stop must leave the write under
+// way to finish rather than cut it off, and every count must reach the
+// database once the connection is free.
 func TestStopPendingCountWrites_WritesPendingCounts(t *testing.T) {
 	t.Parallel()
 
@@ -229,21 +220,32 @@ func TestStopPendingCountWrites_WritesPendingCounts(t *testing.T) {
 	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)
 
+	// Counted after that write read the pending counts
+	cache.IncrementStats(ended, false, 0)
+	cache.IncrementUpstreamFetch(ended, 1024)
+	cache.IncrementTransformCount(ended)
+
 	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)
+	// Had the stop cut off the write under way, its own write would wait for
+	// the connection too.
+	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()
 
@@ -261,3 +263,54 @@ 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})
+}