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 while holding the database
lock every batch write needs. It now runs in PASSIVE mode, which never
waits. A WAL file over 64 MiB instead gets one TRUNCATE checkpoint whose
wait for reads is capped at 500 ms and logged; it empties the file once
no read is open, and while reads follow one another with no pause it
can give up every time. 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,23 @@ const (
 	sqliteHardHeapLimitBytes = 2147483648
 )
 
+// Settings that bound the WAL file's size once no read is open (see
+// Checkpoint).
+const (
+	// sqliteWALSizeLimitBytes (64 MiB) is the WAL file size past which
+	// Checkpoint runs a TRUNCATE checkpoint. 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 +129,12 @@ 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 that runs when no read is open
+	// 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 +217,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 +2285,24 @@ 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.
-// Uses TRUNCATE mode which blocks writers briefly but ensures complete checkpoint,
-// keeping the WAL small for fast read performance. It holds the database lock.
+// Checkpoint copies the WAL into the database file. It holds the database
+// lock, so batch writes wait while it runs.
+//
+// It runs a PASSIVE checkpoint, which never waits: it copies what no open read
+// still needs, and leaves the file as large as it is. When the file is over
+// sqliteWALSizeLimitBytes, Checkpoint runs one TRUNCATE checkpoint instead,
+// which waits at most truncateBusyTimeoutMs for the open reads to end, then
+// copies all of the WAL and empties the file, or gives up. It is logged with
+// how long it took and whether it emptied the file.
+//
+// So a file over the limit is emptied by the first Checkpoint that finds no
+// read open, at its start or at some moment of its wait. While reads follow
+// one another with no pause between them, every TRUNCATE checkpoint can give
+// up after its full wait, and the file can keep growing for as long as they
+// do.
+//
+// It returns ErrCheckpointIncomplete when it leaves part of the WAL uncopied,
+// or a file over the limit not emptied.
 func (d *Database) Checkpoint(ctx context.Context) error {
 	d.lock("Checkpoint")
 	defer d.unlock()
@@ -2267,11 +2312,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 +2353,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,15 @@ 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
+	// truncateCheckpointLimit is how long a checkpoint of a WAL file over
+	// sqliteWALSizeLimitBytes may take to give up while a reader is open: its
+	// wait for the reader plus a second, well under the busy timeout.
+	truncateCheckpointLimit = truncateBusyTimeoutMs*time.Millisecond +
+		time.Second
 )
 
 // Parameters for the prefix-length sample tests.
@@ -457,6 +467,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 +836,218 @@ 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)
+	}
+}
+
+// TestCheckpointEmptiesWALOnceNoReadIsOpen fills the WAL file past
+// sqliteWALSizeLimitBytes while the test holds a read open. With the read
+// open, it checks that Checkpoint gives up within truncateCheckpointLimit and
+// reports the checkpoint incomplete, and that a batch write still succeeds.
+// After the test closes the read, it checks that the next Checkpoint brings
+// the WAL file under the limit and gives the writing connection its busy
+// timeout back (https://git.eeqj.de/sneak/routewatch/issues/102).
+func TestCheckpointEmptiesWALOnceNoReadIsOpen(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()
+
+	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, so no checkpoint can copy what is written after that read.
+	_, err = reader.ExecContext(ctx, "BEGIN")
+	if err != nil {
+		t.Fatalf("failed to begin the read transaction: %v", err)
+	}
+
+	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)
+	}
+
+	fillWALPastLimit(ctx, t, db)
+
+	started := time.Now()
+
+	err = db.Checkpoint(ctx)
+	if !errors.Is(err, ErrCheckpointIncomplete) {
+		t.Errorf("Checkpoint = %v with a read open, want %v",
+			err, ErrCheckpointIncomplete)
+	}
+
+	took := time.Since(started)
+	if took >= truncateCheckpointLimit {
+		t.Errorf("Checkpoint took %v with a read open, want under %v",
+			took, truncateCheckpointLimit)
+	}
+
+	err = db.UpsertLiveRouteBatch([]*LiveRoute{
+		mkV4Route(t, "192.0.2.0/24", lockTestASN, time.Now().UTC()),
+	})
+	if err != nil {
+		t.Errorf("batch write after Checkpoint gave up: %v", err)
+	}
+
+	_, err = reader.ExecContext(ctx, "ROLLBACK")
+	if err != nil {
+		t.Fatalf("failed to end the read transaction: %v", err)
+	}
+
+	err = db.Checkpoint(ctx)
+	if err != nil {
+		t.Errorf("Checkpoint with no read open: %v", err)
+	}
+
+	size := walFileSize(t, db)
+	if size >= sqliteWALSizeLimitBytes {
+		t.Errorf("WAL file of %d bytes after Checkpoint with no read open, "+
+			"want under %d", size, sqliteWALSizeLimitBytes)
+	}
+
+	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)
+	}
+}
+
+// fillWALPastLimit writes sqliteWALSizeLimitBytes of random bytes to a table
+// of the test's own, which with SQLite's page and frame headers leaves the WAL
+// file of db over the limit, and checks that it does.
+func fillWALPastLimit(ctx context.Context, t *testing.T, db *Database) {
+	t.Helper()
+
+	_, err := db.writer.ExecContext(ctx, "CREATE TABLE wal_fill (data BLOB)")
+	if err != nil {
+		t.Fatalf("failed to create the table the test writes to: %v", err)
+	}
+
+	_, err = db.writer.ExecContext(ctx,
+		"INSERT INTO wal_fill VALUES (randomblob(?))", sqliteWALSizeLimitBytes)
+	if err != nil {
+		t.Fatalf("failed to fill the WAL: %v", err)
+	}
+
+	size := walFileSize(t, db)
+	if size <= sqliteWALSizeLimitBytes {
+		t.Fatalf("WAL file of %d bytes after the test's write, want over %d",
+			size, sqliteWALSizeLimitBytes)
+	}
+}
+
+// walFileSize returns the size of the WAL file of db.
+func walFileSize(t *testing.T, db *Database) int64 {
+	t.Helper()
+
+	wal, err := os.Stat(db.path + walFileSuffix)
+	if err != nil {
+		t.Fatalf("failed to read the WAL file's size: %v", err)
+	}
+
+	return wal.Size()
+}
+
 // 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)
+	}
+}