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_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.