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,8 +98,7 @@ type Cache struct {
 	pendingCountsDone         chan struct{}
 	pendingCountsCancel       context.CancelFunc
 
-	// Held through each write of the pending counts, and by
+	// Held through each write of the pending counts, and by Stats to read them
-	// readHitAndMissCounts to read them
 	pendingCountsWriteMutex sync.Mutex
 
 	// metaCache holds the content types of the variants most recently
@@ -500,7 +499,21 @@ func (c *Cache) CleanExpired(ctx context.Context) error {
 func (c *Cache) Stats(ctx context.Context) (*CacheStats, error) {
 	var stats CacheStats
 
-	err := c.readHitAndMissCounts(ctx, &stats)
+	// 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)
 	}
@@ -621,25 +634,6 @@ 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. Once ctx is cancelled it starts no write
+// StartPendingCountWrites starts. It returns when ctx is cancelled, but
-// and returns, but does not cut off a write under way then. A write that
+// does not cut off a write under way then. A write that fails leaves the
-// 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,12 +81,6 @@ 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,14 +10,9 @@ 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, for a request to return past its deadline, or for
+// connection, or for StopPendingCountWrites to return.
-// 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() {
@@ -73,9 +68,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 return with the deadline's error while the
+// deadline. The request must still return by its deadline with the
-// connection is still held, as it waits for no database write past its
+// deadline's error, and its miss must reach the database once the
-// deadline, and its miss must reach the database once the connection is free.
+// connection is free.
 func TestService_Get_ReturnsByItsDeadlineWhileTheDatabaseIsBusy(t *testing.T) {
 	t.Parallel()
 
@@ -112,8 +107,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) + pendingCountsWait):
+	case <-time.After(time.Until(deadline) + time.Second):
-		t.Fatal("request did not return while the database was busy")
+		t.Fatal("request did not return by its deadline while the database was busy")
 	}
 
 	releaseDatabase()
@@ -150,7 +145,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(otherWritesWait)
+	time.Sleep(arrivalWait)
 
 	waits := cache.db.Stats().WaitCount - waitsBefore
 	t.Logf("%d writes waited for the database", waits)
@@ -205,8 +200,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 once the connection is free it
+// way to finish rather than cut it off, and every count must reach the
-// must write the counts that write left.
+// database once the connection is free.
 func TestStopPendingCountWrites_WritesPendingCounts(t *testing.T) {
 	t.Parallel()
 
@@ -234,12 +229,6 @@ 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() {
@@ -248,7 +237,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(otherWritesWait)
+	time.Sleep(arrivalWait)
 
 	waits := cache.db.Stats().WaitCount - waitsBefore
 	t.Logf("%d writes waited for the database", waits)