Checkpoint without waiting on readers and keep the WAL file small (closes #102)

The 5-second checkpoint ran in TRUNCATE mode, which waits up to the
5-second busy timeout for every open read while holding the database
lock every batch write needs. It now runs in PASSIVE mode, which never
waits. Because PASSIVE alone never lets the WAL start again while reads
overlap, a WAL file over 64 MiB gets a TRUNCATE checkpoint whose wait
for reads is capped at 500 ms and logged. A checkpoint that leaves part
of the WAL uncopied returns ErrCheckpointIncomplete, which the
maintainer logs as incomplete instead of completed. The writing
connection's journal_size_limit is 64 MiB.

Model: opus-5-5

internal/database/database.go

@@ -81,6 +81,21 @@ const (
 	sqliteHardHeapLimitBytes = 2147483648
 )
 
+// Settings that keep the WAL file small (see Checkpoint).
+const (
+	// sqliteWALSizeLimitBytes (64 MiB) is the WAL file size past which
+	// Checkpoint truncates the file. It is also 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.
+	sqliteWALSizeLimitBytes = 67108864
+	// truncateBusyTimeoutMs is the busy timeout of the TRUNCATE checkpoint:
+	// the longest it waits for open reads, and so the longest it keeps batch
+	// writes waiting beyond the copy itself.
+	truncateBusyTimeoutMs = 500
+	// walFileSuffix ends the name of the WAL file, next to the database file.
+	walFileSuffix = "-wal"
+)
+
 // Settings of the ANALYZE that Analyze runs.
 const (
 	// analysisLimit is about how many rows of each index ANALYZE reads
@@ -112,6 +127,11 @@ var (
 	ErrNoRoute = errors.New("no route found")
 	// ErrNoStaleASN is returned when no ASN needs WHOIS refresh
 	ErrNoStaleASN = errors.New("no stale ASN found")
+	// ErrCheckpointIncomplete is returned by Checkpoint when it left part of
+	// the WAL uncopied, or left a WAL file over sqliteWALSizeLimitBytes
+	// untruncated, because reads still needed the WAL or SQLite reported the
+	// checkpoint busy. A later checkpoint finishes the work.
+	ErrCheckpointIncomplete = errors.New("WAL checkpoint incomplete")
 
 	errInvalidASN       = errors.New("invalid ASN")
 	errSameASNInPeering = errors.New("cannot create peering with same ASN")
@@ -194,6 +214,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{},
@@ -2255,9 +2282,25 @@ 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 and keeps the WAL file
-// Uses TRUNCATE mode which blocks writers briefly but ensures complete checkpoint,
+// small. It holds the database lock, so batch writes wait while it runs.
-// keeping the WAL small for fast read performance. It holds the database lock.
+//
+// It runs a PASSIVE checkpoint, which never waits: it copies what no open read
+// still needs. That alone does not keep the file small: SQLite starts the WAL
+// again from the beginning only once a checkpoint has copied all of it and no
+// read that began before is still open, which never happens while reads keep
+// overlapping. So when the file is over sqliteWALSizeLimitBytes, Checkpoint
+// runs a TRUNCATE checkpoint instead, which waits for the reads that still need
+// the WAL, copies all of it and empties the file. That checkpoint waits at most
+// truncateBusyTimeoutMs, and it is logged with how long it took.
+//
+// So while reads overlap, the WAL file stays under sqliteWALSizeLimitBytes
+// plus what batch writes add between two checkpoints, as long as no read stays
+// open longer than that wait. A longer read makes the TRUNCATE checkpoint give
+// up, and the file grows until a later one empties it.
+//
+// It returns ErrCheckpointIncomplete when it leaves part of the WAL uncopied,
+// or a file over the limit untruncated.
 func (d *Database) Checkpoint(ctx context.Context) error {
 	d.lock("Checkpoint")
 	defer d.unlock()
@@ -2267,11 +2310,32 @@ 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)")
+	wal, err := os.Stat(d.path + walFileSuffix)
+	if err != nil {
+		return fmt.Errorf("failed to read the WAL file's size: %w", err)
+	}
+
+	if wal.Size() > sqliteWALSizeLimitBytes {
+		return d.truncateWAL(ctx, wal.Size())
+	}
+
+	var busy, walPages, copiedPages int
+
+	err = d.writer.QueryRowContext(ctx, "PRAGMA wal_checkpoint(PASSIVE)").
+		Scan(&busy, &walPages, &copiedPages)
 	if err != nil {
 		return fmt.Errorf("failed to checkpoint WAL: %w", err)
 	}
 
+	if busy != 0 {
+		return fmt.Errorf("%w: SQLite reported it busy", ErrCheckpointIncomplete)
+	}
+
+	if copiedPages < walPages {
+		return fmt.Errorf("%w: copied %d of %d WAL pages",
+			ErrCheckpointIncomplete, copiedPages, walPages)
+	}
+
 	return nil
 }
 
@@ -2287,6 +2351,50 @@ func (d *Database) Ping(ctx context.Context) error {
 	return nil
 }
 
+// truncateWAL runs the TRUNCATE checkpoint of Checkpoint on a WAL file of
+// walBytes, with the writing connection's busy timeout lowered to
+// truncateBusyTimeoutMs for it, and logs how long it took. The caller holds the
+// database lock.
+func (d *Database) truncateWAL(ctx context.Context, walBytes int64) error {
+	_, err := d.writer.ExecContext(ctx,
+		fmt.Sprintf("PRAGMA busy_timeout=%d", truncateBusyTimeoutMs))
+	if err != nil {
+		return fmt.Errorf("failed to lower the busy timeout to truncate WAL: %w",
+			err)
+	}
+
+	started := time.Now()
+
+	var busy, walPages, copiedPages int
+
+	checkpointErr := d.writer.QueryRowContext(ctx,
+		"PRAGMA wal_checkpoint(TRUNCATE)").Scan(&busy, &walPages, &copiedPages)
+	took := time.Since(started)
+
+	// Restored even when ctx has ended, so that no batch write runs with the
+	// short timeout.
+	_, restoreErr := d.writer.ExecContext(context.WithoutCancel(ctx),
+		fmt.Sprintf("PRAGMA busy_timeout=%d", sqliteBusyTimeoutMs))
+	if checkpointErr != nil || restoreErr != nil {
+		return fmt.Errorf("failed to truncate WAL: %w",
+			errors.Join(checkpointErr, restoreErr))
+	}
+
+	if busy != 0 {
+		d.logger.Warn("WAL file over its size limit not truncated",
+			"wal_bytes", walBytes, "wal_pages", walPages,
+			"copied_pages", copiedPages, "duration", took)
+
+		return fmt.Errorf("%w: WAL file of %d bytes not truncated",
+			ErrCheckpointIncomplete, walBytes)
+	}
+
+	d.logger.Info("WAL file over its size limit truncated",
+		"wal_bytes", walBytes, "duration", took)
+
+	return nil
+}
+
 // The debug messages lock logs, with the operation's name, when it starts to
 // wait for the database mutex and when it holds it.
 const (

internal/database/database_internal_test.go

@@ -8,6 +8,7 @@ import (
 	"log/slog"
 	"math/rand/v2"
 	"net/netip"
+	"os"
 	"slices"
 	"strconv"
 	"sync"
@@ -67,6 +68,28 @@ 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 of the test of the WAL file's size while reads overlap.
+const (
+	// walFillStep is how much the test writes between two checkpoints.
+	walFillStep = sqliteWALSizeLimitBytes / 16
+	// walFillTotal is how much it writes in all: past sqliteWALSizeLimitBytes,
+	// and on until a WAL file that no checkpoint empties ends over walFillBound.
+	walFillTotal = sqliteWALSizeLimitBytes + 4*walFillStep
+	// walFillBound is the size the WAL file must stay under: the limit, plus
+	// the writes since the last checkpoint, plus one more step for SQLite's
+	// own pages besides the written data.
+	walFillBound = sqliteWALSizeLimitBytes + 2*walFillStep
+	// overlappingReads is how many reads the test keeps open at once.
+	overlappingReads = 2
+	// overlappingReadDuration is how long each read stays open: well under
+	// truncateBusyTimeoutMs, so the TRUNCATE checkpoint can wait for it.
+	overlappingReadDuration = 50 * time.Millisecond
 )
 
 // Parameters for the prefix-length sample tests.
@@ -457,6 +480,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, which cuts the
+// WAL file back when SQLite starts the WAL again from the beginning.
+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 +849,229 @@ func TestCheckpointLockWaitEndsWithContext(t *testing.T) {
 	db.mu.Unlock()
 }
 
+// TestCheckpointDoesNotWaitForReader keeps a read transaction on
+// live_routes_v4 open while a route is committed, then starts Checkpoint and a
+// batch write together. It checks that both end within readerCheckpointLimit,
+// whichever of them takes the database lock first, and that Checkpoint reports
+// the part of the WAL the reader still needs as left uncopied. 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, err := New(&config.Config{StateDir: t.TempDir()}, logger.New())
+	if err != nil {
+		t.Fatalf("failed to create database: %v", err)
+	}
+	defer func() { _ = db.Close() }()
+
+	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) }()
+
+	// 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 !errors.Is(err, ErrCheckpointIncomplete) {
+		t.Errorf("Checkpoint = %v with a reader open, want %v",
+			err, ErrCheckpointIncomplete)
+	}
+
+	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)
+	}
+}
+
+// TestCheckpointKeepsWALSmallWhileReadsOverlap writes walFillTotal while
+// overlappingReads reads keep overlapping, so that one is open at every write,
+// and runs Checkpoint after every walFillStep, as the maintainer does between
+// batch writes. It checks that the WAL file stays under walFillBound and that
+// the writing connection has its busy timeout back afterwards. With only a
+// PASSIVE checkpoint, SQLite never started the WAL again from the beginning
+// while reads overlapped, and the file grew for as long as they did
+// (https://git.eeqj.de/sneak/routewatch/issues/102).
+func TestCheckpointKeepsWALSmallWhileReadsOverlap(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() }()
+
+	ctx := t.Context()
+
+	// The test writes to a table of its own: each write replaces the one
+	// row's data with walFillStep of new random bytes, so the WAL grows by that
+	// much while the database file stays small. SQLite writes no page whose
+	// content stays the same, so the bytes must change. Like batch writes,
+	// the writes use the writing connection between the checkpoints.
+	_, err = db.writer.ExecContext(ctx,
+		"CREATE TABLE wal_fill (data BLOB); INSERT INTO wal_fill VALUES (NULL)")
+	if err != nil {
+		t.Fatalf("failed to create the table the test writes to: %v", err)
+	}
+
+	stop := make(chan struct{})
+
+	var readers sync.WaitGroup
+
+	for i := range overlappingReads {
+		readers.Go(func() {
+			// Reader i starts i/overlappingReads of a read after the first,
+			// so that the reads overlap.
+			time.Sleep(time.Duration(i) * overlappingReadDuration /
+				overlappingReads)
+			keepReading(ctx, t, db, stop)
+		})
+	}
+
+	defer func() {
+		close(stop)
+		readers.Wait()
+	}()
+
+	for written := 0; written < walFillTotal; written += walFillStep {
+		_, err = db.writer.ExecContext(ctx,
+			"UPDATE wal_fill SET data = randomblob(?)", walFillStep)
+		if err != nil {
+			t.Fatalf("failed to write: %v", err)
+		}
+
+		wal, statErr := os.Stat(db.path + walFileSuffix)
+		if statErr != nil {
+			t.Fatalf("failed to read the WAL file's size: %v", statErr)
+		}
+
+		if wal.Size() > walFillBound {
+			t.Fatalf("WAL file of %d bytes after writing %d with reads "+
+				"overlapping, want at most %d", wal.Size(), written+walFillStep,
+				walFillBound)
+		}
+
+		err = db.Checkpoint(ctx)
+		if err != nil && !errors.Is(err, ErrCheckpointIncomplete) {
+			t.Fatalf("Checkpoint: %v", err)
+		}
+	}
+
+	busyTimeout := readPragma(ctx, t, db.writer, 0, "busy_timeout")
+	if busyTimeout != sqliteBusyTimeoutMs {
+		t.Errorf("writing connection: busy_timeout = %d after the checkpoints, "+
+			"want %d", busyTimeout, sqliteBusyTimeoutMs)
+	}
+}
+
+// keepReading runs one read transaction after another on a connection of its
+// own, each open for overlappingReadDuration, until stop is closed.
+func keepReading(
+	ctx context.Context, t *testing.T, db *Database, stop <-chan struct{},
+) {
+	t.Helper()
+
+	conn, err := db.db.Conn(ctx)
+	if err != nil {
+		t.Errorf("failed to open a reading connection: %v", err)
+
+		return
+	}
+	defer func() { _ = conn.Close() }()
+
+	for {
+		select {
+		case <-stop:
+			return
+		default:
+		}
+
+		// 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 = conn.ExecContext(ctx, "BEGIN")
+		if err != nil {
+			t.Errorf("failed to begin a read transaction: %v", err)
+
+			return
+		}
+
+		var rows int
+
+		err = conn.QueryRowContext(ctx, "SELECT count(*) FROM wal_fill").
+			Scan(&rows)
+		if err != nil {
+			t.Errorf("failed to read: %v", err)
+
+			return
+		}
+
+		time.Sleep(overlappingReadDuration)
+
+		_, err = conn.ExecContext(ctx, "ROLLBACK")
+		if err != nil {
+			t.Errorf("failed to end a read transaction: %v", err)
+
+			return
+		}
+	}
+}
+
 // 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,12 @@ const (
 	analyzeTimeout = 5 * time.Minute
 )
 
