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
2026-10-19 22:17:13 +00:00
parent 69d477159c
commit 4b2d3a9ff2
4 changed files with 452 additions and 6 deletions
@@ -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.