Test the image proxy flow end to end (closes #80)

TestImageProxyFlow in internal/server starts the database, handlers and
middleware from the constructors pixad uses, in an fx app with a fresh
state directory, and replaces only the upstream origin with a local test
server, reached through the real fetcher by its new DialContext. For a
resize with a change to JPEG and for orig, the first request answers 200
with the right content type, decoded size and X-Pixa-Cache MISS; the
second answers HIT with the same image while the upstream has had one
request; the source and the converted image are then on disk with their
rows in SQLite. TODO.md records it and drops the item from Future Steps.

Model: opus-5-5

README.md

@@ -490,6 +490,12 @@ Key settings in more detail:
   waits for an upstream connection and for a processing slot (up to 10 seconds
   each), so keep it longer than `upstream_fetch_timeout` plus 20 seconds
 - `signing_key` — HMAC secret for URL signatures
+- `db_url` — the SQLite database to open; omitted, it is
+  `file:<state_dir>/state.sqlite3?_pragma=journal_mode(WAL)`, which keeps the
+  database in WAL mode. pixa adds `_pragma=busy_timeout(5000)` to any `db_url`,
+  so a write that finds another in progress waits up to five seconds for it
+  instead of failing. WAL mode comes only from the URL: keep
+  `_pragma=journal_mode(WAL)` in one you set
 - `cache_max_bytes` — disk cache size limit in bytes; `0` disables the
   disk cache entirely; omitted defaults to 75% of the sum of the free space on
   the filesystem containing `<state_dir>/cache/` and the bytes of source and

TODO.md

@@ -46,6 +46,20 @@ P2: security: referer blacklist
   `httpfetcher.Config.DialContext` connects in place of the dialer that refuses
   internal addresses, the URL and redirect checks still running, and
   `handlers.Params.Fetcher` replaces the fetcher the handlers build.
+- 2026-10-04 SQLite writes no longer fail with "database is locked" (closes
+  #198): pixa adds `_pragma=busy_timeout(5000)` to every `db_url`, so a write
+  that finds another in progress on another connection waits up to five seconds
+  for it, and the default `db_url` turns on WAL mode with
+  `_pragma=journal_mode(WAL)`. The old default's `_journal_mode=WAL` is not a
+  parameter the driver reads, so the database was never in WAL mode.
+- 2026-10-04 `TestPeriodicReconciliationAdoptsFileThatAppearsAfterStartup`
+  only passes through a periodic pass (closes #189): it slept for three
+  eviction intervals before writing its file, and a startup pass still running
+  then could adopt the file itself. It now holds the test database's only
+  connection until the startup pass waits for it after walking the empty
+  variant directory, writes the file and lets the connection go, as
+  `TestEvictionRunsOnPeriodicSchedule` does, so only a periodic reconciliation
+  pass can adopt the file. Test only.
 - 2026-10-04 logging in, logging out, the URL generator and `/v1/e/` have
   handler tests (closes #77): new tests in `internal/handlers`, with no
   network, check that `GET /` without a login session shows the login form; a

config.example.yml

@@ -34,9 +34,10 @@ maintenance_mode: false
 state_dir: ./data
 
 # SQLite database URL (default:
-# file:<state_dir>/state.sqlite3?_journal_mode=WAL). An empty value aborts
-# startup; leave the key out to use the default.
-# db_url: "file:./data/state.sqlite3?_journal_mode=WAL"
+# file:<state_dir>/state.sqlite3?_pragma=journal_mode(WAL)). pixa adds
+# _pragma=busy_timeout(5000) to it. An empty value aborts startup; leave the
+# key out to use the default.
+# db_url: "file:./data/state.sqlite3?_pragma=journal_mode(WAL)"
 
 # Image proxy settings
 # HMAC signing key for URL signatures (required, at least 32 characters)

internal/config/config.go

@@ -320,7 +320,8 @@ func newFromSmartConfig(sc *smartconfig.Config) (*Config, error) {
 				settingName(keyDBURL), errValueEmpty)
 		}
 
-		c.DBURL = fmt.Sprintf("file:%s/state.sqlite3?_journal_mode=WAL", c.StateDir)
+		// The driver sets the journal mode only through a _pragma parameter.
+		c.DBURL = fmt.Sprintf("file:%s/state.sqlite3?_pragma=journal_mode(WAL)", c.StateDir)
 	}
 
 	if loader.err != nil {

internal/config/config_validation_internal_test.go

@@ -1,6 +1,7 @@
 package config
 
 import (
+	"database/sql"
 	"log/slog"
 	"os"
 	"path/filepath"
@@ -9,6 +10,8 @@ import (
 	"time"
 
 	"git.eeqj.de/sneak/smartconfig"
+
+	_ "modernc.org/sqlite" // SQLite driver registration
 )
 
 // validTestSigningKey is a 32-character signing key that satisfies the
@@ -94,12 +97,43 @@ func TestOmittedValuesUseDefaults(t *testing.T) {
 		t.Errorf("AllowlistHosts = %v, want empty", c.AllowlistHosts)
 	}
 
-	wantDBURL := "file:" + DefaultStateDir + "/state.sqlite3?_journal_mode=WAL"
+	wantDBURL := "file:" + DefaultStateDir +
+		"/state.sqlite3?_pragma=journal_mode(WAL)"
 	if c.DBURL != wantDBURL {
 		t.Errorf("DBURL = %q, want derived default %q", c.DBURL, wantDBURL)
 	}
 }
 
+// TestDefaultDBURLOpensTheDatabaseInWALMode opens the db_url derived from
+// state_dir with the SQLite driver pixad uses and checks that the database
+// is in WAL mode: the driver ignores any parameter it does not know.
+func TestDefaultDBURLOpensTheDatabaseInWALMode(t *testing.T) {
+	t.Parallel()
+
+	c, err := configFromYAML(t, signingKeyLine+"state_dir: "+t.TempDir()+"\n")
+	if err != nil {
+		t.Fatalf("config with only state_dir set should be valid, got: %v", err)
+	}
+
+	db, err := sql.Open("sqlite", c.DBURL)
+	if err != nil {
+		t.Fatalf("failed to open %q: %v", c.DBURL, err)
+	}
+
+	t.Cleanup(func() { _ = db.Close() })
+
+	var journalMode string
+
+	err = db.QueryRowContext(t.Context(), "PRAGMA journal_mode").Scan(&journalMode)
+	if err != nil {
+		t.Fatalf("failed to read the journal mode of %q: %v", c.DBURL, err)
+	}
+
+	if journalMode != "wal" {
+		t.Errorf("journal mode of %q = %q, want wal", c.DBURL, journalMode)
+	}
+}
+
 func TestExplicitValidValuesAreUsed(t *testing.T) {
 	t.Parallel()
 

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

internal/database/database.go

@@ -243,7 +243,17 @@ func (s *Database) DB() *sql.DB {
 }
 
 func (s *Database) connect(ctx context.Context) error {
-	dbURL := s.config.DBURL
+	// Requests and the eviction pass write on separate connections. With
+	// a busy timeout, a write that finds another one in progress waits up
+	// to five seconds for it instead of failing at once with "database is
+	// locked". The driver runs each _pragma parameter on every connection
+	// it opens.
+	separator := "?"
+	if strings.Contains(s.config.DBURL, "?") {
+		separator = "&"
+	}
+
+	dbURL := s.config.DBURL + separator + "_pragma=busy_timeout(5000)"
 
 	s.log.Info("connecting to database", "url", dbURL)
 

internal/handlers/fetcher_internal_test.go

@@ -2,8 +2,10 @@ package handlers
 
 import (
 	"net/http"
+	"net/netip"
 	"path/filepath"
 	"testing"
+	"time"
 
 	"github.com/go-chi/chi/v5"
 	"go.uber.org/fx"
@@ -18,17 +20,27 @@ import (
 
 // TestHandlersBuildTheirOwnFetcherWhenNoneIsProvided builds the handlers as
 // pixad does, in an fx app that provides no fetcher, and requests an image
-// from localhost, which is on the allowlist. The fetcher the handlers build
-// from the config refuses localhost, so the answer is 403.
+// from 192.0.2.10, which is on the allowlist and in blocked_networks. The URL
+// check accepts that address; only the dialer that refuses internal
+// addresses checks blocked_networks, so the answer is 403 only if the
+// fetcher the handlers build from the config connects with that dialer. Any
+// other dialer would try to connect until the upstream fetch timeout, which
+// is short so that the test then fails quickly.
 func TestHandlersBuildTheirOwnFetcherWhenNoneIsProvided(t *testing.T) {
 	t.Parallel()
 
+	const host = "192.0.2.10"
+
 	stateDir := t.TempDir()
 	cfg := &config.Config{
-		SigningKey:     testSigningKey,
-		StateDir:       stateDir,
-		DBURL:          "file:" + filepath.Join(stateDir, "state.sqlite3"),
-		AllowlistHosts: []string{"localhost"},
+		SigningKey:           testSigningKey,
+		StateDir:             stateDir,
+		DBURL:                "file:" + filepath.Join(stateDir, "state.sqlite3"),
+		AllowlistHosts:       []string{host},
+		BlockedNetworks:      []netip.Prefix{netip.MustParsePrefix("192.0.2.0/24")},
+		UpstreamFetchTimeout: 2 * time.Second,
+		// With no connection slots, the fetch would fail before dialing.
+		UpstreamConnections: config.DefaultUpstreamConnections,
 	}
 
 	var h *Handlers
@@ -44,6 +56,6 @@ func TestHandlersBuildTheirOwnFetcherWhenNoneIsProvided(t *testing.T) {
 	r := chi.NewRouter()
 	r.Get("/v1/image/*", h.HandleImage())
 
-	rec := sendGet(t, r, photoURL("localhost"))
+	rec := sendGet(t, r, photoURL(host))
 	checkErrorBody(t, rec, http.StatusForbidden, "forbidden")
 }

internal/httpfetcher/httpfetcher.go

@@ -195,9 +195,10 @@ func New(config *Config) *HTTPFetcher {
 		config = DefaultConfig()
 	}
 
-	// Create transport with SSRF-safe dialer. The dialer re-resolves and
-	// re-checks at connect time (closing the DNS-rebinding window) against
-	// both the built-in ranges and the operator-supplied blocklist.
+	// Unless config.DialContext replaces it, the transport connects with
+	// the SSRF-safe dialer, which re-resolves and re-checks at connect time
+	// (closing the DNS-rebinding window) against both the built-in ranges
+	// and the operator-supplied blocklist.
 	dialContext := config.DialContext
 	if dialContext == nil {
 		dialContext = func(ctx context.Context, network, addr string) (net.Conn, error) {

internal/imgcache/eviction_internal_test.go

@@ -847,15 +847,28 @@ func TestPeriodicReconciliationAdoptsFileThatAppearsAfterStartup(t *testing.T) {
 
 	cache, _ := newEvictionTestCache(t, 1<<30)
 
-	const interval = 100 * time.Millisecond
+	// Hold the test database's only connection, so the startup pass
+	// waits for it after walking the still empty variant directory: the
+	// file written while it waits is first seen by a periodic pass.
+	conn, err := cache.db.Conn(t.Context())
+	if err != nil {
+		t.Fatalf("failed to take the database connection: %v", err)
+	}
 
-	cache.StartEviction(interval)
+	defer func() { _ = conn.Close() }()
+
+	cache.StartEviction(100 * time.Millisecond)
 	defer func() { _ = cache.StopEviction(t.Context()) }()
 
-	// Let startup reconciliation run and settle on an empty cache
-	// before introducing the untracked file, so the adoption we assert
-	// below can only be the work of a later, periodic pass.
-	time.Sleep(3 * interval)
+	deadline := time.Now().Add(5 * time.Second)
+
+	for cache.db.Stats().WaitCount == 0 {
+		if time.Now().After(deadline) {
+			t.Fatal("the startup pass never waited for the database")
+		}
+
+		time.Sleep(10 * time.Millisecond)
+	}
 
 	// Simulate a variant whose accounting insert failed after the
 	// process was already running and serving requests: the content
@@ -864,14 +877,16 @@ func TestPeriodicReconciliationAdoptsFileThatAppearsAfterStartup(t *testing.T) {
 	// insert had failed and only the file write had succeeded.
 	untracked := bytes.Repeat([]byte{0x41}, 900)
 
-	_, err := cache.variants.Store(
+	_, err = cache.variants.Store(
 		"aabbccdd0099", bytes.NewReader(untracked), "image/webp",
 	)
 	if err != nil {
 		t.Fatalf("failed to store untracked variant file: %v", err)
 	}
 
-	deadline := time.Now().Add(5 * time.Second)
+	_ = conn.Close()
+
+	deadline = time.Now().Add(5 * time.Second)
 
 	var usage int64
 

internal/imgcache/service.go

@@ -42,7 +42,8 @@ type Service struct {
 type ServiceConfig struct {
 	// Cache is the cache instance
 	Cache *Cache
-	// FetcherConfig configures the upstream fetcher (ignored if Fetcher is set)
+	// FetcherConfig configures the upstream fetcher built when Fetcher is
+	// not set. Its AllowHTTP and MaxResponseSize are used either way.
 	FetcherConfig *httpfetcher.Config
 	// Fetcher is an optional custom fetcher (for testing)
 	Fetcher httpfetcher.Fetcher