Wait on a busy SQLite database and turn on WAL mode (closes #198)

Requests and the eviction pass write on separate connections, and with
no busy timeout a write that met another one failed at once with
"database is locked" and was lost. internal/database now adds
_pragma=busy_timeout(5000) to every db_url, the default or one the
operator sets, so such a write waits up to five seconds. The default
db_url's _journal_mode=WAL is not a parameter the driver reads, so it
is now _pragma=journal_mode(WAL). README.md and config.example.yml say
what pixa adds to db_url.

Model: opus-5-5

internal/database/concurrent_writes_internal_test.go

@@ -0,0 +1,153 @@
+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, so the writes run on
+// separate connections, 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
+}