pixa/internal/imgcache/pendingcounts_internal_test.go

package imgcache

import (
	"bytes"
	"context"
	"database/sql"
	"errors"
	"log/slog"
	"strings"
	"testing"
	"time"

	"sneak.berlin/go/pixa/internal/imagerequest"
)

// 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
// 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() {
	t.Helper()

	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)
	}

	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)
	}
}

// waitForPendingCountsWritten starts the goroutine that writes cache's
// pending counts, unless it is running, and waits until it has written them
// all, so that the database holds every count made so far. The goroutine is
// stopped when the test ends. The database is limited to one connection
// first: each connection to an in-memory SQLite database opens a new, empty
// one, so the goroutine must share the connection that holds the schema.
func waitForPendingCountsWritten(t *testing.T, cache *Cache) {
	t.Helper()

	if cache.pendingCountsCancel == nil {
		cache.db.SetMaxOpenConns(1)
		cache.StartPendingCountWrites()

		t.Cleanup(func() { _ = cache.StopPendingCountWrites(context.Background()) })
	}

	deadline := time.Now().Add(pendingCountsWait)

	for cache.pendingHits.Load()+cache.pendingMisses.Load()+
		cache.pendingUpstreamFetches.Load()+cache.pendingUpstreamFetchBytes.Load()+
		cache.pendingTransforms.Load() > 0 {
		if time.Now().After(deadline) {
			t.Fatal("the pending counts were not written")
		}

		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 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()

	svc, fixtures, fetcher := setupHeldFetchService(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 = 2 * time.Second

	ctx, cancel := context.WithTimeout(t.Context(), timeout)
	defer cancel()

	deadline, _ := ctx.Deadline()

	results := startGet(ctx, svc, photoVariant(fixtures, 85, imagerequest.FitCover))

	// The request has made its database reads by the time it fetches.
	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:
		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) + pendingCountsWait):
		t.Fatal("request did not return while the database was busy")
	}

	releaseDatabase()

	waitForCounters(t, svc.cache, cacheStatsCounters{missCount: 1})
}

// TestIncrementStats_ManyCountsLeaveOneWriteWaiting holds the database's one
// connection while many misses are added to the pending counts, which the
// goroutine StartPendingCountWrites starts writes to the database. At most one
// write may then wait for the connection, and every miss must reach the
// database once it is free.
func TestIncrementStats_ManyCountsLeaveOneWriteWaiting(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

	const misses = 20

	for range misses {
		cache.IncrementStats(false, 0)
	}

	// Once one write waits for the connection, any others start waiting too.
	waitForConnectionWaits(t, cache.db, waitsBefore+1)
	time.Sleep(otherWritesWait)

	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)
	}

	releaseDatabase()

	waitForCounters(t, cache, cacheStatsCounters{missCount: misses})
}

// TestStats_IncludesPendingCounts adds hits and misses to the pending counts,
// with no goroutine started to write them, 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)
	}

	cache.IncrementStats(true, 0)
	cache.IncrementStats(true, 0)
	cache.IncrementStats(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,
// 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
// must write the counts that write left.
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

	cache.IncrementStats(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(false, 0)
	cache.IncrementUpstreamFetch(1024)
	cache.IncrementTransformCount()

	// 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() {
		stopped <- cache.StopPendingCountWrites(t.Context())
	}()

	// Had the stop cut off the write under way, its own write would wait for
	// the connection too.
	time.Sleep(otherWritesWait)

	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()

	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)
	}
}

// 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

	cache.IncrementStats(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})
}

// TestLookupSource_HandsAccessTimeToTheBackgroundWriter finds a cached source
// whose access time is an hour old. LookupSource must not write the access
// time itself but hand it over to be written, and StopPendingCountWrites must
// then write it.
func TestLookupSource_HandsAccessTimeToTheBackgroundWriter(t *testing.T) {
	t.Parallel()

	cache, _ := newEvictionTestCache(t, 1<<30)
	key := storeEvictionTestSource(t, cache, "example.com", "/photo.jpg",
		[]byte("source bytes"))
	anHourAgo := time.Now().Add(-time.Hour)
	setSourceLastAccessed(t, cache, key, anHourAgo)

	touchedQuery := `
		SELECT COUNT(*) FROM source_content
		WHERE source_key = ? AND last_accessed_at > ?
	`
	anHourAgoInDatabase := anHourAgo.UTC().Format(sqliteTimestampFormat)

	found, err := cache.LookupSource(t.Context(), &imagerequest.Request{
		Source: imagerequest.Source{Host: "example.com", Path: "/photo.jpg"},
	})
	if err != nil || found != key {
		t.Fatalf("LookupSource() = %q, %v; want %q", found, err, key)
	}

	if countRows(t, cache, touchedQuery, string(key), anHourAgoInDatabase) != 0 {
		t.Error("LookupSource wrote the source's access time itself")
	}

	if waiting := len(cache.pendingSourceAccessTimes); waiting != 1 {
		t.Errorf("%d access times of sources wait to be written, want 1", waiting)
	}

	err = cache.StopPendingCountWrites(t.Context())
	if err != nil {
		t.Fatalf("StopPendingCountWrites() error = %v", err)
	}

	if countRows(t, cache, touchedQuery, string(key), anHourAgoInDatabase) != 1 {
		t.Error("the source's access time was not written")
	}
}

// TestStopPendingCountWrites_LogsWhatWasNotWritten makes the write of the
// access times fail at shutdown, by dropping the table they are written to,
// while a hit count is written. The stop must return an error and log that
// access times were not written, and must not log the counts as not written.
func TestStopPendingCountWrites_LogsWhatWasNotWritten(t *testing.T) {
	t.Parallel()

	cache, _ := newEvictionTestCache(t, 1<<30)

	var logs bytes.Buffer

	cache.log = slog.New(slog.NewJSONHandler(&logs, nil))

	cache.touchVariant(testVariantKeyOne)
	cache.IncrementStats(true, 0)

	_, err := cache.db.ExecContext(t.Context(), `DROP TABLE variant_content`)
	if err != nil {
		t.Fatalf("dropping variant_content: %v", err)
	}

	err = cache.StopPendingCountWrites(t.Context())
	t.Logf("StopPendingCountWrites() error = %v; logged:\n%s", err, logs.String())

	if err == nil {
		t.Error("StopPendingCountWrites() error = nil, want the access times' error")
	}

	if !strings.Contains(logs.String(), "access times not written at shutdown") {
		t.Error("the access times not written were not logged")
	}

	if strings.Contains(logs.String(), "counts not written at shutdown") {
		t.Error("the counts, which were written, were logged as not written")
	}

	if got := readCacheStatsCounters(t, cache); got.hitCount != 1 {
		t.Errorf("hit count in the database = %d, want 1", got.hitCount)
	}
}