feat: SQLite VFS Pool - share WASM instances across actors

rivetkit-typescript/packages/rivetkit/src/actor/driver.ts

@@ -5,7 +5,7 @@ import { type AnyConn } from "./conn/mod";
 import type { AnyActorInstance } from "./instance/mod";
 import type { RegistryConfig } from "@/registry/config";
 import type { RawDatabaseClient } from "@/db/config";
-import type { SqliteVfs } from "@rivetkit/sqlite-vfs";
+import type { ISqliteVfs } from "@rivetkit/sqlite-vfs";
 import { BaseSQLiteDatabase } from "drizzle-orm/sqlite-core";
 
 export type ActorDriverBuilder = (
@@ -92,13 +92,12 @@ export interface ActorDriver {
 	 * Creates a SQLite VFS instance for creating KV-backed databases.
 	 * If not provided, the database provider will need an override.
 	 *
-	 * @rivetkit/sqlite's async build is not re-entrant per module instance. Drivers
-	 * should return a new instance per call for actor-level isolation.
+	 * Drivers may use the actorId for pool-based instance assignment.
 	 *
 	 * This is a method (not a property) so drivers can use dynamic imports,
 	 * keeping the core driver tree-shakeable from @rivetkit/sqlite.
 	 */
-	createSqliteVfs?(): SqliteVfs | Promise<SqliteVfs>;
+	createSqliteVfs?(actorId: string): ISqliteVfs | Promise<ISqliteVfs>;
 
 	/**
 	 * Requests the actor to go to sleep.

rivetkit-typescript/packages/rivetkit/src/actor/instance/mod.ts

@@ -1,4 +1,4 @@
 import type { OtlpExportTraceServiceRequestJson } from "@rivetkit/traces";
 import {
 	createNoopTraces,
 	createTraces,
@@ -6,7 +6,7 @@
 	type SpanStatusInput,
 	type Traces,
 } from "@rivetkit/traces";
-import type { SqliteVfs } from "@rivetkit/sqlite-vfs";
+import type { ISqliteVfs } from "@rivetkit/sqlite-vfs";
 import invariant from "invariant";
 import type { ActorKey } from "@/actor/mod";
 import type { Client } from "@/client/client";
@@ -171,7 +171,7 @@
 	// MARK: - Variables & Database
 	#vars?: V;
 	#db?: InferDatabaseClient<DB>;
-	#sqliteVfs?: SqliteVfs;
+	#sqliteVfs?: ISqliteVfs;
 
 	// MARK: - Background Tasks
 	#backgroundPromises: Promise<void>[] = [];
@@ -1471,11 +1471,10 @@
 
 		let client: InferDatabaseClient<DB> | undefined;
 		try {
-			// Every actor gets its own SqliteVfs/@rivetkit/sqlite instance. The async
-			// @rivetkit/sqlite build is not re-entrant, and sharing one instance across
-			// actors can cause cross-actor contention and runtime corruption.
+			// Acquire a SQLite VFS handle for this actor. The driver may return a
+			// standalone VFS or a pooled handle that shares a WASM instance.
 			if (!this.#sqliteVfs && this.driver.createSqliteVfs) {
-				this.#sqliteVfs = await this.driver.createSqliteVfs();
+				this.#sqliteVfs = await this.driver.createSqliteVfs(this.#actorId);
 			}
 
 			client = await this.#config.db.createClient({

rivetkit-typescript/packages/rivetkit/src/db/config.ts

@@ -1,4 +1,4 @@
-import type { SqliteVfs } from "@rivetkit/sqlite-vfs";
+import type { ISqliteVfs } from "@rivetkit/sqlite-vfs";
 
 export type AnyDatabaseProvider = DatabaseProvider<any> | undefined;
 
@@ -35,11 +35,10 @@ export interface DatabaseProviderContext {
 	};
 
 	/**
-	 * SQLite VFS instance for creating KV-backed databases.
-	 * This should be actor-scoped because @rivetkit/sqlite is not re-entrant per
-	 * module instance.
+	 * SQLite VFS handle for creating KV-backed databases.
+	 * May be a standalone VFS or a pooled handle from SqliteVfsPool.
 	 */
-	sqliteVfs?: SqliteVfs;
+	sqliteVfs?: ISqliteVfs;
 }
 
 export type DatabaseProvider<DB extends RawAccess> = {

rivetkit-typescript/packages/rivetkit/src/db/drizzle/mod.ts

@@ -1,4 +1,4 @@
 import type { Database } from "@rivetkit/sqlite-vfs";
 import {
 	drizzle as proxyDrizzle,
 	type SqliteRemoteDatabase,
@@ -40,27 +40,27 @@
 		params: any[],
 		method: "run" | "all" | "values" | "get",
 	): Promise<{ rows: any }> => {
-		return mutex.run(async () => {
+		await mutex.run(async () => {
 			if (isClosed()) {
 				throw new Error("database is closed");
 			}
+		});
 
-			if (method === "run") {
-				await waDb.run(sql, toSqliteBindings(params));
-				return { rows: [] };
-			}
+		if (method === "run") {
+			await waDb.run(sql, toSqliteBindings(params));
+			return { rows: [] };
+		}
 
-			// For all/get/values, use parameterized query
-			const result = await waDb.query(sql, toSqliteBindings(params));
+		// For all/get/values, use parameterized query
+		const result = await waDb.query(sql, toSqliteBindings(params));
 
-			// drizzle's mapResultRow accesses rows by column index (positional arrays)
-			// so we return raw arrays for all methods
-			if (method === "get") {
-				return { rows: result.rows[0] };
-			}
+		// drizzle's mapResultRow accesses rows by column index (positional arrays)
+		// so we return raw arrays for all methods
+		if (method === "get") {
+			return { rows: result.rows[0] };
+		}
 
-			return { rows: result.rows };
-		});
+		return { rows: result.rows };
 	};
 }
 
@@ -70,29 +70,24 @@
  */
 async function runInlineMigrations(
 	waDb: Database,
-	mutex: AsyncMutex,
 	migrations: any,
 ): Promise<void> {
 	// Create migrations table
-	await mutex.run(() =>
-		waDb.exec(`
+	await waDb.exec(`
 		CREATE TABLE IF NOT EXISTS __drizzle_migrations (
 			id INTEGER PRIMARY KEY AUTOINCREMENT,
 			hash TEXT NOT NULL,
 			created_at INTEGER
 		)
-	`),
-	);
+	`);
 
 	// Get the last applied migration
 	let lastCreatedAt = 0;
-	await mutex.run(() =>
-		waDb.exec(
-			"SELECT id, hash, created_at FROM __drizzle_migrations ORDER BY created_at DESC LIMIT 1",
-			(row) => {
-				lastCreatedAt = Number(row[2]) || 0;
-			},
-		),
+	await waDb.exec(
+		"SELECT id, hash, created_at FROM __drizzle_migrations ORDER BY created_at DESC LIMIT 1",
+		(row) => {
+			lastCreatedAt = Number(row[2]) || 0;
+		},
 	);
 
 	// Apply pending migrations from journal entries
@@ -109,14 +104,12 @@
 		if (!sql) continue;
 
 		// Execute migration SQL
-		await mutex.run(() => waDb.exec(sql));
+		await waDb.exec(sql);
 
 		// Record migration
-		await mutex.run(() =>
-			waDb.run(
-				"INSERT INTO __drizzle_migrations (hash, created_at) VALUES (?, ?)",
-				[entry.tag, entry.when],
-			),
+		await waDb.run(
+			"INSERT INTO __drizzle_migrations (hash, created_at) VALUES (?, ?)",
+			[entry.tag, entry.when],
 		);
 	}
 }
@@ -165,63 +158,61 @@
 					query: string,
 					...args: unknown[]
 				): Promise<TRow[]> => {
-					return mutex.run(async () => {
-						ensureOpen();
-
-						if (args.length > 0) {
-							const result = await waDb.query(
-								query,
-								toSqliteBindings(args),
-							);
-							return result.rows.map((row: unknown[]) => {
-								const obj: Record<string, unknown> = {};
-								for (
-									let i = 0;
-									i < result.columns.length;
-									i++
-								) {
-									obj[result.columns[i]] = row[i];
-								}
-								return obj;
-							}) as TRow[];
-						}
-						// Use exec for non-parameterized queries since
-						// @rivetkit/sqlite's query() can crash on some statements.
-						const results: Record<string, unknown>[] = [];
-						let columnNames: string[] | null = null;
-						await waDb.exec(
+					await mutex.run(async () => { ensureOpen(); });
+
+					if (args.length > 0) {
+						const result = await waDb.query(
 							query,
-							(row: unknown[], columns: string[]) => {
-								if (!columnNames) {
-									columnNames = columns;
-								}
-								const obj: Record<string, unknown> = {};
-								for (let i = 0; i < row.length; i++) {
-									obj[columnNames[i]] = row[i];
-								}
-								results.push(obj);
-							},
+							toSqliteBindings(args),
 						);
-						return results as TRow[];
-					});
+						return result.rows.map((row: unknown[]) => {
+							const obj: Record<string, unknown> = {};
+							for (
+								let i = 0;
+								i < result.columns.length;
+								i++
+							) {
+								obj[result.columns[i]] = row[i];
+							}
+							return obj;
+						}) as TRow[];
+					}
+					// Use exec for non-parameterized queries since
+					// @rivetkit/sqlite's query() can crash on some statements.
+					const results: Record<string, unknown>[] = [];
+					let columnNames: string[] | null = null;
+					await waDb.exec(
+						query,
+						(row: unknown[], columns: string[]) => {
+							if (!columnNames) {
+								columnNames = columns;
+							}
+							const obj: Record<string, unknown> = {};
+							for (let i = 0; i < row.length; i++) {
+								obj[columnNames[i]] = row[i];
+							}
+							results.push(obj);
+						},
+					);
+					return results as TRow[];
 				},
 				close: async () => {
-					await mutex.run(async () => {
-						if (closed) {
-							return;
-						}
+					const shouldClose = await mutex.run(async () => {
+						if (closed) return false;
 						closed = true;
+						return true;
+					});
+					if (shouldClose) {
 						await waDb.close();
 						waDbInstance = null;
-					});
+					}
 				},
 			} satisfies RawAccess);
 		},
 		onMigrate: async (_client) => {
 			if (config?.migrations && waDbInstance) {
 				await runInlineMigrations(
 					waDbInstance,
-					mutex,
 					config.migrations,
 				);
 			}