Log the start-up checkpoint's figures under their own names (closes #117)

Initialize scanned the row of PRAGMA wal_checkpoint(TRUNCATE), which is
the busy flag, the frames in the WAL and the frames copied, into
wal_pages, checkpointed and moved; and a completed TRUNCATE reads 0 for
both counts. The line now logs busy, wal_frames and checkpointed_frames,
plus wal_bytes, the WAL file's size, which New reads before it opens the
database because the open writes a frame to the WAL. The line is a
warning when busy is 1. Tests open a database a second time while the
first stays open, with and without a read open on the first, and check
the line's level and figures.

Model: opus-5-5

README.md

@@ -5,6 +5,22 @@ table in real time: it streams BGP UPDATE messages from the RIPE RIS Live
 service, maintains a live routing table in SQLite, and provides HTTP APIs for
 querying routing information.
 
+## Rationale
+
+RIPE RIS Live streams the BGP updates that RIPE's route collectors receive, as
+they happen, from the moment a client connects. It reports changes, not the
+routing table they add up to, so it cannot say which AS an IP address is routed
+to now or which routes a prefix has. RouteWatch applies the feed's updates to
+live routing tables in its own SQLite database and answers IP-to-AS lookups and
+prefix pages from them.
+
+The feed sends no table when a client connects, so these tables hold only the
+routes the feed has announced since RouteWatch first started: an IP-to-AS lookup
+finds nothing for an address whose routes have not been announced since then.
+The tables also miss the updates RouteWatch drops when it falls behind under
+load and those sent while it is stopped, so a lookup or a prefix page can show a
+route the feed has withdrawn.
+
 ## Features
 
 - Real-time streaming of BGP updates from RIPE RIS Live

internal/database/database.go

@@ -169,6 +169,20 @@ func New(cfg *config.Config, logger *logger.Logger) (*Database, error) {
 		return nil, fmt.Errorf("failed to create database directory: %w", err)
 	}
 
+	// The size of the WAL file the previous run left, which Initialize logs
+	// with the start-up checkpoint. It is read before the database is opened,
+	// because opening it writes a frame to the WAL. There is no WAL file when
+	// the database is new or the last run closed the database, which deletes
+	// it.
+	var walBytes int64
+
+	wal, err := os.Stat(dbPath + walFileSuffix)
+	if err == nil {
+		walBytes = wal.Size()
+	} else if !errors.Is(err, os.ErrNotExist) {
+		logger.Warn("Failed to read the WAL file's size on startup", "error", err)
+	}
+
 	// Per-connection SQLite settings go in the DSN so every pooled connection
 	// gets them, not just the one that runs the Initialize pragmas. _txlock=
 	// immediate makes every transaction take the write lock at BEGIN. Without it
@@ -229,7 +243,7 @@ func New(cfg *config.Config, logger *logger.Logger) (*Database, error) {
 		counts: &liveCounts{},
 	}
 
-	err = database.Initialize()
+	err = database.Initialize(walBytes)
 	if err != nil {
 		return nil, fmt.Errorf("failed to initialize database: %w", err)
 	}
@@ -245,8 +259,16 @@ func New(cfg *config.Config, logger *logger.Logger) (*Database, error) {
 	return database, nil
 }
 
