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

@@ -98,7 +98,8 @@ type Cache struct {
 	pendingCountsDone         chan struct{}
 	pendingCountsCancel       context.CancelFunc
 
-	// Held through each write of the pending counts, and by Stats to read them
+	// Held through each write of the pending counts, and by
+	// readHitAndMissCounts to read them
 	pendingCountsWriteMutex sync.Mutex
 
 	// metaCache holds the content types of the variants most recently
@@ -499,21 +500,7 @@ 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()
-
+	err := c.readHitAndMissCounts(ctx, &stats)
 	if err != nil && !errors.Is(err, sql.ErrNoRows) {
 		return nil, fmt.Errorf("failed to get cache stats: %w", err)
 	}
@@ -634,6 +621,25 @@ func (c *Cache) IncrementTransformCount(ctx context.Context) {
 	}
 }
 
+// readHitAndMissCounts sets the hit and miss counts of stats to those of the
+// cache_stats row plus the pending hits and misses (see
+// StartPendingCountWrites). It holds pendingCountsWriteMutex, so that no
+// write of the pending counts falls between the two reads.
+func (c *Cache) readHitAndMissCounts(ctx context.Context, stats *CacheStats) error {
+	c.pendingCountsWriteMutex.Lock()
+	defer c.pendingCountsWriteMutex.Unlock()
+
+	err := c.db.QueryRowContext(ctx, `
+		SELECT hit_count, miss_count
+		FROM cache_stats WHERE id = 1
+	`).Scan(&stats.HitCount, &stats.MissCount)
+
+	stats.HitCount += c.pendingHits.Load()
+	stats.MissCount += c.pendingMisses.Load()
+
+	return err
+}
+
 // loadVariantWithMeta is GetVariant for a variant metaCache does not
 // hold: it reads the content type from the variant's .meta file and
 // keeps it in metaCache, unless a StoreVariant has put one there

internal/imgcache/pendingcounts.go

@@ -68,9 +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, but
-// does not cut off a write under way then. A write that fails leaves the
-// counts pending, for the next write.
+// StartPendingCountWrites starts. Once ctx is cancelled it starts no write
+// and returns, 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)
 
@@ -81,6 +81,12 @@ func (c *Cache) pendingCountWriteLoop(ctx context.Context) {
 		case <-c.pendingCountsAdded:
 		}
 
+		// select picks at random when ctx is cancelled and a wakeup
+		// waits too.
+		if ctx.Err() != nil {
+			return
+		}
+
 		err := c.writePendingCounts(context.WithoutCancel(ctx))
 		if err != nil {
 			c.log.Warn("failed to write pending counts", "error", err)

internal/imgcache/pendingcounts_internal_test.go

@@ -10,9 +10,14 @@ import (
 
 // 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.
+// connection, for a request to return past its deadline, or for
+// StopPendingCountWrites to return.
 const pendingCountsWait = 5 * time.Second
 
+// otherWritesWait is how long a test gives other writes, once one write waits
+// for the database's connection, to start waiting for it too.
+const otherWritesWait = 100 * time.Millisecond
+
 // 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() {
@@ -68,9 +73,9 @@ func waitForCounters(t *testing.T, cache *Cache, want cacheStatsCounters) {
 
 // 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.
+// deadline. The request must return with the deadline's error while the
+// connection is still held, as it waits for no database write past its
+// deadline, and its miss must reach the database once the connection is free.
 func TestService_Get_ReturnsByItsDeadlineWhileTheDatabaseIsBusy(t *testing.T) {
 	t.Parallel()
 
@@ -107,8 +112,8 @@ func TestService_Get_ReturnsByItsDeadlineWhileTheDatabaseIsBusy(t *testing.T) {
 		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):
-		t.Fatal("request did not return by its deadline while the database was busy")
+	case <-time.After(time.Until(deadline) + pendingCountsWait):
+		t.Fatal("request did not return while the database was busy")
 	}
 
 	releaseDatabase()
@@ -145,7 +150,7 @@ func TestIncrementStats_ManyCountsPastTheirDeadlineLeaveOneWriteWaiting(t *testi
 
 	// Once one write waits for the connection, any others start waiting too.
 	waitForConnectionWaits(t, cache.db, waitsBefore+1)
-	time.Sleep(arrivalWait)
+	time.Sleep(otherWritesWait)
 
 	waits := cache.db.Stats().WaitCount - waitsBefore
 	t.Logf("%d writes waited for the database", waits)
@@ -200,8 +205,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
-// way to finish rather than cut it off, and every count must reach the
-// database once the connection is free.
+// way to finish rather than cut it off, and once the connection is free it
+// must write the counts that write left.
 func TestStopPendingCountWrites_WritesPendingCounts(t *testing.T) {
 	t.Parallel()
 
@@ -229,6 +234,12 @@ func TestStopPendingCountWrites_WritesPendingCounts(t *testing.T) {
 	cache.IncrementUpstreamFetch(ended, 1024)
 	cache.IncrementTransformCount(ended)
 
+	// The goroutine is told to finish here as well as by the stop, so that it
+	// is told before its write under way ends, however late the stop runs.
+	// The counts above then reach the database only through the stop's own
+	// write.
+	cache.pendingCountsCancel()
+
 	stopped := make(chan error, 1)
 
 	go func() {
@@ -237,7 +248,7 @@ func TestStopPendingCountWrites_WritesPendingCounts(t *testing.T) {
 
 	// Had the stop cut off the write under way, its own write would wait for
 	// the connection too.
-	time.Sleep(arrivalWait)
+	time.Sleep(otherWritesWait)
 
 	waits := cache.db.Stats().WaitCount - waitsBefore
 	t.Logf("%d writes waited for the database", waits)