feat(storage): a local SQLite store several processes can share

apps/e2e/demo-e2e-dts-consumer/e2e/dts-consumer.e2e.spec.ts

@@ -26,6 +26,7 @@ function toFixtureDiagnostic(diagnostic: ts.Diagnostic): FixtureDiagnostic {
 
 const fixtureFiles = [
   'provider-lifecycle-must-compile.ts',
+  'sqlite-storage-must-compile.ts',
   'tool-must-fail.ts',
   'prompts-must-compile.ts',
   'hooks-must-compile.ts',
@@ -104,6 +105,10 @@ describe('@frontmcp/sdk declarations in a strict consumer project', () => {
     expect(describeEach(inFixtureFile('provider-lifecycle-must-compile.ts'))).toEqual([]);
   });
 
+  it('types sqliteStorageProvider() and the storage SqliteStorageToken resolves to', () => {
+    expect(describeEach(inFixtureFile('sqlite-storage-must-compile.ts'))).toEqual([]);
+  });
+
   it('accepts the documented @Prompt forms', () => {
     expect(describeEach(inFixtureFile('prompts-must-compile.ts'))).toEqual([]);
   });

apps/e2e/demo-e2e-dts-consumer/fixture/sqlite-storage-must-compile.ts

@@ -0,0 +1,43 @@
+import 'reflect-metadata';
+
+import {
+  App,
+  createContainer,
+  FrontMcp,
+  sqliteStorageProvider,
+  SqliteStorageToken,
+  type SqliteStorageProviderOptions,
+} from '@frontmcp/sdk';
+
+const options: SqliteStorageProviderOptions = {
+  path: '/var/lib/counter/state.sqlite',
+  walMode: true,
+  busyTimeoutMs: 10_000,
+  prefix: 'counter:',
+};
+
+@App({ name: 'counter' })
+class CounterApp {}
+
+@FrontMcp({
+  info: { name: 'sqlite-storage', version: '1.0.0' },
+  apps: [CounterApp],
+  providers: [sqliteStorageProvider(options)],
+})
+export class SqliteStorageServer {}
+
+// @ts-expect-error the database file's path is required
+sqliteStorageProvider({ walMode: true });
+
+// @ts-expect-error walMode is a boolean
+sqliteStorageProvider({ path: '/var/lib/counter/state.sqlite', walMode: 'yes' });
+
+export async function countVisit(): Promise<number> {
+  const container = await createContainer({ providers: [sqliteStorageProvider(options)] });
+  try {
+    const visits: number = await container.get(SqliteStorageToken).incr('visits');
+    return visits;
+  } finally {
+    await container.dispose();
+  }
+}

apps/e2e/demo-e2e-dts-consumer/fixture/tsconfig.json

@@ -12,6 +12,7 @@
   },
   "files": [
     "provider-lifecycle-must-compile.ts",
+    "sqlite-storage-must-compile.ts",
     "tool-must-fail.ts",
     "prompts-must-compile.ts",
     "hooks-must-compile.ts",

apps/e2e/demo-e2e-local-runtime/e2e/helpers/workers.ts

@@ -0,0 +1,80 @@
+/** Start fixture scripts that follow `fixture/worker-protocol.ts` and drive them line by line from a spec. */
+import { spawn, type ChildProcess } from 'node:child_process';
+import { join } from 'node:path';
+
+const projectRoot = join(__dirname, '..', '..');
+
+export interface Worker<Result> {
+  readonly child: ChildProcess;
+  /** Resolves when the worker prints `line`; rejects if it exits first. */
+  waitFor(line: string): Promise<void>;
+  send(line: string): void;
+  /** What the worker reported; rejects when it exits with a code other than 0. */
+  readonly result: Promise<Result>;
+}
+
+/** Run `fixture/<script>` with tsx from source, with `env` added to this process's environment. */
+export function spawnWorker<Result>(script: string, env: Record<string, string> = {}): Worker<Result> {
+  const child = spawn(process.execPath, ['--import', 'tsx', join(projectRoot, 'fixture', script)], {
+    cwd: projectRoot,
+    env: { ...process.env, ...env },
+    stdio: ['pipe', 'pipe', 'pipe'],
+  });
+  const lines: string[] = [];
+  const waiters: { line: string; resolve: () => void }[] = [];
+  let pending = '';
+  let stderr = '';
+
+  child.stdin?.on('error', () => undefined);
+  child.stderr?.on('data', (chunk: Buffer) => (stderr += chunk.toString()));
+  child.stdout?.on('data', (chunk: Buffer) => {
+    pending += chunk.toString();
+    const complete = pending.split('\n');
+    pending = complete.pop() ?? '';
+    for (const line of complete) {
+      lines.push(line);
+      for (const waiter of waiters.filter((candidate) => candidate.line === line)) waiter.resolve();
+    }
+  });
+
+  const exited = new Promise<number | null>((resolve, reject) => {
+    child.once('error', reject);
+    child.once('close', (code) => resolve(code));
+  });
+
+  const result = exited.then((code) => {
+    const reported = lines.find((line) => line.startsWith('result '));
+    if (code !== 0 || !reported) {
+      throw new Error(`worker ${script} exited with ${code}${reported ? '' : ' without a result'}:\n${stderr}`);
+    }
+    return JSON.parse(reported.slice('result '.length)) as Result;
+  });
+  result.catch(() => undefined);
+
+  return {
+    child,
+    result,
+    send: (line) => child.stdin?.write(`${line}\n`),
+    waitFor: (line) =>
+      lines.includes(line)
+        ? Promise.resolve()
+        : new Promise<void>((resolve, reject) => {
+            waiters.push({ line, resolve });
+            void exited.then((code) =>
+              reject(new Error(`worker ${script} exited with ${code} before "${line}":\n${stderr}`)),
+            );
+          }),
+  };
+}
+
+/** Wait until every worker is ready, start them all at once, and collect what they report. */
+export async function runTogether<Result>(workers: Worker<Result>[]): Promise<Result[]> {
+  await Promise.all(workers.map((worker) => worker.waitFor('ready')));
+  for (const worker of workers) worker.send('go');
+  return Promise.all(workers.map((worker) => worker.result));
+}
+
+/** Kill the workers still running, so a failed spec does not leave processes behind. */
+export function killAll(workers: Worker<unknown>[]): void {
+  for (const { child } of workers) if (child.exitCode === null) child.kill('SIGKILL');
+}

apps/e2e/demo-e2e-local-runtime/e2e/sqlite-contention.e2e.spec.ts

@@ -0,0 +1,74 @@
+/** E2E: ten processes contend on one SQLite file while another holds its write lock. */
+import * as fs from 'node:fs';
+import * as os from 'node:os';
+import * as path from 'node:path';
+
+import { loadBetterSqlite3 } from '@frontmcp/storage-sqlite';
+
+import type { StorageWorkerResult } from '../fixture/storage-worker';
+import { killAll, runTogether, spawnWorker, type Worker } from './helpers/workers';
+
+const WORKERS = 10;
+const INCREMENTS = 40;
+const HOLD_MS = 1000;
+
+function readCounter(dbPath: string): number {
+  const db = new (loadBetterSqlite3())(dbPath, { readonly: true });
+  try {
+    const row = db.prepare("SELECT value FROM kv WHERE key = 'total'").get() as { value: string } | undefined;
+    return Number(row?.value);
+  } finally {
+    db.close();
+  }
+}
+
+describe.each([
+  ['WAL', true],
+  ['rollback journal', false],
+])('ten processes on one SQLite file (%s)', (_mode, walMode) => {
+  let dir: string;
+  let dbPath: string;
+  let started: Worker<unknown>[] = [];
+
+  beforeEach(() => {
+    dir = fs.mkdtempSync(path.join(os.tmpdir(), 'local-runtime-sqlite-'));
+    dbPath = path.join(dir, 'shared.sqlite');
+  });
+
+  afterEach(() => {
+    killAll(started);
+    started = [];
+    fs.rmSync(dir, { recursive: true, force: true });
+  });
+
+  it('counts every increment once, gives the lock to one process, and keeps every event loop running', async () => {
+    const workers = Array.from({ length: WORKERS }, (_, index) =>
+      spawnWorker<StorageWorkerResult>('storage-worker.ts', {
+        DB_PATH: dbPath,
+        WAL_MODE: String(walMode),
+        INCREMENTS: String(INCREMENTS),
+        WORKER_ID: `worker-${index}`,
+      }),
+    );
+    started = workers;
+    await Promise.all(workers.map((worker) => worker.waitFor('ready')));
+
+    const holder = spawnWorker('lock-holder.ts', { DB_PATH: dbPath, HOLD_MS: String(HOLD_MS) });
+    started = [...workers, holder];
+    await holder.waitFor('ready');
+    holder.send('go');
+    await holder.waitFor('locked');
+
+    const results = await runTogether(workers);
+    await holder.result;
+
+    const total = WORKERS * INCREMENTS;
+    const values = results.flatMap((result) => result.incrValues).sort((a, b) => a - b);
+    expect(results.flatMap((result) => result.errors)).toEqual([]);
+    expect(values).toEqual(Array.from({ length: total }, (_, index) => index + 1));
+    expect(readCounter(dbPath)).toBe(total);
+    expect(results.filter((result) => result.holdsNxLock)).toHaveLength(1);
+    expect(results.filter((result) => result.acquiredSlot)).toHaveLength(1);
+    expect(Math.max(...results.map((result) => result.maxTimerGapMs))).toBeLessThan(HOLD_MS / 2);
+  });
+});

apps/e2e/demo-e2e-local-runtime/e2e/sqlite-storage-provider.e2e.spec.ts

@@ -0,0 +1,40 @@
+/** E2E: several processes share the storage `sqliteStorageProvider()` registers, read from `createContainer()`. */
+import * as fs from 'node:fs';
+import * as os from 'node:os';
+import * as path from 'node:path';
+
+import type { ContainerWorkerResult } from '../fixture/container-worker';
+import { killAll, runTogether, spawnWorker, type Worker } from './helpers/workers';
+
+const WORKERS = 3;
+const INCREMENTS = 20;
+
+describe('SqliteStorageToken in several processes', () => {
+  let dir: string;
+  let started: Worker<unknown>[] = [];
+
+  beforeEach(() => {
+    dir = fs.mkdtempSync(path.join(os.tmpdir(), 'local-runtime-provider-'));
+  });
+
+  afterEach(() => {
+    killAll(started);
+    fs.rmSync(dir, { recursive: true, force: true });
+  });
+
+  it('shares one counter between the processes, and each one exits after dispose()', async () => {
+    const workers = Array.from({ length: WORKERS }, () =>
+      spawnWorker<ContainerWorkerResult>('container-worker.ts', {
+        DB_PATH: path.join(dir, 'app.sqlite'),
+        INCREMENTS: String(INCREMENTS),
+      }),
+    );
+    started = workers;
+
+    const results = await runTogether(workers);
+
+    const values = results.flatMap((result) => result.incrValues).sort((a, b) => a - b);
+    expect(values).toEqual(Array.from({ length: WORKERS * INCREMENTS }, (_, index) => index + 1));
+    expect(workers.map(({ child }) => child.exitCode)).toEqual(Array(WORKERS).fill(0));
+  });
+});

apps/e2e/demo-e2e-local-runtime/fixture/container-worker.ts

@@ -0,0 +1,25 @@
+/** On "go": increment a shared counter through `SqliteStorageToken`, dispose the container, report the values. */
+import 'reflect-metadata';
+
+import { createContainer, LogLevel, sqliteStorageProvider, SqliteStorageToken } from '@frontmcp/sdk';
+
+import { readyThenWaitForGo, report, requiredEnv, runWorker } from './worker-protocol';
+
+export interface ContainerWorkerResult {
+  incrValues: number[];
+}
+
+runWorker(async () => {
+  const container = await createContainer({
+    providers: [sqliteStorageProvider({ path: requiredEnv('DB_PATH'), prefix: 'app:' })],
+    logging: { level: LogLevel.Off, enableConsole: false },
+  });
+  const storage = container.get(SqliteStorageToken);
+  await readyThenWaitForGo();
+
+  const incrValues: number[] = [];
+  for (let i = 0; i < Number(requiredEnv('INCREMENTS')); i++) incrValues.push(await storage.incr('total'));
+
+  await container.dispose();
+  report({ incrValues } satisfies ContainerWorkerResult);
+});

apps/e2e/demo-e2e-local-runtime/fixture/lock-holder.ts

@@ -0,0 +1,19 @@
+/** On "go": hold the SQLite write lock for HOLD_MS, printing "locked" once it has it. */
+import { loadBetterSqlite3 } from '@frontmcp/storage-sqlite';
+
+import { readyThenWaitForGo, report, requiredEnv, runWorker, signal } from './worker-protocol';
+
+runWorker(async () => {
+  const BetterSqlite3 = loadBetterSqlite3();
+  const db = new BetterSqlite3(requiredEnv('DB_PATH'));
+  db.pragma('busy_timeout = 5000');
+  await readyThenWaitForGo();
+
+  db.exec('BEGIN IMMEDIATE');
+  db.prepare("INSERT OR REPLACE INTO kv (key, value) VALUES ('held-by', 'lock-holder')").run();
+  signal('locked');
+  await new Promise((resolve) => setTimeout(resolve, Number(requiredEnv('HOLD_MS'))));
+  db.exec('COMMIT');
+  db.close();
+  report({ released: true });
+});

apps/e2e/demo-e2e-local-runtime/fixture/storage-worker.ts

@@ -0,0 +1,68 @@
+/** On "go": increment, take an NX lock and a semaphore slot, and report the errors and longest timer gap seen. */
+import { DistributedSemaphore } from '@frontmcp/guard';
+import { createStorage } from '@frontmcp/utils';
+
+import { readyThenWaitForGo, report, requiredEnv, runWorker } from './worker-protocol';
+
+export interface StorageWorkerResult {
+  incrValues: number[];
+  errors: string[];
+  holdsNxLock: boolean;
+  acquiredSlot: boolean;
+  maxTimerGapMs: number;
+}
+
+const TIMER_INTERVAL_MS = 5;
+
+function watchTimerGaps(): { stop(): number } {
+  let last = performance.now();
+  let maxGap = 0;
+  const tick = () => {
+    const now = performance.now();
+    maxGap = Math.max(maxGap, now - last);
+    last = now;
+  };
+  const timer = setInterval(tick, TIMER_INTERVAL_MS);
+  return {
+    stop() {
+      clearInterval(timer);
+      tick();
+      return Math.round(maxGap);
+    },
+  };
+}
+
+function describeError(error: unknown): string {
+  const code = (error as { code?: unknown } | null)?.code;
+  return `${typeof code === 'string' ? `${code}: ` : ''}${error instanceof Error ? error.message : String(error)}`;
+}
+
+runWorker(async () => {
+  const workerId = requiredEnv('WORKER_ID');
+  const increments = Number(requiredEnv('INCREMENTS'));
+  const storage = await createStorage({
+    type: 'sqlite',
+    fallback: 'error',
+    sqlite: { path: requiredEnv('DB_PATH'), walMode: requiredEnv('WAL_MODE') === 'true', ttlCleanupIntervalMs: 200 },
+  });
+  const semaphore = new DistributedSemaphore(storage, 60);
+  await readyThenWaitForGo();
+
+  const timerGaps = watchTimerGaps();
+  const errors: string[] = [];
+  const incrValues: number[] = [];
+  let holdsNxLock = false;
+  let acquiredSlot = false;
+  try {
+    for (let i = 0; i < increments; i++) incrValues.push(await storage.incr('total'));
+    await storage.set('nx-lock', workerId, { ifNotExists: true, ttlSeconds: 60 });
+    holdsNxLock = (await storage.get('nx-lock')) === workerId;
+    acquiredSlot = (await semaphore.acquire('job', 1, 0, 'job')) !== null;
+  } catch (error) {
+    errors.push(describeError(error));
+  }
+  const maxTimerGapMs = timerGaps.stop();
+
+  await storage.disconnect();
+  report({ incrValues, errors, holdsNxLock, acquiredSlot, maxTimerGapMs } satisfies StorageWorkerResult);
+});