hold: set busy_timeout on every pooled connection, not just the first

A PRAGMA is per-connection, and database/sql opens one connection per
concurrent caller. hold_db.go and scan_broadcaster.go each set
busy_timeout = 5000 with a one-shot query against the pool, which configures
whichever connection happened to serve it and leaves every other writer at
zero. Probing the live pools showed exactly that: conn 0 at 5000, conns 1 and 2
at 0. That is the database is locked on COMMIT, and the carstore had no
busy_timeout and no WAL on any connection at all.

Opening a database now goes through a connector that runs the PRAGMA on every
connection the pool creates, ported from the appview, which had already solved
this. Applied to the carstore, the shared hold DB, the records index, the event
broadcaster and the scan broadcaster. In-memory databases are left unwrapped,
since libsql gives each connection its own, and the embedded replica path still
skips PRAGMAs because a remote rejects them.

WAL is not a new risk: production was already WAL, because journal mode is a
persistent property of the file and OpenHoldDB has been setting it. This makes
the carstore's own opener agree rather than leaving standalone and test
databases in rollback-journal mode. A database that refuses WAL logs a warning
instead of failing boot, so a network mount cannot stop the hold starting.

NewHoldPDS's file mode opened the same file a second time for the records
index; it now shares the carstore's pool. Two pools on one file buy nothing,
since SQLite's write lock is per database, so the second only adds contenders.
Production already shared a pool and was unaffected by that half.

Note for anyone reading OpenHoldDB: its foreign_keys pragma has the same
one-connection scope, but libsql reports foreign_keys on for every fresh
connection anyway, so it is decorative rather than broken.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01U1Km3N3uUmeGaj7VbaM8PF

pkg/hold/db/hold_db.go

