Checkpoint without waiting on readers and log a busy checkpoint (closes #102)

The 5-second checkpoint ran in TRUNCATE mode, which waits up to the
5-second busy timeout for every open read to finish, and it did so
holding the database lock every batch write needs. It now runs in
PASSIVE mode, which never waits; the writing connection's
journal_size_limit (64 MiB) shrinks the WAL file when SQLite starts it
again. Checkpoint reads SQLite's busy flag and returns
ErrCheckpointBusy, which the maintainer logs as busy, at warning level,
instead of as a completed checkpoint.

Model: opus-5-5

internal/database/database.go

@@ -81,6 +81,11 @@ const (
 	sqliteHardHeapLimitBytes = 2147483648
 )
 
+// sqliteWALSizeLimitBytes (64 MiB) is the journal_size_limit of the connection
+// kept for writing: when SQLite starts the WAL again from the beginning, it cuts
+// a larger WAL file down to this size.
+const sqliteWALSizeLimitBytes = 67108864
+
 // Settings of the ANALYZE that Analyze runs.
 const (
 	// analysisLimit is about how many rows of each index ANALYZE reads
@@ -112,6 +117,9 @@ var (
 	ErrNoRoute = errors.New("no route found")
 	// ErrNoStaleASN is returned when no ASN needs WHOIS refresh
 	ErrNoStaleASN = errors.New("no stale ASN found")
+	// ErrCheckpointBusy is returned by Checkpoint when SQLite reports the
+	// checkpoint busy: another connection was checkpointing at the same time.
+	ErrCheckpointBusy = errors.New("WAL checkpoint busy")
 
 	errInvalidASN       = errors.New("invalid ASN")
 	errSameASNInPeering = errors.New("cannot create peering with same ASN")
@@ -194,6 +202,13 @@ func New(cfg *config.Config, logger *logger.Logger) (*Database, error) {
 			err)
 	}
 
+	_, err = writer.ExecContext(ctx,
+		fmt.Sprintf("PRAGMA journal_size_limit=%d", sqliteWALSizeLimitBytes))
+	if err != nil {
+		return nil, fmt.Errorf(
+			"failed to set the writing connection's WAL size limit: %w", err)
+	}
+
 	database := &Database{
 		db: db, writer: writer, logger: logger, path: dbPath,
 		counts: &liveCounts{},
@@ -2245,9 +2260,13 @@ func (d *Database) Analyze(ctx context.Context) error {
 	return nil
 }
 
-// Checkpoint runs a WAL checkpoint to transfer data from the WAL to the main database.
+// Checkpoint copies the WAL into the database file. It holds the database lock,
-// Uses TRUNCATE mode which blocks writers briefly but ensures complete checkpoint,
+// so it runs a PASSIVE checkpoint, which never waits: it copies what no reader
-// keeping the WAL small for fast read performance. It holds the database lock.
+// still needs and leaves the rest for the next one. TRUNCATE or RESTART would
+// wait up to the busy timeout for every reader to finish. The WAL file shrinks
+// instead when SQLite starts it again from the beginning
+// (sqliteWALSizeLimitBytes). It returns ErrCheckpointBusy when SQLite reports
+// the checkpoint busy.
 func (d *Database) Checkpoint(ctx context.Context) error {
 	d.lock("Checkpoint")
 	defer d.unlock()
@@ -2257,11 +2276,18 @@ func (d *Database) Checkpoint(ctx context.Context) error {
 		return fmt.Errorf("failed to checkpoint WAL: %w", err)
 	}
 
-	_, err = d.writer.ExecContext(ctx, "PRAGMA wal_checkpoint(TRUNCATE)")
+	var busy, walPages, checkpointedPages int
+
+	err = d.writer.QueryRowContext(ctx, "PRAGMA wal_checkpoint(PASSIVE)").
+		Scan(&busy, &walPages, &checkpointedPages)
 	if err != nil {
 		return fmt.Errorf("failed to checkpoint WAL: %w", err)
 	}
 
+	if busy != 0 {
+		return ErrCheckpointBusy
+	}
+
 	return nil
 }
 

internal/database/database_internal_test.go

@@ -67,6 +67,10 @@ const (
 	lockWaitTimeout = time.Millisecond
 	// lockTestASN is the AS the batch write in the lock tests stores.
 	lockTestASN = 64500
+	// readerCheckpointLimit is how long a checkpoint, and a batch write
+	// waiting for it, may take while a reader is open: well under the busy
+	// timeout, which a checkpoint that waits for the reader takes.
+	readerCheckpointLimit = time.Second
 )
 
 // Parameters for the prefix-length sample tests.
@@ -457,6 +461,33 @@ func TestWritingConnectionCache(t *testing.T) {
 	}
 }
 
+// TestWritingConnectionWALSizeLimit checks that the connection kept for
+// writing, the one that writes the WAL, has the WAL size limit: Checkpoint does
+// not truncate the WAL file, so the limit is what shrinks it.
+func TestWritingConnectionWALSizeLimit(t *testing.T) {
+	t.Parallel()
+
+	db, err := New(&config.Config{StateDir: t.TempDir()}, logger.New())
+	if err != nil {
+		t.Fatalf("failed to create database: %v", err)
+	}
+	defer func() { _ = db.Close() }()
+
+	var limit int64
+
+	err = db.writer.QueryRowContext(t.Context(), "PRAGMA journal_size_limit").
+		Scan(&limit)
+	if err != nil {
+		t.Fatalf("failed to read the writing connection's journal_size_limit: %v",
+			err)
+	}
+
+	if limit != sqliteWALSizeLimitBytes {
+		t.Errorf("writing connection: journal_size_limit = %d, want %d",
+			limit, sqliteWALSizeLimitBytes)
+	}
+}
+
 // readPragma reads the integer value of PRAGMA name on connection i, c.
 func readPragma(
 	ctx context.Context, t *testing.T, c *sql.Conn, i int, name string,
@@ -799,6 +830,91 @@ func TestCheckpointLockWaitEndsWithContext(t *testing.T) {
 	db.mu.Unlock()
 }
 
+// TestCheckpointDoesNotWaitForReader keeps a read transaction on
+// live_routes_v4 open while a route is committed, and checks that Checkpoint,
+// and a batch write started while Checkpoint holds the database lock, both end
+// within readerCheckpointLimit. A TRUNCATE checkpoint waited the busy timeout
+// for the reader while holding the lock
+// (https://git.eeqj.de/sneak/routewatch/issues/102).
+func TestCheckpointDoesNotWaitForReader(t *testing.T) {
+	t.Parallel()
+
+	db, messages := newDatabaseWithLockMessages(t)
+	ctx := context.Background()
+	ts := time.Now().UTC()
+
+	err := db.UpsertLiveRouteBatch(
+		[]*LiveRoute{mkV4Route(t, "198.51.100.0/24", lockTestASN, ts)})
+	if err != nil {
+		t.Fatalf("failed to store the route the reader reads: %v", err)
+	}
+
+	reader, err := db.db.Conn(ctx)
+	if err != nil {
+		t.Fatalf("failed to open the reading connection: %v", err)
+	}
+	defer func() { _ = reader.Close() }()
+
+	// A plain BEGIN starts a transaction that only reads, unlike the BEGIN
+	// IMMEDIATE of _txlock. It keeps the snapshot its first read takes until
+	// it ends.
+	_, err = reader.ExecContext(ctx, "BEGIN")
+	if err != nil {
+		t.Fatalf("failed to begin the read transaction: %v", err)
+	}
+	defer func() { _, _ = reader.ExecContext(ctx, "ROLLBACK") }()
+
+	var routes int
+
+	err = reader.QueryRowContext(ctx,
+		"SELECT count(*) FROM live_routes_v4").Scan(&routes)
+	if err != nil {
+		t.Fatalf("failed to read the live routes: %v", err)
+	}
+
+	err = db.UpsertLiveRouteBatch(
+		[]*LiveRoute{mkV4Route(t, "203.0.113.0/24", lockTestASN, ts)})
+	if err != nil {
+		t.Fatalf("failed to store a route after the read: %v", err)
+	}
+
+	started := time.Now()
+	checkpointed := make(chan error, 1)
+
+	go func() { checkpointed <- db.Checkpoint(ctx) }()
+
+	waitForLockMessage(t, messages, lockHeldMessage, "Checkpoint", checkpointed)
+
+	// A route write, not an AS write: the first new AS loads the embedded AS
+	// names, which takes seconds under -race.
+	route := mkV4Route(t, "192.0.2.0/24", lockTestASN, ts)
+	written := make(chan error, 1)
+
+	go func() { written <- db.UpsertLiveRouteBatch([]*LiveRoute{route}) }()
+
+	err = <-checkpointed
+	if err != nil {
+		t.Errorf("Checkpoint: %v", err)
+	}
+
+	took := time.Since(started)
+	if took >= readerCheckpointLimit {
+		t.Errorf("Checkpoint took %v with a reader open, want under %v",
+			took, readerCheckpointLimit)
+	}
+
+	err = <-written
+	if err != nil {
+		t.Errorf("batch write during Checkpoint: %v", err)
+	}
+
+	took = time.Since(started)
+	if took >= readerCheckpointLimit {
+		t.Errorf("batch write during Checkpoint ended after %v, want under %v",
+			took, readerCheckpointLimit)
+	}
+}
+
 // 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.

internal/routewatch/dbmaintainer.go

@@ -2,6 +2,7 @@ package routewatch
 
 import (
 	"context"
+	"errors"
 	"log/slog"
 	"sync"
 	"time"
@@ -34,6 +35,9 @@ const (
 	analyzeTimeout = 5 * time.Minute
 )
 
+// checkpointBusyMessage is logged when SQLite reports a checkpoint busy.
+const checkpointBusyMessage = "WAL checkpoint busy"
+
 // DBMaintainer handles background database maintenance tasks.
 type DBMaintainer struct {
 	db     database.Store
@@ -123,9 +127,12 @@ func (m *DBMaintainer) runCheckpoint(ctx context.Context) {
 	duration, err := runWithTimeout(ctx, checkpointTimeout, m.db.Checkpoint)
 	m.recordResult(&m.lastCheckpoint, &m.lastCheckpointError, &m.checkpointCount, err)
 
-	if err != nil {
+	switch {
+	case errors.Is(err, database.ErrCheckpointBusy):
+		m.logger.Warn(checkpointBusyMessage, "duration", duration)
+	case err != nil:
 		m.logger.Error("WAL checkpoint failed", "error", err, "duration", duration)
-	} else {
+	default:
 		m.logger.Debug("WAL checkpoint completed", "duration", duration)
 	}
 }

internal/routewatch/dbmaintainer_internal_test.go

@@ -0,0 +1,54 @@
+package routewatch
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"log/slog"
+	"testing"
+
+	"git.eeqj.de/sneak/routewatch/internal/database"
+)
+
+// busyCheckpointStore is a store whose every checkpoint SQLite reports busy.
+type busyCheckpointStore struct{ *mockStore }
+
+// Checkpoint reports the checkpoint busy.
+func (busyCheckpointStore) Checkpoint(context.Context) error {
+	return database.ErrCheckpointBusy
+}
+
+// TestBusyCheckpointLoggedAsBusy runs a checkpoint that SQLite reports busy
+// and checks that the maintainer logs it as busy, at warning level, and does
+// not count it as a checkpoint that completed.
+func TestBusyCheckpointLoggedAsBusy(t *testing.T) {
+	t.Parallel()
+
+	var logs bytes.Buffer
+
+	logger := slog.New(slog.NewJSONHandler(&logs,
+		&slog.HandlerOptions{Level: slog.LevelDebug}))
+	m := NewDBMaintainer(busyCheckpointStore{newMockStore()}, logger)
+
+	m.runCheckpoint(t.Context())
+
+	var logged struct {
+		Level string `json:"level"`
+		Msg   string `json:"msg"`
+	}
+
+	err := json.Unmarshal(logs.Bytes(), &logged)
+	if err != nil {
+		t.Fatalf("want one log message, got %q: %v", logs.String(), err)
+	}
+
+	if logged.Level != slog.LevelWarn.String() || logged.Msg != checkpointBusyMessage {
+		t.Errorf("logged %s %q, want %s %q", logged.Level, logged.Msg,
+			slog.LevelWarn, checkpointBusyMessage)
+	}
+
+	if m.checkpointCount != 0 {
+		t.Errorf("checkpointCount = %d after a busy checkpoint, want 0",
+			m.checkpointCount)
+	}
+}