pixa/internal/imgcache/pendingaccesstimes_internal_test.go

package imgcache

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

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

// pendingAccessTimesWait is how long a test waits for the database's
// connection, for an access time to reach the database, for a write to start
// waiting for the connection, or for StopPendingAccessTimeWrites to return.
const pendingAccessTimesWait = 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(), pendingAccessTimesWait)
	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(pendingAccessTimesWait)

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

// TestStopPendingAccessTimeWrites_LetsTheWriteUnderWayFinish holds the
// database's one connection while the goroutine that writes the pending
// access times waits for it, hands over another access time, 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 access time
// that write left.
func TestStopPendingAccessTimeWrites_LetsTheWriteUnderWayFinish(t *testing.T) {
	t.Parallel()

	cache, _ := newEvictionTestCache(t, 1<<30)
	anHourAgo := time.Now().Add(-time.Hour)

	for _, key := range []imagerequest.VariantKey{testVariantKeyOne, testVariantKeyTwo} {
		storeEvictionTestVariant(t, cache, key, []byte("variant bytes"))
		setVariantLastAccessed(t, cache, key, anHourAgo)
	}

	cache.StartPendingAccessTimeWrites()
	defer func() { _ = cache.StopPendingAccessTimeWrites(t.Context()) }()

	releaseDatabase := holdDatabase(t, cache)
	defer releaseDatabase()

	waitsBefore := cache.db.Stats().WaitCount

	cache.touchVariant(testVariantKeyOne)

	// The goroutine's write waits for the connection.
	waitForConnectionWaits(t, cache.db, waitsBefore+1)

	// Handed over after that write took the pending access times
	cache.touchVariant(testVariantKeyTwo)

	// 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 access time above then reaches the database only through the
	// stop's own write.
	cache.pendingAccessTimesCancel()

	stopped := make(chan error, 1)

	go func() {
		stopped <- cache.StopPendingAccessTimeWrites(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("StopPendingAccessTimeWrites() error = %v", err)
		}
	case <-time.After(pendingAccessTimesWait):
		t.Fatal("StopPendingAccessTimeWrites() did not return once the database was free")
	}

	for _, key := range []imagerequest.VariantKey{testVariantKeyOne, testVariantKeyTwo} {
		if countRows(t, cache, `
			SELECT COUNT(*) FROM variant_content
			WHERE cache_key = ? AND last_accessed_at > ?
		`, string(key), anHourAgo.UTC().Format(sqliteTimestampFormat)) != 1 {
			t.Errorf("the access time of variant %s was not written", key)
		}
	}
}

// TestStopPendingAccessTimeWrites_ReturnsWhenItsContextEnds holds the
// database's one connection while the goroutine that writes the pending
// access times 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 TestStopPendingAccessTimeWrites_ReturnsWhenItsContextEnds(t *testing.T) {
	t.Parallel()

	cache, _ := newEvictionTestCache(t, 1<<30)
	storeEvictionTestVariant(t, cache, testVariantKeyOne, []byte("variant bytes"))

	anHourAgo := time.Now().Add(-time.Hour)
	setVariantLastAccessed(t, cache, testVariantKeyOne, anHourAgo)

	cache.StartPendingAccessTimeWrites()
	defer func() { _ = cache.StopPendingAccessTimeWrites(t.Context()) }()

	releaseDatabase := holdDatabase(t, cache)
	defer releaseDatabase()

	waitsBefore := cache.db.Stats().WaitCount

	cache.touchVariant(testVariantKeyOne)

	// 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.StopPendingAccessTimeWrites(stopCtx)
	}()

	select {
	case err := <-stopped:
		if !errors.Is(err, context.Canceled) {
			t.Errorf("StopPendingAccessTimeWrites() error = %v, want %v",
				err, context.Canceled)
		}
	case <-time.After(pendingAccessTimesWait):
		t.Fatal("StopPendingAccessTimeWrites() did not return once its context ended")
	}

	releaseDatabase()

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

	for countRows(t, cache, `
		SELECT COUNT(*) FROM variant_content
		WHERE cache_key = ? AND last_accessed_at > ?
	`, string(testVariantKeyOne), anHourAgo.UTC().Format(sqliteTimestampFormat)) == 0 {
		if time.Now().After(deadline) {
			t.Fatal("the access time being written was not written")
		}

		time.Sleep(10 * time.Millisecond)
	}
}

// TestSource_HandsAccessTimeToTheBackgroundWriter finds a cached source
// whose access time is an hour old. Source must not write the access
// time itself but hand it over to be written, and StopPendingAccessTimeWrites
// must then write it.
func TestSource_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)

	reader, _, err := cache.Source(t.Context(),
		imagerequest.Source{Host: "example.com", Path: "/photo.jpg"})
	if err != nil {
		t.Fatalf("Source() error = %v; want the source of %q", err, key)
	}

	found, err := io.ReadAll(reader)
	_ = reader.Close()

	if err != nil || string(found) != "source bytes" {
		t.Fatalf("Source() reads %q, %v; want the source of %q", found, err, key)
	}

	if countRows(t, cache, touchedQuery, string(key), anHourAgoInDatabase) != 0 {
		t.Error("Source 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.StopPendingAccessTimeWrites(t.Context())
	if err != nil {
		t.Fatalf("StopPendingAccessTimeWrites() error = %v", err)
	}

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

// TestStopPendingAccessTimeWrites_LogsWhatWasNotWritten makes the write of
// the access times fail at shutdown, by dropping the table they are written
// to. The stop must return an error and log that access times were not
// written.
func TestStopPendingAccessTimeWrites_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)

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

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

	if err == nil {
		t.Error("StopPendingAccessTimeWrites() 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")
	}
}