Run database maintenance under the database lock (closes #94)

The database maintainer ran Vacuum and Analyze without the database lock
every batch write holds, so on the live feed every incremental vacuum
failed with "database is locked", and an hourly ANALYZE made a batch write
fail and lose its updates. Vacuum, Analyze and Checkpoint now take that
lock, so maintenance and batch writes wait for each other. Analyze runs
PRAGMA optimize with an analysis limit, which analyzes only tables whose
size changed a lot, so it holds the lock briefly however large the file.

Model: opus-5-5

internal/database/database_internal_test.go

@@ -5,6 +5,7 @@ import (
 	"database/sql"
 	"errors"
 	"fmt"
+	"log/slog"
 	"net/netip"
 	"slices"
 	"sync"
@@ -47,6 +48,23 @@ const (
 	asnSecondBand = 100
 )
 
+// Parameters for the tests of maintenance and the database lock.
+const (
+	// lockMessageBuffer is how many lock messages a test database's logger
+	// holds before logging waits for the test to read them.
+	lockMessageBuffer = 64
+	// operationKey is the log attribute that names the operation taking the
+	// lock.
+	operationKey = "operation"
+	// testBatchWrite is the operation name the tests take the lock under.
+	testBatchWrite = "test batch write"
+	// lockWaitTimeout is the context timeout of a maintenance call that the
+	// tests let run out while it waits for the lock.
+	lockWaitTimeout = time.Millisecond
+	// lockTestASN is the AS the batch write in the lock tests stores.
+	lockTestASN = 64500
+)
+
 // Parameters for the prefix-length sample tests.
 const (
 	// sampleLimit is how many prefixes the tests ask the sample for.
@@ -474,14 +492,10 @@ func TestVacuumReturnsFreePages(t *testing.T) {
 	}
 }
 
-// TestBatchWriteDuringCheckpoint reproduces issue #25. A batch write reads
-// (SELECT) before it writes (INSERT/UPDATE). Under the default deferred locking
-// the transaction begins as a reader and, when the maintainer's WAL checkpoint
-// holds the write lock, its upgrade to writer fails immediately with "database
-// is locked" without honouring busy_timeout, dropping the batch. With
-// _txlock=immediate the transaction takes the write lock at BEGIN and waits, so
-// no batch is dropped. The checkpoint runs without the Database mutex, exactly
-// as the background maintainer does in production.
+// TestBatchWriteDuringCheckpoint runs checkpoints back to back while batch
+// writes run and checks that every batch write succeeds: a checkpoint holds the
+// database lock as a batch write does, and checkpoints must not keep batch
+// writes from getting it.
 func TestBatchWriteDuringCheckpoint(t *testing.T) {
 	t.Parallel()
 
@@ -530,6 +544,210 @@ func TestBatchWriteDuringCheckpoint(t *testing.T) {
 	wg.Wait()
 }
 
+// lockMessage is a debug message the database lock logged, and the name of the
+// operation that logged it.
+type lockMessage struct {
+	text      string
+	operation string
+}
+
+// lockMessages is a log handler that passes on each message the database lock
+// logs. The lock tests read it to know that an operation waits for the lock or
+// holds it, instead of guessing how long that takes.
+type lockMessages chan lockMessage
+
+// Enabled lets every message through, so the lock's debug messages arrive.
+func (m lockMessages) Enabled(context.Context, slog.Level) bool { return true }
+
+// Handle passes on a message that names an operation and drops the others.
+func (m lockMessages) Handle(_ context.Context, record slog.Record) error {
+	record.Attrs(func(attr slog.Attr) bool {
+		if attr.Key != operationKey {
+			return true
+		}
+
+		m <- lockMessage{text: record.Message, operation: attr.Value.String()}
+
+		return false
+	})
+
+	return nil
+}
+
+// WithAttrs returns m: the tests need no attributes added.
+func (m lockMessages) WithAttrs([]slog.Attr) slog.Handler { return m }
+
+// WithGroup returns m: the tests need no groups.
+func (m lockMessages) WithGroup(string) slog.Handler { return m }
+
+// newDatabaseWithLockMessages opens a new database whose logger passes the lock's
+// messages to the returned channel.
+func newDatabaseWithLockMessages(t *testing.T) (*Database, lockMessages) {
+	t.Helper()
+
+	messages := make(lockMessages, lockMessageBuffer)
+	cfg := &config.Config{StateDir: t.TempDir()}
+
+	db, err := New(cfg, &logger.Logger{Logger: slog.New(messages)})
+	if err != nil {
+		t.Fatalf("failed to create database: %v", err)
+	}
+
+	t.Cleanup(func() { _ = db.Close() })
+
+	return db, messages
+}
+
+// waitForLockMessage waits until operation logs text, and fails the test if
+// returned delivers first: the call running operation returned without logging
+// text.
+func waitForLockMessage(
+	t *testing.T,
+	messages lockMessages,
+	text, operation string,
+	returned <-chan error,
+) {
+	t.Helper()
+
+	for {
+		select {
+		case message := <-messages:
+			if message.text == text && message.operation == operation {
+				return
+			}
+		case err := <-returned:
+			t.Fatalf("%s returned (error: %v) without logging %q",
+				operation, err, text)
+		}
+	}
+}
+
+// TestVacuumWaitsForBatchWrite holds the database lock as a batch write does and
+// checks that Vacuum waits for it and then succeeds. Vacuum ran without the
+// lock, met the batch writes at SQLite's write lock and failed with "database
+// is locked" (https://git.eeqj.de/sneak/routewatch/issues/94).
+func TestVacuumWaitsForBatchWrite(t *testing.T) {
+	t.Parallel()
+
+	db, messages := newDatabaseWithLockMessages(t)
+
+	db.lock(testBatchWrite)
+
+	vacuumed := make(chan error, 1)
+
+	go func() { vacuumed <- db.Vacuum(context.Background()) }()
+
+	waitForLockMessage(t, messages, lockWaitMessage, "Vacuum", vacuumed)
+
+	db.unlock()
+
+	err := <-vacuumed
+	if err != nil {
+		t.Fatalf("Vacuum after the batch write: %v", err)
+	}
+}
+
+// TestBatchWriteWaitsForAnalyze keeps Analyze waiting for SQLite's write lock,
+// which another connection holds, and checks that a batch write started
+// meanwhile waits for Analyze to give up the database lock, that both then
+// succeed, and that Analyze stored statistics. ANALYZE ran without the database
+// lock, and a batch write that waited longer than the busy timeout for it failed
+// with "database is locked".
+func TestBatchWriteWaitsForAnalyze(t *testing.T) {
+	t.Parallel()
+
+	db, messages := newDatabaseWithLockMessages(t)
+	ctx := context.Background()
+	asns := map[int]time.Time{lockTestASN: time.Now().UTC()}
+
+	// A stored AS gives ANALYZE a table to analyze and statistics to write.
+	err := db.GetOrCreateASNBatch(asns)
+	if err != nil {
+		t.Fatalf("failed to store an AS: %v", err)
+	}
+
+	// With _txlock=immediate this transaction takes SQLite's write lock at
+	// BEGIN, so ANALYZE has to wait for it while Analyze holds the database lock.
+	writer, err := db.db.BeginTx(ctx, nil)
+	if err != nil {
+		t.Fatalf("failed to begin the transaction ANALYZE waits for: %v", err)
+	}
+
+	defer func() { _ = writer.Rollback() }()
+
+	analyzed := make(chan error, 1)
+
+	go func() { analyzed <- db.Analyze(ctx) }()
+
+	waitForLockMessage(t, messages, lockHeldMessage, "Analyze", analyzed)
+
+	written := make(chan error, 1)
+
+	go func() { written <- db.GetOrCreateASNBatch(asns) }()
+
+	waitForLockMessage(t, messages, lockWaitMessage, "GetOrCreateASNBatch", written)
+
+	err = writer.Rollback()
+	if err != nil {
+		t.Fatalf("failed to end the transaction ANALYZE waits for: %v", err)
+	}
+
+	err = <-analyzed
+	if err != nil {
+		t.Errorf("Analyze: %v", err)
+	}
+
+	err = <-written
+	if err != nil {
+		t.Errorf("batch write after Analyze: %v", err)
+	}
+
+	var statistics int
+
+	err = db.db.QueryRowContext(ctx,
+		"SELECT count(*) FROM sqlite_stat1 WHERE tbl = 'asns'").Scan(&statistics)
+	if err != nil {
+		t.Fatalf("failed to read the statistics: %v", err)
+	}
+
+	if statistics == 0 {
+		t.Error("Analyze stored no statistics for the asns table")
+	}
+}
+
+// TestCheckpointLockWaitEndsWithContext holds the database lock while the
+// context of a waiting Checkpoint times out, and checks that Checkpoint then
+// returns the context's error and leaves the lock free.
+func TestCheckpointLockWaitEndsWithContext(t *testing.T) {
+	t.Parallel()
+
+	db, messages := newDatabaseWithLockMessages(t)
+
+	db.lock(testBatchWrite)
+
+	ctx, cancel := context.WithTimeout(context.Background(), lockWaitTimeout)
+	defer cancel()
+
+	checkpointed := make(chan error, 1)
+
+	go func() { checkpointed <- db.Checkpoint(ctx) }()
+
+	waitForLockMessage(t, messages, lockWaitMessage, "Checkpoint", checkpointed)
+	<-ctx.Done()
+	db.unlock()
+
+	err := <-checkpointed
+	if !errors.Is(err, context.DeadlineExceeded) {
+		t.Fatalf("Checkpoint = %v, want %v", err, context.DeadlineExceeded)
+	}
+
+	if !db.mu.TryLock() {
+		t.Fatal("Checkpoint kept the database lock after its context ended")
+	}
+
+	db.mu.Unlock()
+}
+
 // TestGetRandomPrefixesByLengthListsPrefixOnce stores one prefix of each
 // family learned from two peers in the same second and checks that the
 // prefix-length sample lists each of them once.