-// Initialize creates the database schema if it doesn't exist.
-func (d *Database) Initialize() error {
+// The messages Initialize logs the start-up WAL checkpoint with.
+const (
+	startupCheckpointMessage           = "WAL checkpoint on startup"
+	startupCheckpointIncompleteMessage = "WAL checkpoint on startup incomplete"
+)
+
+// Initialize creates the database schema if it doesn't exist. walBytes is the
+// size of the WAL file the previous run left, which New reads before it opens
+// the database.
+func (d *Database) Initialize(walBytes int64) error {
 	// Set SQLite pragmas for performance. Per-connection settings (cache_size,
 	// synchronous, busy_timeout, journal_mode) live in the DSN, and New sets the
 	// writing connection's cache_size; temp_store is left at its default so
@@ -266,19 +288,33 @@ func (d *Database) Initialize() error {
 		}
 	}
 
-	// Run WAL checkpoint on startup to consolidate any existing WAL
-	var walPages, checkpointedPages, movedPages int
+	// The start-up checkpoint copies the WAL a previous run left into the
+	// database file and empties the WAL file; its frame counts then read 0,
+	// so walBytes is logged with them.
+	var busy, walFrames, checkpointedFrames int
 
 	err := d.writer.QueryRowContext(context.Background(),
 		"PRAGMA wal_checkpoint(TRUNCATE)",
-	).Scan(&walPages, &checkpointedPages, &movedPages)
-	if err != nil {
+	).Scan(&busy, &walFrames, &checkpointedFrames)
+
+	switch {
+	case err != nil:
 		d.logger.Warn("Failed to checkpoint WAL on startup", "error", err)
-	} else {
-		d.logger.Info("WAL checkpoint on startup",
-			"wal_pages", walPages,
-			"checkpointed", checkpointedPages,
-			"moved", movedPages,
+	case busy != 0:
+		// Another process using the database kept the checkpoint from
+		// copying or emptying all of the WAL.
+		d.logger.Warn(startupCheckpointIncompleteMessage,
+			"wal_bytes", walBytes,
+			"busy", busy,
+			"wal_frames", walFrames,
+			"checkpointed_frames", checkpointedFrames,
+		)
+	default:
+		d.logger.Info(startupCheckpointMessage,
+			"wal_bytes", walBytes,
+			"busy", busy,
+			"wal_frames", walFrames,
+			"checkpointed_frames", checkpointedFrames,
 		)
 	}
 

internal/database/database_internal_test.go

@@ -1,8 +1,10 @@
 package database
 
 import (
+	"bytes"
 	"context"
 	"database/sql"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"log/slog"
@@ -1052,6 +1054,176 @@ func walFileSize(t *testing.T, db *Database) int64 {
 	return wal.Size()
 }
 
+// TestStartupCheckpointLogsItsFigures opens a new database, writes a route,
+// and opens the same file again while the first stays open, so that its WAL
+// file stays. The first open finds no WAL file: it logs a size of 0 and no
+// warning. The second logs the size of the WAL file the first left, and the
+// checkpoint's busy flag and frame counts, each under its own name; the
+// counts read 0 once the checkpoint has emptied the file.
+// Initialize logged the busy flag as wal_pages and the counts under each
+// other's names (https://git.eeqj.de/sneak/routewatch/issues/117).
+func TestStartupCheckpointLogsItsFigures(t *testing.T) {
+	t.Parallel()
+
+	cfg := &config.Config{StateDir: t.TempDir()}
+
+	db, lines := openLogged(t, cfg)
+
+	for _, line := range lines {
+		if line[slog.LevelKey] == slog.LevelWarn.String() {
+			t.Errorf("opening a new database logged a warning: %v", line)
+		}
+	}
+
+	line := findLine(t, lines, startupCheckpointMessage)
+	if line["wal_bytes"] != 0.0 {
+		t.Errorf("new database: start-up checkpoint logged wal_bytes = %v, "+
+			"want 0", line["wal_bytes"])
+	}
+
+	err := db.UpsertLiveRouteBatch([]*LiveRoute{
+		mkV4Route(t, "192.0.2.0/24", lockTestASN, time.Now().UTC()),
+	})
+	if err != nil {
+		t.Fatalf("failed to store a route: %v", err)
+	}
+
+	walBytes := walFileSize(t, db)
+	if walBytes == 0 {
+		t.Fatal("WAL file empty after the route write, want frames in it")
+	}
+
+	_, lines = openLogged(t, cfg)
+	line = findLine(t, lines, startupCheckpointMessage)
+
+	for _, key := range []string{"busy", "wal_frames", "checkpointed_frames"} {
+		if line[key] != 0.0 {
+			t.Errorf("start-up checkpoint logged %s = %v, want 0", key, line[key])
+		}
+	}
+
+	if line["wal_bytes"] != float64(walBytes) {
+		t.Errorf("start-up checkpoint logged wal_bytes = %v, want the %d "+
+			"bytes the first database left", line["wal_bytes"], walBytes)
+	}
+}
+
+// TestStartupCheckpointWarnsWhenIncomplete keeps a read open on a database
+// while a route is written after the read began, then opens the same file
+// again. The second open's start-up checkpoint cannot copy the frames written
+// after the read began, and gives up once the busy timeout has passed. It
+// checks that the checkpoint is logged as a warning with busy 1 and fewer
+// frames copied than the WAL holds.
+func TestStartupCheckpointWarnsWhenIncomplete(t *testing.T) {
+	t.Parallel()
+
+	cfg := &config.Config{StateDir: t.TempDir()}
+	db, _ := openLogged(t, cfg)
+	ctx := context.Background()
+
+	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, "192.0.2.0/24", lockTestASN, time.Now().UTC()),
+	})
+	if err != nil {
+		t.Fatalf("failed to store a route after the read: %v", err)
+	}
+
+	_, lines := openLogged(t, cfg)
+	line := findLine(t, lines, startupCheckpointIncompleteMessage)
+
+	if line[slog.LevelKey] != slog.LevelWarn.String() {
+		t.Errorf("start-up checkpoint logged at level %v, want %s",
+			line[slog.LevelKey], slog.LevelWarn)
+	}
+
+	if line["busy"] != 1.0 {
+		t.Errorf("start-up checkpoint logged busy = %v, want 1", line["busy"])
+	}
+
+	walFrames, _ := line["wal_frames"].(float64)
+	checkpointedFrames, _ := line["checkpointed_frames"].(float64)
+
+	if checkpointedFrames >= walFrames {
+		t.Errorf("start-up checkpoint logged checkpointed_frames = %v and "+
+			"wal_frames = %v with a read open, want fewer copied than the WAL "+
+			"holds", line["checkpointed_frames"], line["wal_frames"])
+	}
+}
+
+// openLogged opens the database of cfg with a logger that writes JSON, and
+// returns it with the lines it logged while it opened, decoded.
+func openLogged(t *testing.T, cfg *config.Config) (*Database, []map[string]any) {
+	t.Helper()
+
+	var logged bytes.Buffer
+
+	db, err := New(cfg,
+		&logger.Logger{Logger: slog.New(slog.NewJSONHandler(&logged, nil))})
+	if err != nil {
+		t.Fatalf("failed to open the database: %v", err)
+	}
+
+	t.Cleanup(func() { _ = db.Close() })
+
+	var lines []map[string]any
+
+	decoder := json.NewDecoder(&logged)
+	for decoder.More() {
+		var line map[string]any
+
+		err := decoder.Decode(&line)
+		if err != nil {
+			t.Fatalf("failed to decode the log: %v", err)
+		}
+
+		lines = append(lines, line)
+	}
+
+	return db, lines
+}
+
+// findLine returns the first of lines logged with message.
+func findLine(
+	t *testing.T,
+	lines []map[string]any,
+	message string,
+) map[string]any {
+	t.Helper()
+
+	for _, line := range lines {
+		if line[slog.MessageKey] == message {
+			return line
+		}
+	}
+
+	t.Fatalf("no line %q in the log", message)
+
+	return nil
+}
+
 // 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.