From 3b3fa8cee3ed7ad6668c71a76b9986ce12e90cf5 Mon Sep 17 00:00:00 2001 From: Noah Pederson Date: Wed, 14 Oct 2026 15:44:24 -0500 Subject: [PATCH] Increase archive WAL checkpoint intervals Use a 65536-page checkpoint threshold during incomplete full all-author replay. Reapply the connection-local policy from durable index state after preparation and process restarts. Restore the normal 1000-page threshold before live readiness. Keep normal startup, subsets, participant replay, and completed sources on the normal threshold. Preserve schema, cache configuration, atomic checkpoints, and synchronous=NORMAL. Larger intervals reduce repeated database page writes. Local benchmarks improve new-record throughput by 13% and update throughput by 25%, with the final checkpoint cost included. Document WAL growth, longer stalls, and the larger power-loss replay window. Race tests, vet, SQL checks, both Nix builds, and module checks pass. tsk-47 --- docs/jetstream-replay.md | 11 ++++++++++ internal/storage/replay_indexes.go | 33 +++++++++++++++++++++++++++++- internal/storage/sqlite.go | 3 +++ 3 files changed, 46 insertions(+), 1 deletion(-) diff --git a/docs/jetstream-replay.md b/docs/jetstream-replay.md index bdaf046..94dda66 100644 --- a/docs/jetstream-replay.md +++ b/docs/jetstream-replay.md @@ -6,6 +6,17 @@ The native SQLite driver uses the bundled SQLite source and requires CGO. Existing SQLite databases need no format conversion or schema change for the driver switch. The driver switch preserves the cache configuration, WAL/NORMAL durability, index deferral, and atomic replay checkpoints. +Incomplete full all-author replay uses a WAL checkpoint threshold of 65,536 pages, approximately 256 MiB with 4 KiB pages. +Larger thresholds combine repeated page writes and reduce checkpoint frequency. +The WAL can exceed the threshold by one transaction and retain its allocated size after a checkpoint. +Checkpoints still run on the writer, so individual checkpoint stalls can grow. + +Replay restores the normal 1,000-page threshold before live readiness. +Retries and restarts retain the archive threshold while the source has deferred indexes. +Completed sources, disabled replay, subsets, and participant replay use the normal threshold. +Schema, atomic checkpoints, and `synchronous=NORMAL` remain unchanged. +The larger interval can require more replay after a power loss because WAL synchronization occurs less often. + Historical replay is optional. The default appview still uses the legacy Microcosm stream. The v2 adapter reconstructs current records, not a complete mutation audit. The provider can compact deleted records from its archive. diff --git a/internal/storage/replay_indexes.go b/internal/storage/replay_indexes.go index f1b5058..f838ff4 100644 --- a/internal/storage/replay_indexes.go +++ b/internal/storage/replay_indexes.go @@ -4,6 +4,7 @@ import ( "context" "database/sql" "errors" + "fmt" "log/slog" "strings" "time" @@ -26,7 +27,23 @@ func (s *SQLite) PrepareReplay(ctx context.Context, source string) (mode ReplayI if source == "" { return ReplayIndexesReady, errors.New("replay source is required") } - tx, err := s.conn.BeginTx(ctx, nil) + conn, err := s.conn.Conn(ctx) + if err != nil { + return ReplayIndexesReady, err + } + defer conn.Close() + // Reapply connection-local policy after every preparation, including archive restarts. + defer func() { + if err != nil { + return + } + pages := sqliteWALPages + if mode == ReplayIndexesDeferred { + pages = sqliteReplayWALPages + } + err = setCheckpointLimit(ctx, conn, pages) + }() + tx, err := conn.BeginTx(ctx, nil) if err != nil { return ReplayIndexesReady, err } @@ -61,6 +78,12 @@ func (s *SQLite) FinishReplay(ctx context.Context) (err error) { return err } defer conn.Close() + // Restore the live threshold only after a successful rebuild, before readiness publication. + defer func() { + if err == nil { + err = setCheckpointLimit(ctx, conn, sqliteWALPages) + } + }() state, err := db.New(conn).ReplayIndexState(ctx) if err != nil { return err @@ -79,6 +102,9 @@ func (s *SQLite) FinishReplay(ctx context.Context) (err error) { err = errors.Join(err, restoreError) }() tx, err := conn.BeginTx(ctx, nil) + if err != nil { + return err + } defer tx.Rollback() if err := replayIndexBuild(ctx, tx); err != nil { return err @@ -89,6 +115,11 @@ func (s *SQLite) FinishReplay(ctx context.Context) (err error) { return replayCommit(ctx, tx) } +func setCheckpointLimit(ctx context.Context, conn *sql.Conn, pages int) error { + _, err := conn.ExecContext(ctx, fmt.Sprintf("PRAGMA wal_autocheckpoint=%d", pages)) + return err +} + func replayIndexDrop(ctx context.Context, tx *sql.Tx) error { for _, index := range []string{"soft_vouches_pair", "soft_vouches_subject"} { if err := tracedDDL(ctx, tx, "replay.index_drop."+index, "DROP INDEX IF EXISTS "+index); err != nil { diff --git a/internal/storage/sqlite.go b/internal/storage/sqlite.go index c42919d..55724df 100644 --- a/internal/storage/sqlite.go +++ b/internal/storage/sqlite.go @@ -68,6 +68,8 @@ const SchemaVersion = 12 const sqliteDriver = "sqlite3" const sqliteCacheKiB = 16 * 1024 * 1024 const sqliteBusyTimeoutMS = 5000 +const sqliteWALPages = 1000 +const sqliteReplayWALPages = 64 * 1024 type SQLite struct { path string @@ -119,6 +121,7 @@ func openSQLite(path string, mode domain.SoftMode, indexes ReplayIndexMode) (sto "PRAGMA journal_mode=WAL", fmt.Sprintf("PRAGMA busy_timeout=%d", sqliteBusyTimeoutMS), "PRAGMA synchronous=NORMAL", + fmt.Sprintf("PRAGMA wal_autocheckpoint=%d", sqliteWALPages), fmt.Sprintf("PRAGMA cache_size=-%d", sqliteCacheKiB), "PRAGMA foreign_keys=ON", "PRAGMA temp_store=MEMORY", -- 2.51.2