+// checkpointIncompleteMessage is logged, at debug level like a completed
+// checkpoint, when a checkpoint leaves part of the WAL for a later one: while
+// reads overlap, most do. The database warns when the WAL file stays over its
+// size limit.
+const checkpointIncompleteMessage = "WAL checkpoint incomplete"
+
 // DBMaintainer handles background database maintenance tasks.
 type DBMaintainer struct {
 	db     database.Store
@@ -123,9 +130,13 @@ 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.ErrCheckpointIncomplete):
+		m.logger.Debug(checkpointIncompleteMessage,
+			"error", err, "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"
+)
+
+// incompleteCheckpointStore is a store whose every checkpoint leaves part of
+// the WAL uncopied.
+type incompleteCheckpointStore struct{ *mockStore }
+
+// Checkpoint reports the checkpoint incomplete.
+func (incompleteCheckpointStore) Checkpoint(context.Context) error {
+	return database.ErrCheckpointIncomplete
+}
+
+// TestIncompleteCheckpointNotLoggedAsCompleted runs a checkpoint that leaves
+// part of the WAL uncopied and checks that the maintainer logs it as
+// incomplete, not as completed, and does not count it as a checkpoint that
+// completed.
+func TestIncompleteCheckpointNotLoggedAsCompleted(t *testing.T) {
+	t.Parallel()
+
+	var logs bytes.Buffer
+
+	logger := slog.New(slog.NewJSONHandler(&logs,
+		&slog.HandlerOptions{Level: slog.LevelDebug}))
+	m := NewDBMaintainer(incompleteCheckpointStore{newMockStore()}, logger)
+
+	m.runCheckpoint(t.Context())
+
+	var logged struct {
+		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.Msg != checkpointIncompleteMessage {
+		t.Errorf("logged %q, want %q", logged.Msg, checkpointIncompleteMessage)
+	}
+
+	if m.checkpointCount != 0 {
+		t.Errorf("checkpointCount = %d after an incomplete checkpoint, want 0",
+			m.checkpointCount)
+	}
+}