diff --git a/CLAUDE.md b/CLAUDE.md index b99d4a4..b1334d7 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -173,7 +173,8 @@ db.prepare("SELECT * FROM streams WHERE sender = @sender").run({ sender: "G..." ## Pragmas and Performance -SQLite WAL mode is already enabled in `db.ts` (line 24). If issues #360 mentions additional pragmas, they should be added to the DB init: +SQLite WAL mode and related pragmas are applied in `backend/src/services/sqlite/` (see `docs/adr/0006-sqlite-wal-and-pool-tuning.md`): +- `PRAGMA journal_mode=WAL`: Concurrent reads during writes - `PRAGMA synchronous=NORMAL`: Balance durability and speed - `PRAGMA busy_timeout=5000`: Prevent SQLITE_BUSY on concurrent writes - `PRAGMA cache_size=-64000`: 64MB page cache for read perf diff --git a/README.md b/README.md index 1e154a9..6e670ac 100644 --- a/README.md +++ b/README.md @@ -424,6 +424,7 @@ Copy `backend/.env.example` to `backend/.env` and `frontend/.env.example` to `fr | `NETWORK_PASSPHRASE` | Optional (Default: `Test SDF Network ; September 2015`) | Non-empty string passphrase matching Stellar network | `Test SDF Network ; September 2015` | Network identifier passphrase | | `ALLOWED_ASSETS` | Optional (Default: `USDC,XLM`) | Comma-separated list of 1–12 alphanumeric asset codes | `USDC,XLM,EURC` | Supported token assets for payment streams | | `DB_PATH` | Optional (Default: `backend/data/streams.db`) | Valid filesystem file path string | `backend/data/streams.db` | SQLite database file location | +| `SQLITE_READ_POOL_SIZE` | Optional (Default: `1`, max `4`) | Integer | `1` | Extra readonly SQLite connections for WAL reads (`0` uses the writer only) | | `WEBHOOK_DESTINATION_URL` | Optional | Valid HTTP/HTTPS URL (`z.string().url()`) | `https://example.com/webhooks/stellar` | Destination URL for outbound stream event webhooks | | `WEBHOOK_SIGNING_SECRET` | Optional (Recommended if URL set) | Secret string ($\ge 32$ random characters recommended) | `whsec_9a8b7c6d5e4f3a2b1c0d9e8f7a6b5c4d` | Secret used to sign webhook payloads via HMAC-SHA256 | | `JWT_SECRET` | Optional (Required in Production) | Secret string ($\ge 32$ random characters recommended) | `jwt_sec_8f7e6d5c4b3a2f1e0d9c8b7a6f5e4d3c` | Secret used for signing JWT authentication tokens | diff --git a/backend/eslint.config.mjs b/backend/eslint.config.mjs index ba653c8..d8f4ddd 100644 --- a/backend/eslint.config.mjs +++ b/backend/eslint.config.mjs @@ -18,6 +18,7 @@ export default tseslint.config( 'src/services/auth.ts', 'src/services/cache.ts', 'src/services/db.ts', + 'src/services/sqlite/connection-pool.ts', 'src/services/eventHistory.ts', 'src/services/indexer.ts', 'src/services/metricsHistory.ts', diff --git a/backend/src/db/concurrency.test.ts b/backend/src/db/concurrency.test.ts index a810eee..fcaf773 100644 --- a/backend/src/db/concurrency.test.ts +++ b/backend/src/db/concurrency.test.ts @@ -5,6 +5,7 @@ import path from "path"; import { Worker } from "worker_threads"; import { afterEach, beforeEach, describe, expect, it } from "vitest"; import { runMigrations } from "../services/migrations"; +import { applySqlitePragmas } from "../services/sqlite/apply-pragmas"; function createTempDbPath(): string { return path.join( @@ -36,22 +37,19 @@ describe("SQLite WAL mode and concurrent read/write safety", () => { it("verifies PRAGMA journal_mode = wal on connection", () => { const db = new Database(dbPath); - db.pragma("journal_mode = WAL"); - const result = db.pragma("journal_mode"); + const applied = applySqlitePragmas(db); db.close(); - expect(result).toEqual([{ journal_mode: "wal" }]); + expect(applied.journalMode).toBe("wal"); + expect(applied.busyTimeoutMs).toBe(5000); }); it("does not throw SQLITE_BUSY during concurrent read and write", () => { const writerDb = new Database(dbPath); - writerDb.pragma("journal_mode = WAL"); - writerDb.pragma("busy_timeout = 5000"); - writerDb.pragma("synchronous = NORMAL"); + applySqlitePragmas(writerDb); runMigrations(writerDb); const readerDb = new Database(dbPath); - readerDb.pragma("journal_mode = WAL"); - readerDb.pragma("busy_timeout = 5000"); + applySqlitePragmas(readerDb); try { writerDb @@ -106,13 +104,11 @@ describe("SQLite WAL mode and concurrent read/write safety", () => { it("returns consistent data when reader reads during active write transaction", () => { const writerDb = new Database(dbPath); - writerDb.pragma("journal_mode = WAL"); - writerDb.pragma("busy_timeout = 5000"); + applySqlitePragmas(writerDb); runMigrations(writerDb); const readerDb = new Database(dbPath); - readerDb.pragma("journal_mode = WAL"); - readerDb.pragma("busy_timeout = 5000"); + applySqlitePragmas(readerDb); try { writerDb @@ -154,8 +150,7 @@ describe("SQLite WAL mode and concurrent read/write safety", () => { it("handles true concurrent read/write via worker threads without SQLITE_BUSY", async () => { const setupDb = new Database(dbPath); - setupDb.pragma("journal_mode = WAL"); - setupDb.pragma("busy_timeout = 5000"); + applySqlitePragmas(setupDb); runMigrations(setupDb); setupDb diff --git a/backend/src/services/db.ts b/backend/src/services/db.ts index b730ac9..6c034a1 100644 --- a/backend/src/services/db.ts +++ b/backend/src/services/db.ts @@ -1,11 +1,14 @@ -import Database from "better-sqlite3"; import path from "path"; import { runMigrations } from "./migrations"; +import { SQLITE_DEFAULT_READ_POOL_SIZE } from "./sqlite/constants"; +import { SqliteConnectionPool } from "./sqlite/connection-pool"; -const DB_PATH = - process.env.DB_PATH || path.join(__dirname, "..", "..", "data", "streams.db"); +function resolveDbPath(): string { + return process.env.DB_PATH || path.join(__dirname, "..", "..", "data", "streams.db"); +} let db: any; +let sqlitePool: SqliteConnectionPool | null = null; export function getDb(): any { if (!db) { @@ -14,6 +17,26 @@ export function getDb(): any { return db; } +export function getReadDb(): any { + if (sqlitePool) { + return sqlitePool.getReader(); + } + return getDb(); +} + +export function closeDb(): void { + if (sqlitePool) { + sqlitePool.close(); + sqlitePool = null; + db = undefined; + return; + } + if (db) { + db.close(); + db = undefined; + } +} + export function isPostgres(): boolean { return !!process.env.DATABASE_URL; } @@ -259,39 +282,37 @@ class PostgresDatabase { } public prepare(sql: string): any { - const dbInstance = this; return { - run(...params: any[]): any { + run: (...params: any[]): any => { const paramObj = params.length === 1 && typeof params[0] === "object" && params[0] !== null && !Array.isArray(params[0]) ? params[0] : params; - const res = dbInstance.querySync(sql, paramObj); + const res = this.querySync(sql, paramObj); return { changes: res.rowCount, lastInsertRowid: 0, }; }, - get(...params: any[]): any { + get: (...params: any[]): any => { const paramObj = params.length === 1 && typeof params[0] === "object" && params[0] !== null && !Array.isArray(params[0]) ? params[0] : params; - const res = dbInstance.querySync(sql, paramObj); + const res = this.querySync(sql, paramObj); return res.rows[0] || undefined; }, - all(...params: any[]): any { + all: (...params: any[]): any => { const paramObj = params.length === 1 && typeof params[0] === "object" && params[0] !== null && !Array.isArray(params[0]) ? params[0] : params; - const res = dbInstance.querySync(sql, paramObj); + const res = this.querySync(sql, paramObj); return res.rows; }, }; } - public transaction(fn: Function): any { - const dbInstance = this; + public transaction(fn: (...args: any[]) => any): any { return (...args: any[]) => { - dbInstance.exec("BEGIN"); + this.exec("BEGIN"); try { const result = fn(...args); - dbInstance.exec("COMMIT"); + this.exec("COMMIT"); return result; } catch (error) { - dbInstance.exec("ROLLBACK"); + this.exec("ROLLBACK"); throw error; } }; @@ -308,22 +329,33 @@ class PostgresDatabase { } } +function resolveReadPoolSize(): number { + const raw = process.env.SQLITE_READ_POOL_SIZE; + if (!raw) { + return SQLITE_DEFAULT_READ_POOL_SIZE; + } + const parsed = Number.parseInt(raw, 10); + return Number.isFinite(parsed) ? parsed : SQLITE_DEFAULT_READ_POOL_SIZE; +} + export function initDb(): void { + closeDb(); + if (isPostgres()) { db = new PostgresDatabase(process.env.DATABASE_URL!); } else { - const dir = path.dirname(DB_PATH); + const dbPath = resolveDbPath(); + const dir = path.dirname(dbPath); const fs = require("fs"); if (!fs.existsSync(dir)) { fs.mkdirSync(dir, { recursive: true }); } - db = new Database(DB_PATH); - db.pragma("journal_mode = WAL"); - db.pragma("foreign_keys = ON"); - db.pragma("synchronous = NORMAL"); - db.pragma("busy_timeout = 5000"); - db.pragma("cache_size = -64000"); + sqlitePool = new SqliteConnectionPool({ + filePath: dbPath, + readPoolSize: resolveReadPoolSize(), + }); + db = sqlitePool.getWriter(); } runMigrations(db); diff --git a/backend/src/services/sqlite/apply-pragmas.test.ts b/backend/src/services/sqlite/apply-pragmas.test.ts new file mode 100644 index 0000000..b795a08 --- /dev/null +++ b/backend/src/services/sqlite/apply-pragmas.test.ts @@ -0,0 +1,78 @@ +import Database from "better-sqlite3"; +import fs from "fs"; +import os from "os"; +import path from "path"; +import { afterEach, describe, expect, it } from "vitest"; +import { + applySqlitePragmas, + DEFAULT_SQLITE_PRAGMA_SETTINGS, + readSqlitePragmas, +} from "./apply-pragmas"; +import { + SQLITE_BUSY_TIMEOUT_MS, + SQLITE_CACHE_SIZE_KIB, + SqliteConnectionRole, +} from "./constants"; + +function tempDbPath(): string { + return path.join( + os.tmpdir(), + `stellar-stream-pragmas-${Date.now()}-${Math.random().toString(36).slice(2)}.db`, + ); +} + +describe("applySqlitePragmas", () => { + const paths: string[] = []; + + afterEach(() => { + for (const filePath of paths) { + for (const suffix of ["", "-wal", "-shm"]) { + try { + fs.unlinkSync(filePath + suffix); + } catch {} + } + } + paths.length = 0; + }); + + it("enables WAL, busy timeout, and tuned cache on a writer connection", () => { + const filePath = tempDbPath(); + paths.push(filePath); + const db = new Database(filePath); + const applied = applySqlitePragmas(db); + + expect(applied.journalMode).toBe("wal"); + expect(applied.busyTimeoutMs).toBe(SQLITE_BUSY_TIMEOUT_MS); + expect(applied.cacheSize).toBe(SQLITE_CACHE_SIZE_KIB); + + db.close(); + }); + + it("applies busy timeout and cache size on a reader without requiring journal writes", () => { + const filePath = tempDbPath(); + paths.push(filePath); + const writer = new Database(filePath); + applySqlitePragmas(writer, DEFAULT_SQLITE_PRAGMA_SETTINGS, SqliteConnectionRole.Writer); + const reader = new Database(filePath, { readonly: true }); + const applied = applySqlitePragmas( + reader, + DEFAULT_SQLITE_PRAGMA_SETTINGS, + SqliteConnectionRole.Reader, + ); + + expect(applied.journalMode).toBe("wal"); + expect(applied.busyTimeoutMs).toBe(SQLITE_BUSY_TIMEOUT_MS); + expect(applied.cacheSize).toBe(SQLITE_CACHE_SIZE_KIB); + reader.close(); + writer.close(); + }); + + it("reads back the same pragma snapshot after apply", () => { + const filePath = tempDbPath(); + paths.push(filePath); + const db = new Database(filePath); + const applied = applySqlitePragmas(db); + expect(readSqlitePragmas(db)).toEqual(applied); + db.close(); + }); +}); diff --git a/backend/src/services/sqlite/apply-pragmas.ts b/backend/src/services/sqlite/apply-pragmas.ts new file mode 100644 index 0000000..cfcd02c --- /dev/null +++ b/backend/src/services/sqlite/apply-pragmas.ts @@ -0,0 +1,95 @@ +import { + SQLITE_BUSY_TIMEOUT_MS, + SQLITE_CACHE_SIZE_KIB, + SQLITE_MMAP_SIZE_BYTES, + SQLITE_WAL_AUTOCHECKPOINT_PAGES, + SqliteConnectionRole, + SqliteForeignKeys, + SqliteJournalMode, + SqlitePragma, + SqliteSynchronous, + SqliteTempStore, +} from "./constants"; + +export interface SqliteQueryable { + pragma(source: string): unknown; +} + +export interface SqlitePragmaSettings { + journalMode: SqliteJournalMode; + synchronous: SqliteSynchronous; + busyTimeoutMs: number; + cacheSizeKib: number; + foreignKeys: SqliteForeignKeys; + tempStore: SqliteTempStore; + walAutocheckpointPages: number; + mmapSizeBytes: number; +} + +export const DEFAULT_SQLITE_PRAGMA_SETTINGS: SqlitePragmaSettings = { + journalMode: SqliteJournalMode.Wal, + synchronous: SqliteSynchronous.Normal, + busyTimeoutMs: SQLITE_BUSY_TIMEOUT_MS, + cacheSizeKib: SQLITE_CACHE_SIZE_KIB, + foreignKeys: SqliteForeignKeys.On, + tempStore: SqliteTempStore.Memory, + walAutocheckpointPages: SQLITE_WAL_AUTOCHECKPOINT_PAGES, + mmapSizeBytes: SQLITE_MMAP_SIZE_BYTES, +}; + +export interface AppliedSqlitePragmas { + journalMode: string; + synchronous: number | string; + busyTimeoutMs: number; + cacheSize: number; + foreignKeys: number; + tempStore: number | string; + walAutocheckpoint: number; + mmapSize: number; +} + +function pragmaScalar(db: SqliteQueryable, name: SqlitePragma): string | number { + const result = db.pragma(name); + if (Array.isArray(result) && result.length > 0 && typeof result[0] === "object" && result[0] !== null) { + const values = Object.values(result[0] as Record); + const value = values[0]; + if (typeof value === "number" || typeof value === "string") { + return value; + } + } + if (typeof result === "number" || typeof result === "string") { + return result; + } + throw new Error(`Unexpected PRAGMA ${name} result`); +} + +export function applySqlitePragmas( + db: SqliteQueryable, + settings: SqlitePragmaSettings = DEFAULT_SQLITE_PRAGMA_SETTINGS, + role: SqliteConnectionRole = SqliteConnectionRole.Writer, +): AppliedSqlitePragmas { + if (role === SqliteConnectionRole.Writer) { + db.pragma(`${SqlitePragma.JournalMode} = ${settings.journalMode}`); + db.pragma(`${SqlitePragma.Synchronous} = ${settings.synchronous}`); + db.pragma(`${SqlitePragma.ForeignKeys} = ${settings.foreignKeys}`); + db.pragma(`${SqlitePragma.TempStore} = ${settings.tempStore}`); + db.pragma(`${SqlitePragma.WalAutocheckpoint} = ${settings.walAutocheckpointPages}`); + db.pragma(`${SqlitePragma.MmapSize} = ${settings.mmapSizeBytes}`); + } + db.pragma(`${SqlitePragma.BusyTimeout} = ${settings.busyTimeoutMs}`); + db.pragma(`${SqlitePragma.CacheSize} = ${settings.cacheSizeKib}`); + return readSqlitePragmas(db); +} + +export function readSqlitePragmas(db: SqliteQueryable): AppliedSqlitePragmas { + return { + journalMode: String(pragmaScalar(db, SqlitePragma.JournalMode)).toLowerCase(), + synchronous: pragmaScalar(db, SqlitePragma.Synchronous), + busyTimeoutMs: Number(pragmaScalar(db, SqlitePragma.BusyTimeout)), + cacheSize: Number(pragmaScalar(db, SqlitePragma.CacheSize)), + foreignKeys: Number(pragmaScalar(db, SqlitePragma.ForeignKeys)), + tempStore: pragmaScalar(db, SqlitePragma.TempStore), + walAutocheckpoint: Number(pragmaScalar(db, SqlitePragma.WalAutocheckpoint)), + mmapSize: Number(pragmaScalar(db, SqlitePragma.MmapSize)), + }; +} diff --git a/backend/src/services/sqlite/connection-pool.test.ts b/backend/src/services/sqlite/connection-pool.test.ts new file mode 100644 index 0000000..cc62055 --- /dev/null +++ b/backend/src/services/sqlite/connection-pool.test.ts @@ -0,0 +1,140 @@ +import fs from "fs"; +import os from "os"; +import path from "path"; +import { Worker } from "worker_threads"; +import { afterEach, describe, expect, it } from "vitest"; +import { DEFAULT_SQLITE_PRAGMA_SETTINGS, readSqlitePragmas } from "./apply-pragmas"; +import { SqliteConnectionPool, isFileBackedSqlitePath } from "./connection-pool"; +import { + SQLITE_BUSY_TIMEOUT_MS, + SQLITE_CACHE_SIZE_KIB, +} from "./constants"; + +function tempDbPath(): string { + return path.join( + os.tmpdir(), + `stellar-stream-pool-${Date.now()}-${Math.random().toString(36).slice(2)}.db`, + ); +} + +function cleanupDb(filePath: string): void { + for (const suffix of ["", "-wal", "-shm"]) { + try { + fs.unlinkSync(filePath + suffix); + } catch {} + } +} + +describe("SqliteConnectionPool", () => { + let dbPath: string; + let pool: SqliteConnectionPool | undefined; + + afterEach(() => { + pool?.close(); + pool = undefined; + if (dbPath) { + cleanupDb(dbPath); + } + }); + + it("treats file paths as file-backed and memory paths as not", () => { + expect(isFileBackedSqlitePath("/tmp/streams.db")).toBe(true); + expect(isFileBackedSqlitePath(":memory:")).toBe(false); + }); + + it("opens a writer in WAL mode with busy timeout and cache size", () => { + dbPath = tempDbPath(); + pool = new SqliteConnectionPool({ filePath: dbPath }); + const pragmas = readSqlitePragmas(pool.getWriter()); + + expect(pragmas.journalMode).toBe("wal"); + expect(pragmas.busyTimeoutMs).toBe(SQLITE_BUSY_TIMEOUT_MS); + expect(pragmas.cacheSize).toBe(SQLITE_CACHE_SIZE_KIB); + }); + + it("serves a distinct readonly reader so reads can proceed during writes", () => { + dbPath = tempDbPath(); + pool = new SqliteConnectionPool({ filePath: dbPath, readPoolSize: 1 }); + const writer = pool.getWriter(); + const reader = pool.getReader(); + + expect(reader).not.toBe(writer); + expect(readSqlitePragmas(reader).journalMode).toBe("wal"); + expect(readSqlitePragmas(reader).busyTimeoutMs).toBe(SQLITE_BUSY_TIMEOUT_MS); + + writer.exec("CREATE TABLE items (id INTEGER PRIMARY KEY, value TEXT)"); + writer.exec("INSERT INTO items (value) VALUES ('before')"); + + writer.exec("BEGIN IMMEDIATE"); + writer.exec("UPDATE items SET value = 'during' WHERE id = 1"); + + const snapshot = reader.prepare("SELECT value FROM items WHERE id = 1").get() as { + value: string; + }; + expect(snapshot.value).toBe("before"); + + writer.exec("COMMIT"); + + const committed = reader.prepare("SELECT value FROM items WHERE id = 1").get() as { + value: string; + }; + expect(committed.value).toBe("during"); + }); + + it("waits up to busy_timeout instead of failing immediately on a write lock", async () => { + dbPath = tempDbPath(); + pool = new SqliteConnectionPool({ + filePath: dbPath, + readPoolSize: 0, + pragmas: DEFAULT_SQLITE_PRAGMA_SETTINGS, + }); + pool.getWriter().exec("CREATE TABLE locks (id INTEGER PRIMARY KEY)"); + pool.close(); + pool = undefined; + + const holderCode = ` + const { parentPort, workerData } = require("worker_threads"); + const Database = require("better-sqlite3"); + const db = new Database(workerData.dbPath); + db.pragma("journal_mode = WAL"); + db.pragma("busy_timeout = 5000"); + db.exec("BEGIN EXCLUSIVE"); + parentPort.postMessage("locked"); + setTimeout(() => { + db.exec("COMMIT"); + db.close(); + parentPort.postMessage("released"); + }, 200); + `; + + const worker = new Worker(holderCode, { + eval: true, + workerData: { dbPath }, + }); + + await new Promise((resolve, reject) => { + worker.once("message", (msg: string) => { + if (msg === "locked") { + resolve(); + return; + } + reject(new Error(`Unexpected worker message: ${msg}`)); + }); + worker.once("error", reject); + }); + + const contender = new SqliteConnectionPool({ filePath: dbPath, readPoolSize: 0 }); + const started = Date.now(); + try { + expect(() => { + contender.getWriter().exec("INSERT INTO locks DEFAULT VALUES"); + }).not.toThrow(); + const elapsed = Date.now() - started; + expect(elapsed).toBeGreaterThanOrEqual(150); + expect(elapsed).toBeLessThan(SQLITE_BUSY_TIMEOUT_MS); + } finally { + contender.close(); + await worker.terminate(); + } + }); +}); diff --git a/backend/src/services/sqlite/connection-pool.ts b/backend/src/services/sqlite/connection-pool.ts new file mode 100644 index 0000000..0152f77 --- /dev/null +++ b/backend/src/services/sqlite/connection-pool.ts @@ -0,0 +1,79 @@ +import Database from "better-sqlite3"; +import { + SQLITE_DEFAULT_READ_POOL_SIZE, + SQLITE_MAX_READ_POOL_SIZE, + SqliteConnectionRole, +} from "./constants"; +import { + applySqlitePragmas, + DEFAULT_SQLITE_PRAGMA_SETTINGS, + type SqlitePragmaSettings, +} from "./apply-pragmas"; + +export interface SqlitePoolOptions { + filePath: string; + readPoolSize?: number; + pragmas?: SqlitePragmaSettings; +} + +export function isFileBackedSqlitePath(filePath: string): boolean { + const normalized = filePath.trim().toLowerCase(); + return normalized !== ":memory:" && !normalized.startsWith("file::memory:"); +} + +function resolveReadPoolSize(filePath: string, requested?: number): number { + if (!isFileBackedSqlitePath(filePath)) { + return 0; + } + const size = requested ?? SQLITE_DEFAULT_READ_POOL_SIZE; + return Math.max(0, Math.min(SQLITE_MAX_READ_POOL_SIZE, size)); +} + +export class SqliteConnectionPool { + readonly writer: any; + private readonly readers: any[]; + private readerCursor = 0; + + constructor(options: SqlitePoolOptions) { + const pragmas = options.pragmas ?? DEFAULT_SQLITE_PRAGMA_SETTINGS; + this.writer = new Database(options.filePath); + applySqlitePragmas(this.writer, pragmas, SqliteConnectionRole.Writer); + + const readPoolSize = resolveReadPoolSize(options.filePath, options.readPoolSize); + this.readers = []; + for (let i = 0; i < readPoolSize; i += 1) { + const reader = new Database(options.filePath, { readonly: true }); + applySqlitePragmas(reader, pragmas, SqliteConnectionRole.Reader); + this.readers.push(reader); + } + } + + getWriter(): any { + return this.writer; + } + + getReader(): any { + if (this.readers.length === 0) { + return this.writer; + } + const reader = this.readers[this.readerCursor % this.readers.length]; + this.readerCursor = (this.readerCursor + 1) % this.readers.length; + return reader; + } + + close(): void { + for (const reader of this.readers) { + closeQuietly(reader); + } + this.readers.length = 0; + closeQuietly(this.writer); + } +} + +function closeQuietly(connection: { close: () => void }): void { + try { + connection.close(); + } catch { + return; + } +} diff --git a/backend/src/services/sqlite/constants.ts b/backend/src/services/sqlite/constants.ts new file mode 100644 index 0000000..c80a666 --- /dev/null +++ b/backend/src/services/sqlite/constants.ts @@ -0,0 +1,43 @@ +export enum SqliteJournalMode { + Wal = "WAL", +} + +export enum SqliteSynchronous { + Normal = "NORMAL", +} + +export enum SqliteTempStore { + Memory = "MEMORY", +} + +export enum SqliteForeignKeys { + On = "ON", +} + +export enum SqliteConnectionRole { + Writer = "writer", + Reader = "reader", +} + +export enum SqlitePragma { + JournalMode = "journal_mode", + Synchronous = "synchronous", + BusyTimeout = "busy_timeout", + CacheSize = "cache_size", + ForeignKeys = "foreign_keys", + TempStore = "temp_store", + WalAutocheckpoint = "wal_autocheckpoint", + MmapSize = "mmap_size", +} + +export const SQLITE_BUSY_TIMEOUT_MS = 5000; + +export const SQLITE_CACHE_SIZE_KIB = -64_000; + +export const SQLITE_WAL_AUTOCHECKPOINT_PAGES = 1000; + +export const SQLITE_MMAP_SIZE_BYTES = 67_108_864; + +export const SQLITE_DEFAULT_READ_POOL_SIZE = 1; + +export const SQLITE_MAX_READ_POOL_SIZE = 4; diff --git a/backend/src/services/sqlite/init-db.pragmas.test.ts b/backend/src/services/sqlite/init-db.pragmas.test.ts new file mode 100644 index 0000000..dcc3f33 --- /dev/null +++ b/backend/src/services/sqlite/init-db.pragmas.test.ts @@ -0,0 +1,50 @@ +import fs from "fs"; +import os from "os"; +import path from "path"; +import { afterEach, describe, expect, it } from "vitest"; +import { closeDb, getDb, getReadDb, initDb } from "../db"; +import { readSqlitePragmas } from "./apply-pragmas"; +import { SQLITE_BUSY_TIMEOUT_MS, SQLITE_CACHE_SIZE_KIB } from "./constants"; + +function tempDbPath(): string { + return path.join( + os.tmpdir(), + `stellar-stream-initdb-${Date.now()}-${Math.random().toString(36).slice(2)}.db`, + ); +} + +describe("initDb SQLite pragmas", () => { + let dbPath: string; + let previousDbPath: string | undefined; + + afterEach(() => { + closeDb(); + if (previousDbPath === undefined) { + delete process.env.DB_PATH; + } else { + process.env.DB_PATH = previousDbPath; + } + if (dbPath) { + for (const suffix of ["", "-wal", "-shm"]) { + try { + fs.unlinkSync(dbPath + suffix); + } catch {} + } + } + }); + + it("applies WAL, busy_timeout, and cache_size through initDb", () => { + previousDbPath = process.env.DB_PATH; + dbPath = tempDbPath(); + process.env.DB_PATH = dbPath; + delete process.env.DATABASE_URL; + + initDb(); + const pragmas = readSqlitePragmas(getDb()); + + expect(pragmas.journalMode).toBe("wal"); + expect(pragmas.busyTimeoutMs).toBe(SQLITE_BUSY_TIMEOUT_MS); + expect(pragmas.cacheSize).toBe(SQLITE_CACHE_SIZE_KIB); + expect(getReadDb()).not.toBe(getDb()); + }); +}); diff --git a/docs/LOAD_TESTING.md b/docs/LOAD_TESTING.md index fc943b5..fb55dcd 100644 --- a/docs/LOAD_TESTING.md +++ b/docs/LOAD_TESTING.md @@ -314,7 +314,7 @@ Redis replaces the in-memory cache, providing a shared cache across backend inst ### SQLite Performance -SQLite performance is influenced by pragmas set in `db.ts`: +SQLite performance is influenced by pragmas applied in `backend/src/services/sqlite/` (see [ADR 0006](adr/0006-sqlite-wal-and-pool-tuning.md)): | Pragma | Current Value | Description | |--------|---------------|-------------| @@ -325,7 +325,7 @@ SQLite performance is influenced by pragmas set in `db.ts`: **For write-heavy workloads:** -1. Ensure WAL mode is active (confirmed in `db.ts`) +1. Ensure WAL mode is active (confirmed in `backend/src/services/sqlite/`) 2. Reduce `RECONCILIATION_INTERVAL_MS` (default 60000ms) to update streams more frequently 3. Consider batching mutations (the bulk-cancel endpoint already does this) 4. Monitor `cache_size` — increase to -128000 (128MB) if the database grows large @@ -362,7 +362,7 @@ export NODE_OPTIONS="--max-old-space-size=512" - Long-running indexer or reconciliation jobs contending with API writes **Fixes:** -1. Verify WAL mode: `PRAGMA journal_mode=WAL;` (already enabled in `db.ts`) +1. Verify WAL mode: `PRAGMA journal_mode=WAL;` (already enabled in `backend/src/services/sqlite/`) 2. Ensure `busy_timeout` is set: `PRAGMA busy_timeout=5000;` 3. Reduce write concurrency in load tests (`-c 3` for mutation tests) 4. Consider Redis for multi-instance deployments to reduce direct DB contention diff --git a/docs/adr/0001-sqlite-storage.md b/docs/adr/0001-sqlite-storage.md index 1b08703..8f58fe5 100644 --- a/docs/adr/0001-sqlite-storage.md +++ b/docs/adr/0001-sqlite-storage.md @@ -144,9 +144,9 @@ When the application outgrows SQLite: ### Performance Optimizations -- WAL mode for concurrent reads during writes +- WAL mode for concurrent reads during writes (pragma values and the writer/readonly pool are in [ADR 0006](./0006-sqlite-wal-and-pool-tuning.md)) - Indexes on frequently queried columns (sender, recipient, status) -- Connection pooling via better-sqlite3 +- One writer connection plus optional readonly readers (`getReadDb()`) - Redis caching layer for hot data (stream lists, stats) ## PostgreSQL Support Implementation (June 2026) diff --git a/docs/adr/0006-sqlite-wal-and-pool-tuning.md b/docs/adr/0006-sqlite-wal-and-pool-tuning.md new file mode 100644 index 0000000..6c713c0 --- /dev/null +++ b/docs/adr/0006-sqlite-wal-and-pool-tuning.md @@ -0,0 +1,43 @@ +# ADR 0006: SQLite WAL Mode and Connection Pool Tuning + +**Status:** Accepted +**Date:** 2026-08-28 +**Deciders:** Stellar Stream Team + +## Context + +The default SQLite backend is a single-process, file-backed store used by the API, indexer, and webhook worker. Concurrent reads during writes, lock wait behavior, and page cache size need explicit settings. ADR 0001 selected SQLite; this record captures the connection-level tuning. + +## Decision + +Use Write-Ahead Logging with a small writer-plus-readonly pool and the following pragmas, applied in `backend/src/services/sqlite/`. + +| Setting | Value | Reason | +| --- | --- | --- | +| `journal_mode` | `WAL` | Readers use the WAL snapshot and do not take the write lock. | +| `busy_timeout` | `5000` ms | Writers wait up to five seconds for a lock instead of failing immediately or waiting forever. | +| `cache_size` | `-64000` (64 MiB) | Negative values are KiB. 64 MiB fits typical stream/index working sets without unbounded RSS. | +| `synchronous` | `NORMAL` | Durable enough with WAL; avoids FULL fsync on every commit. | +| `wal_autocheckpoint` | `1000` pages | Caps WAL growth under steady write load. | +| `mmap_size` | `64 MiB` | Speeds repeated reads of hot pages. | +| `temp_store` | `MEMORY` | Keeps sorts and temp tables off disk for short queries. | + +Connection pool: + +- One writer connection for mutations and schema changes (`getDb()`). +- One readonly reader by default (`getReadDb()`), overridable with `SQLITE_READ_POOL_SIZE` (capped at 4). +- In-memory databases skip extra readers because each connection would be a separate database. + +`better-sqlite3` is synchronous and does not implement a generic client pool. The extra readonly handles are the SQLite equivalent: concurrent readers against a single writer under WAL. + +## Consequences + +- Concurrent `SELECT`s on a reader connection observe a consistent snapshot while a writer transaction is open. +- `SQLITE_BUSY` after five seconds is a hard failure, not a hang. +- Memory use grows with `cache_size` and reader count; keep the read pool small on constrained hosts. + +## References + +- [SQLite WAL Mode](https://www.sqlite.org/wal.html) +- [PRAGMA busy_timeout](https://www.sqlite.org/pragma.html#pragma_busy_timeout) +- [PRAGMA cache_size](https://www.sqlite.org/pragma.html#pragma_cache_size)