@@ -5,7 +5,6 @@ import (
 	"fmt"
 	"io"
 	"log/slog"
-	"strings"
 	"time"
 
 	"github.com/tursodatabase/go-libsql"
@@ -64,34 +63,19 @@ func OpenHoldDB(path string, cfg LibsqlConfig) (*HoldDB, error) {
 		connector = conn
 		slog.Info("Hold database opened in embedded replica mode", "path", path, "sync_url", cfg.SyncURL)
 	} else {
-		// Local-only mode: plain file via libsql driver
-		dsn := path
-		if !strings.HasPrefix(path, "file:") && !strings.HasPrefix(path, ":memory:") {
-			dsn = "file:" + path
-		}
+		// Local-only mode: WAL journal mode plus a busy_timeout on *every*
+		// pooled connection. In embedded replica mode the remote server manages
+		// these settings and PRAGMA assignments are rejected as "unsupported
+		// statement" (observed with Bunny Database), so OpenLocalDB is only for
+		// the local-only branch.
 		var err error
-		db, err = sql.Open("libsql", dsn)
+		db, err = OpenLocalDB(path)
 		if err != nil {
 			return nil, fmt.Errorf("failed to open hold database: %w", err)
 		}
 		slog.Info("Hold database opened in local-only mode", "path", path)
 	}
 
-	// In local-only mode, configure WAL and busy_timeout locally.
-	// In embedded replica mode, the remote server manages these settings
-	// and PRAGMA assignments are rejected as "unsupported statement"
-	// (observed with Bunny Database).
-	if cfg.SyncURL == "" {
-		var journalMode string
-		if err := db.QueryRow("PRAGMA journal_mode = WAL").Scan(&journalMode); err != nil {
-			return nil, fmt.Errorf("failed to set journal mode: %w", err)
-		}
-		var busyTimeout int
-		if err := db.QueryRow("PRAGMA busy_timeout = 5000").Scan(&busyTimeout); err != nil {
-			return nil, fmt.Errorf("failed to set busy_timeout: %w", err)
-		}
-	}
-
 	// Foreign keys work in both modes
 	if _, err := db.Exec("PRAGMA foreign_keys = ON"); err != nil {
 		return nil, fmt.Errorf("failed to enable foreign keys: %w", err)

pkg/hold/db/libsql_open.go

@@ -0,0 +1,131 @@
+package db
+
+import (
+	"context"
+	"database/sql"
+	"database/sql/driver"
+	"fmt"
+	"log/slog"
+	"strings"
+
+	_ "github.com/tursodatabase/go-libsql"
+)
+
+// DefaultBusyTimeoutMs is how long a connection waits for a lock before giving
+// up with SQLITE_BUSY. Matches the appview and labeler databases.
+const DefaultBusyTimeoutMs = 5000
+
+// localDSN normalizes a filesystem path into a libsql DSN.
+func localDSN(path string) string {
+	if strings.HasPrefix(path, ":memory:") || strings.HasPrefix(path, "file:") {
+		return path
+	}
+	return "file:" + path
+}
+
+// OpenLocalDB opens a local (non-replica) libsql database with WAL journal mode
+// and PRAGMA busy_timeout applied to every pooled connection.
+//
+// Every subsystem that opens a hold database file directly must go through
+// this. database/sql opens connections lazily and on demand, and SQLite's
+// busy_timeout is per-connection, so running the PRAGMA once against the pool
+// only configures whichever connection happened to serve it. Every other
+// connection commits with busy_timeout=0 and fails immediately the moment
+// another writer holds the lock.
+//
+// ":memory:" databases are returned unwrapped: libsql gives each connection its
+// own private in-memory database, so there is no cross-connection lock to wait
+// on and WAL is meaningless.
+func OpenLocalDB(path string) (*sql.DB, error) {
+	dsn := localDSN(path)
+
+	if strings.HasPrefix(dsn, ":memory:") {
+		return sql.Open("libsql", dsn)
+	}
+
+	base, err := openLibsqlLocalConnector(dsn)
+	if err != nil {
+		return nil, err
+	}
+	db := sql.OpenDB(&busyTimeoutConnector{base: base, timeoutMs: DefaultBusyTimeoutMs})
+
+	if err := applyWAL(db, dsn); err != nil {
+		_ = db.Close()
+		return nil, err
+	}
+	return db, nil
+}
+
+// applyWAL puts the database file into WAL journal mode. Unlike busy_timeout,
+// journal mode is a property of the file itself, so one call covers every
+// connection and every other process that opens the same file.
+func applyWAL(db *sql.DB, dsn string) error {
+	var journalMode string
+	// libsql treats PRAGMA assignments as queries that return a row.
+	if err := db.QueryRow("PRAGMA journal_mode = WAL").Scan(&journalMode); err != nil {
+		return fmt.Errorf("failed to set journal mode: %w", err)
+	}
+	if !strings.EqualFold(journalMode, "wal") {
+		// Some filesystems (notably network mounts) cannot support WAL. Carry
+		// on rather than refusing to boot: busy_timeout still applies, and the
+		// rollback journal is correct, just less concurrent.
+		slog.Warn("Database did not accept WAL journal mode", "dsn", dsn, "journal_mode", journalMode)
+	}
+	return nil
+}
+
+// openLibsqlLocalConnector returns a driver.Connector for a local libsql DSN.
+// go-libsql exports NewEmbeddedReplicaConnector for replica mode but no public
+// constructor for local files, so we obtain the driver via a probe sql.Open
+// (which is lazy and opens no connection) and ask it for a Connector.
+func openLibsqlLocalConnector(dsn string) (driver.Connector, error) {
+	probe, err := sql.Open("libsql", dsn)
+	if err != nil {
+		return nil, fmt.Errorf("probe libsql driver: %w", err)
+	}
+	drv := probe.Driver()
+	_ = probe.Close()
+
+	dctx, ok := drv.(driver.DriverContext)
+	if !ok {
+		return nil, fmt.Errorf("libsql driver does not implement driver.DriverContext")
+	}
+	return dctx.OpenConnector(dsn)
+}
+
+// busyTimeoutConnector wraps a driver.Connector and runs PRAGMA busy_timeout on
+// every newly opened connection. SQLite's busy_timeout is per-connection, so
+// this is the only way to ensure every conn in the pool waits on lock
+// contention instead of returning SQLITE_BUSY immediately.
+type busyTimeoutConnector struct {
+	base      driver.Connector
+	timeoutMs int
+}
+
+func (c *busyTimeoutConnector) Connect(ctx context.Context) (driver.Conn, error) {
+	conn, err := c.base.Connect(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	// libsql treats PRAGMA assignments as queries that return a row, so we
+	// must use QueryerContext rather than ExecerContext.
+	queryer, ok := conn.(driver.QueryerContext)
+	if !ok {
+		_ = conn.Close()
+		return nil, fmt.Errorf("libsql conn does not support QueryerContext")
+	}
+
+	rows, err := queryer.QueryContext(ctx, fmt.Sprintf("PRAGMA busy_timeout = %d", c.timeoutMs), nil)
+	if err != nil {
+		_ = conn.Close()
+		return nil, fmt.Errorf("set busy_timeout on new conn: %w", err)
+	}
+	_ = rows.Close()
+
+	return conn, nil
+}
+
+func (c *busyTimeoutConnector) Driver() driver.Driver {
+	return c.base.Driver()
+}

pkg/hold/db/libsql_open_test.go

@@ -0,0 +1,93 @@
+package db
+
+import (
+	"context"
+	"database/sql"
+	"path/filepath"
+	"testing"
+)
+
+// assertPoolSettings checks journal mode and busy_timeout on several
+// simultaneously held connections. Holding them at once forces database/sql to
+// open distinct connections: busy_timeout is per-connection, so a regression to
+// a one-shot "PRAGMA busy_timeout" on the pool shows up on conn 1 onward.
+func assertPoolSettings(t *testing.T, label string, db *sql.DB, wantJournal string, wantBusy int) {
+	t.Helper()
+	ctx := context.Background()
+
+	const n = 4
+	conns := make([]*sql.Conn, 0, n)
+	defer func() {
+		for _, c := range conns {
+			_ = c.Close()
+		}
+	}()
+	for i := 0; i < n; i++ {
+		c, err := db.Conn(ctx)
+		if err != nil {
+			t.Fatalf("%s: open conn %d: %v", label, i, err)
+		}
+		conns = append(conns, c)
+	}
+
+	for i, c := range conns {
+		var journal string
+		if err := c.QueryRowContext(ctx, "PRAGMA journal_mode").Scan(&journal); err != nil {
+			t.Fatalf("%s conn %d: read journal_mode: %v", label, i, err)
+		}
+		if journal != wantJournal {
+			t.Errorf("%s conn %d: journal_mode = %q, want %q", label, i, journal, wantJournal)
+		}
+
+		var busy int
+		if err := c.QueryRowContext(ctx, "PRAGMA busy_timeout").Scan(&busy); err != nil {
+			t.Fatalf("%s conn %d: read busy_timeout: %v", label, i, err)
+		}
+		if busy != wantBusy {
+			t.Errorf("%s conn %d: busy_timeout = %d, want %d", label, i, busy, wantBusy)
+		}
+	}
+}
+
+func TestOpenLocalDBSettingsOnEveryConnection(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "db.sqlite3")
+	db, err := OpenLocalDB(path)
+	if err != nil {
+		t.Fatalf("OpenLocalDB: %v", err)
+	}
+	defer db.Close()
+	assertPoolSettings(t, "OpenLocalDB", db, "wal", DefaultBusyTimeoutMs)
+}
+
+func TestOpenHoldDBSettingsOnEveryConnection(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "db.sqlite3")
+	h, err := OpenHoldDB(path, LibsqlConfig{})
+	if err != nil {
+		t.Fatalf("OpenHoldDB: %v", err)
+	}
+	defer h.Close()
+	assertPoolSettings(t, "OpenHoldDB", h.DB, "wal", DefaultBusyTimeoutMs)
+}
+
+func TestCarstoreSettingsOnEveryConnection(t *testing.T) {
+	sqs, err := NewSqliteStore(t.TempDir())
+	if err != nil {
+		t.Fatalf("NewSqliteStore: %v", err)
+	}
+	defer sqs.Close()
+	assertPoolSettings(t, "carstore", sqs.DB(), "wal", DefaultBusyTimeoutMs)
+}
+
+// TestOpenLocalDBMemory documents that ":memory:" is left unwrapped: libsql
+// gives each connection its own private database, so there is no shared lock to
+// wait on and WAL does not apply.
+func TestOpenLocalDBMemory(t *testing.T) {
+	db, err := OpenLocalDB(":memory:")
+	if err != nil {
+		t.Fatalf("OpenLocalDB(:memory:): %v", err)
+	}
+	defer db.Close()
+	if err := db.Ping(); err != nil {
+		t.Fatalf("ping in-memory db: %v", err)
+	}
+}

pkg/hold/db/sqlite_store.go

@@ -23,7 +23,6 @@ import (
 	"log/slog"
 	"os"
 	"path/filepath"
-	"strings"
 
 	"go.opentelemetry.io/otel/attribute"
 
@@ -106,15 +105,12 @@ func (sqs *SQLiteStore) Open(path string) error {
 	}
 	sqs.log.Debug("open db", "path", path)
 
-	// Build DSN for go-libsql
-	dsn := path
-	if path == ":memory:" {
-		dsn = ":memory:"
-	} else if !strings.HasPrefix(path, "file:") {
-		dsn = "file:" + path
-	}
-
-	db, err := sql.Open("libsql", dsn)
+	// OpenLocalDB applies WAL journal mode and a per-connection busy_timeout.
+	// Without them, two concurrent writers (the carstore pool serves one
+	// connection per concurrent caller, and the records index, events and scan
+	// broadcaster may share or shadow this same file) fail their COMMIT with
+	// "database is locked" instead of waiting for the lock.
+	db, err := OpenLocalDB(path)
 	if err != nil {
 		return fmt.Errorf("%s: sqlite could not open, %w", path, err)
 	}
@@ -466,6 +462,13 @@ func (sqs *SQLiteStore) Close() error {
 	return nil
 }
 
+// DB returns the underlying database handle so callers that need a second
+// logical table on the same file (the records index, for one) can share this
+// pool instead of opening the file again.
+func (sqs *SQLiteStore) DB() *sql.DB {
+	return sqs.db
+}
+
 func (sqs *SQLiteStore) getBlock(ctx context.Context, user models.Uid, bcid cid.Cid) (blockformat.Block, error) {
 	tx, err := sqs.db.BeginTx(ctx, &txReadOnly)
 	if err != nil {

pkg/hold/db/sqlite_store_contention_test.go

@@ -0,0 +1,160 @@
+package db
+
+import (
+	"context"
+	"fmt"
+	"path/filepath"
+	"strings"
+	"sync"
+	"testing"
+
+	"github.com/bluesky-social/indigo/models"
+	blockformat "github.com/ipfs/go-block-format"
+	"github.com/ipfs/go-cid"
+)
+
+// isLockErr reports whether err is a SQLite busy/locked error rather than a
+// logic error. Contention tests must fail on lock errors specifically, so a
+// typo cannot masquerade as a reproduction.
+func isLockErr(err error) bool {
+	if err == nil {
+		return false
+	}
+	s := strings.ToLower(err.Error())
+	return strings.Contains(s, "database is locked") ||
+		strings.Contains(s, "database table is locked") ||
+		strings.Contains(s, "sqlite_busy") ||
+		strings.Contains(s, "busy")
+}
+
+func makeBlocks(t *testing.T, n int, salt string) (cid.Cid, map[cid.Cid]blockformat.Block) {
+	t.Helper()
+	blks := make(map[cid.Cid]blockformat.Block, n)
+	var root cid.Cid
+	for i := 0; i < n; i++ {
+		b := blockformat.NewBlock([]byte(fmt.Sprintf("%s-block-%d-%s", salt, i, strings.Repeat("x", 2048))))
+		blks[b.Cid()] = b
+		if i == 0 {
+			root = b.Cid()
+		}
+	}
+	return root, blks
+}
+
+// writeShards hammers the carstore with concurrent shard writes and reports any
+// lock error it hits. This is the exact path every hold record write takes.
+func writeShards(t *testing.T, sqs *SQLiteStore, writers, iters int) error {
+	t.Helper()
+	ctx := context.Background()
+	var wg sync.WaitGroup
+	var mu sync.Mutex
+	var firstErr error
+
+	for w := 0; w < writers; w++ {
+		wg.Add(1)
+		go func(w int) {
+			defer wg.Done()
+			for i := 0; i < iters; i++ {
+				salt := fmt.Sprintf("w%d-i%d", w, i)
+				root, blks := makeBlocks(t, 12, salt)
+				rev := fmt.Sprintf("rev-%03d-%03d", w, i)
+				_, err := sqs.writeNewShard(ctx, root, rev, models.Uid(1), i, blks, nil)
+				if err != nil {
+					mu.Lock()
+					if firstErr == nil {
+						firstErr = err
+					}
+					mu.Unlock()
+					return
+				}
+			}
+		}(w)
+	}
+	wg.Wait()
+	return firstErr
+}
+
+// TestCarstoreConcurrentWritesSharedPool reproduces the production topology:
+// one *sql.DB pool (OpenHoldDB) shared by the carstore, records index, events
+// and scan broadcaster. database/sql opens a fresh connection per concurrent
+// caller, and busy_timeout is per-connection, so any connection beyond the
+// first commits with busy_timeout=0 and fails immediately under contention.
+func TestCarstoreConcurrentWritesSharedPool(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "db.sqlite3")
+	h, err := OpenHoldDB(path, LibsqlConfig{})
+	if err != nil {
+		t.Fatalf("OpenHoldDB: %v", err)
+	}
+	defer h.Close()
+
+	sqs, err := NewSQLiteStoreWithDB(path, h.DB)
+	if err != nil {
+		t.Fatalf("NewSQLiteStoreWithDB: %v", err)
+	}
+
+	if err := writeShards(t, sqs, 8, 25); err != nil {
+		if isLockErr(err) {
+			t.Fatalf("shared-pool carstore write hit a lock error: %v", err)
+		}
+		t.Fatalf("shared-pool carstore write failed: %v", err)
+	}
+}
+
+// TestCarstoreConcurrentWritesTwoOpeners covers the topology where a second
+// subsystem (the scan broadcaster or the event broadcaster, when they are not
+// handed the shared pool) opens <dir>/db.sqlite3 again through its own pool
+// while the carstore is writing.
+func TestCarstoreConcurrentWritesTwoOpeners(t *testing.T) {
+	dir := t.TempDir()
+	sqs, err := NewSqliteStore(dir)
+	if err != nil {
+		t.Fatalf("NewSqliteStore: %v", err)
+	}
+	defer sqs.Close()
+
+	// Second, independent pool on the same file, opened the way every hold
+	// subsystem now opens one.
+	ri, err := OpenLocalDB(filepath.Join(dir, "db.sqlite3"))
+	if err != nil {
+		t.Fatalf("open records index: %v", err)
+	}
+	defer ri.Close()
+	if _, err := ri.Exec("CREATE TABLE IF NOT EXISTS records (collection TEXT, rkey TEXT, cid TEXT, PRIMARY KEY (collection, rkey))"); err != nil {
+		t.Fatalf("create records table: %v", err)
+	}
+
+	ctx := context.Background()
+	stop := make(chan struct{})
+	idxErr := make(chan error, 1)
+	go func() {
+		var err error
+		for i := 0; ; i++ {
+			select {
+			case <-stop:
+				idxErr <- err
+				return
+			default:
+			}
+			_, e := ri.ExecContext(ctx,
+				"INSERT INTO records (collection, rkey, cid) VALUES (?, ?, ?) ON CONFLICT (collection, rkey) DO UPDATE SET cid=excluded.cid",
+				"io.atcr.hold.scan", fmt.Sprintf("rkey-%d", i), fmt.Sprintf("cid-%d", i))
+			if e != nil && err == nil {
+				err = e
+			}
+		}
+	}()
+
+	csErr := writeShards(t, sqs, 4, 30)
+	close(stop)
+	riResult := <-idxErr
+
+	for _, e := range []error{csErr, riResult} {
+		if e == nil {
+			continue
+		}
+		if isLockErr(e) {
+			t.Fatalf("two-opener write hit a lock error: %v", e)
+		}
+		t.Fatalf("two-opener write failed: %v", e)
+	}
+}

pkg/hold/pds/carstore_contention_test.go

@@ -0,0 +1,168 @@
+package pds
+
+import (
+	"context"
+	"fmt"
+	"path/filepath"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"atcr.io/pkg/atproto"
+	holddb "atcr.io/pkg/hold/db"
+)
+
+// isLockError reports whether err is a SQLite busy/locked error rather than a
+// logic error, so a contention test cannot pass off a typo as a reproduction.
+func isLockError(err error) bool {
+	if err == nil {
+		return false
+	}
+	s := strings.ToLower(err.Error())
+	return strings.Contains(s, "database is locked") ||
+		strings.Contains(s, "database table is locked") ||
+		strings.Contains(s, "sqlite_busy")
+}
+
+// TestHoldPDSConcurrentRepoWritesAndSideTable reproduces the production shape of
+// the "database is locked on COMMIT" failure: repo record writes (which land in
+// the carstore) racing against repo reads and against a second subsystem
+// writing its own table in the same database file on a separate goroutine, the
+// way the scan broadcaster now does.
+func TestHoldPDSConcurrentRepoWritesAndSideTable(t *testing.T) {
+	ctx := context.Background()
+	tmpDir := t.TempDir()
+	dbPath := filepath.Join(tmpDir, "pds.db")
+	keyPath := filepath.Join(tmpDir, "signing-key")
+
+	p, err := NewHoldPDS(ctx, "did:web:hold.example.com", "https://hold.example.com", "https://atcr.io", dbPath, keyPath, false)
+	if err != nil {
+		t.Fatalf("NewHoldPDS failed: %v", err)
+	}
+	defer p.Close()
+
+	if err := p.repomgr.InitNewActor(ctx, p.uid, "hold.example.com", p.did, "", "", ""); err != nil {
+		t.Fatalf("InitNewActor failed: %v", err)
+	}
+
+	// A second subsystem with its own pool on the same file, as the scan and
+	// event broadcasters have when they are not handed the shared pool.
+	side, err := holddb.OpenLocalDB(filepath.Join(dbPath, "db.sqlite3"))
+	if err != nil {
+		t.Fatalf("open side pool: %v", err)
+	}
+	defer side.Close()
+	if _, err := side.Exec("CREATE TABLE IF NOT EXISTS scan_jobs (id TEXT PRIMARY KEY, state TEXT)"); err != nil {
+		t.Fatalf("create scan_jobs: %v", err)
+	}
+
+	var mu sync.Mutex
+	var errs []error
+	record := func(err error) {
+		if err == nil {
+			return
+		}
+		mu.Lock()
+		errs = append(errs, err)
+		mu.Unlock()
+	}
+
+	const writers = 4
+	const iters = 15
+
+	var writeWG, loopWG sync.WaitGroup
+	stop := make(chan struct{})
+
+	// Repo record writers.
+	for w := 0; w < writers; w++ {
+		writeWG.Add(1)
+		go func(w int) {
+			defer writeWG.Done()
+			for i := 0; i < iters; i++ {
+				rkey := fmt.Sprintf("w%d-layer-%03d", w, i)
+				rec := atproto.NewLayerRecord(
+					fmt.Sprintf("sha256:%064x", w*1000+i),
+					int64(1024+i),
+					"application/vnd.oci.image.layer.v1.tar+gzip",
+					"did:plc:testuser",
+					"at://did:plc:testuser/io.atcr.manifest/abc",
+				)
+				if _, _, err := p.repomgr.PutRecord(ctx, p.uid, atproto.LayerCollection, rkey, rec); err != nil {
+					record(fmt.Errorf("PutRecord %s: %w", rkey, err))
+					return
+				}
+			}
+		}(w)
+	}
+
+	// Repo readers, hitting the carstore's read transactions concurrently.
+	for r := 0; r < 2; r++ {
+		loopWG.Add(1)
+		go func() {
+			defer loopWG.Done()
+			for {
+				select {
+				case <-stop:
+					return
+				default:
+				}
+				if _, err := p.repomgr.GetRepoRoot(ctx, p.uid); err != nil {
+					record(fmt.Errorf("GetRepoRoot: %w", err))
+					return
+				}
+			}
+		}()
+	}
+
+	// The side subsystem, writing its own table on its own goroutine.
+	loopWG.Add(1)
+	go func() {
+		defer loopWG.Done()
+		for i := 0; ; i++ {
+			select {
+			case <-stop:
+				return
+			default:
+			}
+			_, err := side.ExecContext(ctx,
+				"INSERT INTO scan_jobs (id, state) VALUES (?, ?) ON CONFLICT (id) DO UPDATE SET state=excluded.state",
+				fmt.Sprintf("job-%d", i), "queued")
+			if err != nil {
+				record(fmt.Errorf("scan_jobs insert: %w", err))
+				return
+			}
+		}
+	}()
+
+	writersDone := make(chan struct{})
+	go func() {
+		writeWG.Wait()
+		close(writersDone)
+	}()
+
+	select {
+	case <-writersDone:
+	case <-time.After(60 * time.Second):
+		close(stop)
+		loopWG.Wait()
+		t.Fatal("contention test timed out waiting for writers")
+	}
+	close(stop)
+	loopWG.Wait()
+
+	mu.Lock()
+	defer mu.Unlock()
+	for _, err := range errs {
+		if isLockError(err) {
+			t.Fatalf("hold PDS write hit a lock error: %v", err)
+		}
+		t.Fatalf("hold PDS write failed: %v", err)
+	}
+
+	// The repo must still be readable after all that: a failed COMMIT used to
+	// leave the next read reporting "loading root from blockstore: nothing to read".
+	if _, err := p.repomgr.GetRepoRoot(ctx, p.uid); err != nil {
+		t.Fatalf("repo unreadable after contention: %v", err)
+	}
+}

pkg/hold/pds/events.go

@@ -12,6 +12,7 @@ import (
 	"sync"
 	"time"
 
+	holddb "atcr.io/pkg/hold/db"
 	atproto "github.com/bluesky-social/indigo/api/atproto"
 	"github.com/bluesky-social/indigo/events"
 	lexutil "github.com/bluesky-social/indigo/lex/util"
@@ -161,12 +162,10 @@ func NewEventBroadcasterWithDB(holdDID string, maxHistory int, db *sql.DB) *Even
 
 // initDatabase opens database connection, creates table, and loads last sequence
 func (b *EventBroadcaster) initDatabase() error {
-	// Open database connection
-	dsn := b.dbPath
-	if b.dbPath != ":memory:" && !strings.HasPrefix(b.dbPath, "file:") {
-		dsn = "file:" + b.dbPath
-	}
-	db, err := sql.Open("libsql", dsn)
+	// Open database connection. holddb.OpenLocalDB applies WAL journal mode
+	// and a per-connection busy_timeout; the events table lives in the same
+	// file as the carstore, so both are required to avoid "database is locked".
+	db, err := holddb.OpenLocalDB(b.dbPath)
 	if err != nil {
 		return err
 	}

pkg/hold/pds/hold_pds.go

@@ -134,11 +134,14 @@ func NewHoldPDS(ctx context.Context, did, publicURL, appviewURL, dbPath, keyPath
 		slog.Info("New hold repo - will be initialized in Bootstrap")
 	}
 
-	// Initialize records index for efficient listing queries
-	// Uses same database as carstore for simplicity
+	// Initialize records index for efficient listing queries.
+	// It lives in the same file as the carstore, so it shares the carstore's
+	// connection pool rather than opening that file a second time: two pools on
+	// one file means two writers racing for the same lock, and SQLite's
+	// per-database write lock is not made friendlier by having two pools.
 	var recordsIndex *RecordsIndex
 	if dbPath != ":memory:" {
-		recordsIndex, err = NewRecordsIndex(dbPath + "/db.sqlite3")
+		recordsIndex, err = NewRecordsIndexWithDB(sqlStore.DB())
 		if err != nil {
 			return nil, fmt.Errorf("failed to create records index: %w", err)
 		}

pkg/hold/pds/records.go

@@ -9,6 +9,7 @@ import (
 	"strings"
 
 	"atcr.io/pkg/atproto"
+	holddb "atcr.io/pkg/hold/db"
 	"github.com/bluesky-social/indigo/repo"
 	"github.com/ipfs/go-cid"
 	_ "github.com/tursodatabase/go-libsql"
@@ -47,11 +48,11 @@ CREATE INDEX IF NOT EXISTS idx_records_collection_did ON records(collection, did
 // NewRecordsIndex creates or opens a records index
 // If the schema is outdated (missing did column), drops and rebuilds the table
 func NewRecordsIndex(dbPath string) (*RecordsIndex, error) {
-	dsn := dbPath
-	if dbPath != ":memory:" && !strings.HasPrefix(dbPath, "file:") {
-		dsn = "file:" + dbPath
-	}
-	db, err := sql.Open("libsql", dsn)
+	// holddb.OpenLocalDB applies WAL journal mode and a per-connection
+	// busy_timeout. The records index shares its file with the carstore, so
+	// without them a write here can collide with a carstore commit and fail
+	// immediately with "database is locked".
+	db, err := holddb.OpenLocalDB(dbPath)
 	if err != nil {
 		return nil, fmt.Errorf("failed to open records database: %w", err)
 	}

pkg/hold/pds/scan_broadcaster.go

@@ -13,12 +13,12 @@ import (
 	"net"
 	"net/http"
 	"net/url"
-	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
 
 	"atcr.io/pkg/atproto"
+	holddb "atcr.io/pkg/hold/db"
 	"atcr.io/pkg/s3"
 	lexutil "github.com/bluesky-social/indigo/lex/util"
 	"github.com/gorilla/websocket"
@@ -332,11 +332,12 @@ type VulnerabilitySummary struct {
 // NewScanBroadcaster creates a new scan job broadcaster
 // dbPath should point to a SQLite database file (e.g., "/path/to/pds/db.sqlite3")
 func NewScanBroadcaster(holdDID, holdEndpoint, secret string, relayEndpoints []string, dbPath string, s3svc *s3.S3Service, holdPDS *HoldPDS, rescanInterval time.Duration) (*ScanBroadcaster, error) {
-	dsn := dbPath
-	if dbPath != ":memory:" && !strings.HasPrefix(dbPath, "file:") {
-		dsn = "file:" + dbPath
-	}
-	db, err := sql.Open("libsql", dsn)
+	// holddb.OpenLocalDB sets WAL journal mode and applies busy_timeout to
+	// every pooled connection. The one-shot "PRAGMA busy_timeout = 5000" that
+	// used to live here only configured whichever connection served it, so
+	// every other connection in the pool still failed immediately on a busy
+	// lock.
+	db, err := holddb.OpenLocalDB(dbPath)
 	if err != nil {
 		return nil, fmt.Errorf("failed to open scan jobs database: %w", err)
 	}
@@ -345,18 +346,6 @@ func NewScanBroadcaster(holdDID, holdEndpoint, secret string, relayEndpoints []s
 		return nil, fmt.Errorf("failed to ping scan jobs database: %w", err)
 	}
 
-	// Set WAL mode and busy timeout (libsql PRAGMAs return rows)
-	var journalMode string
-	if err := db.QueryRow("PRAGMA journal_mode = WAL").Scan(&journalMode); err != nil {
-		db.Close()
-		return nil, fmt.Errorf("failed to set journal mode: %w", err)
-	}
-	var busyTimeout int
-	if err := db.QueryRow("PRAGMA busy_timeout = 5000").Scan(&busyTimeout); err != nil {
-		db.Close()
-		return nil, fmt.Errorf("failed to set busy_timeout: %w", err)
-	}
-
 	relayEndpoints = normalizeRelayEndpoints(relayEndpoints)
 
 	sb := &ScanBroadcaster{