pixa/internal/database/concurrent_writes_internal_test.go

package database

import (
	"context"
	"database/sql"
	"fmt"
	"log/slog"
	"path/filepath"
	"sync"
	"testing"

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

// TestConcurrentWritesAllSucceed opens a database the way pixad does and
// writes to it from several goroutines at once, as one request's writes and
// the background eviction pass do. Every write must succeed, none failing
// with "database is locked", whether or not db_url already has parameters,
// and the parameters it has must still apply.
func TestConcurrentWritesAllSucceed(t *testing.T) {
	t.Parallel()

	tests := []struct {
		name            string
		query           string
		wantJournalMode string
	}{
		{
			name:            "db_url without parameters",
			query:           "",
			wantJournalMode: "delete",
		},
		{
			name:            "db_url with the WAL parameter",
			query:           "?_pragma=journal_mode(WAL)",
			wantJournalMode: "wal",
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			t.Parallel()

			dbURL := "file:" + filepath.Join(t.TempDir(), "state.sqlite3") + tt.query

			d := &Database{
				log:    slog.New(slog.DiscardHandler),
				config: &config.Config{DBURL: dbURL},
			}

			err := d.connect(t.Context())
			if err != nil {
				t.Fatalf("failed to connect to %q: %v", dbURL, err)
			}

			t.Cleanup(func() { _ = d.db.Close() })

			writeConcurrently(t, d.db)

			var journalMode string

			err = d.db.QueryRowContext(t.Context(), "PRAGMA journal_mode").
				Scan(&journalMode)
			if err != nil {
				t.Fatalf("failed to read the journal mode: %v", err)
			}

			if journalMode != tt.wantJournalMode {
				t.Errorf("journal mode = %q, want %q", journalMode, tt.wantJournalMode)
			}
		})
	}
}

// writeConcurrently runs writeLikeOneRequest from several goroutines at once
// and checks that every write was made.
func writeConcurrently(t *testing.T, db *sql.DB) {
	t.Helper()

	const (
		writers       = 4
		requestsEach  = 20
		totalRequests = writers * requestsEach
	)

	ctx := t.Context()

	var wg sync.WaitGroup

	for writer := range writers {
		wg.Go(func() {
			for request := range requestsEach {
				key := fmt.Sprintf("%d-%d", writer, request)

				err := writeLikeOneRequest(ctx, db, key)
				if err != nil {
					t.Errorf("writer %d: %v", writer, err)

					return
				}
			}
		})
	}

	wg.Wait()

	var hits, sources int

	err := db.QueryRowContext(ctx, `
		SELECT hit_count, (SELECT COUNT(*) FROM source_content)
		FROM cache_stats WHERE id = 1
	`).Scan(&hits, &sources)
	if err != nil {
		t.Fatalf("failed to count the writes: %v", err)
	}

	if hits != totalRequests || sources != totalRequests {
		t.Errorf("hit_count = %d and %d source_content rows, want %d of each",
			hits, sources, totalRequests)
	}
}

// writeLikeOneRequest makes the writes one request and the eviction pass
// make: it counts a cache hit, stores a source, records a transformed image
// and deletes that record again.
func writeLikeOneRequest(ctx context.Context, db *sql.DB, key string) error {
	_, err := db.ExecContext(ctx,
		`UPDATE cache_stats SET hit_count = hit_count + 1 WHERE id = 1`)
	if err != nil {
		return fmt.Errorf("counting a cache hit: %w", err)
	}

	_, err = db.ExecContext(ctx, `INSERT INTO source_content
		(content_hash, content_type, size_bytes) VALUES (?, 'image/png', 1)`, key)
	if err != nil {
		return fmt.Errorf("storing a source: %w", err)
	}

	_, err = db.ExecContext(ctx, `INSERT INTO variant_content
		(cache_key, size_bytes, content_type) VALUES (?, 1, 'image/png')`, key)
	if err != nil {
		return fmt.Errorf("recording a transformed image: %w", err)
	}

	_, err = db.ExecContext(ctx,
		`DELETE FROM variant_content WHERE cache_key = ?`, key)
	if err != nil {
		return fmt.Errorf("evicting a transformed image: %w", err)
	}

	return nil
}