feat(sqlite): add deferred commit mode with flush sequencing

docs-internal/engine/sqlite-vfs.md

@@ -15,6 +15,14 @@ Rules for the SQLite VFS implementation.
 - SQLite VFS v2 storage keys use literal ASCII path segments under the `0x02` subspace prefix with big-endian numeric suffixes so `scan_prefix` and `BTreeMap` ordering stay numerically correct.
 - SQLite v2 slow-path staging writes encoded LTX bytes directly under DELTA chunk keys. Do not expect `/STAGE` keys or a fixed one-chunk-per-page mapping in tests or recovery code.
 
+## Deferred commits
+
+- Deferred mode stages locally committed pages in the VFS overlay and exposes them to reads until the single background flusher receives a durable acknowledgement.
+- Every flush batch carries the durable head fence captured when the batch forms. Retries reuse the same bytes and fence. A lost acknowledgement or divergent head breaks the database because durability is indeterminate.
+- Flush waiters check the terminal error before sequence progress. Once broken, no sequence wait may report success and the core failure monitor stops the actor generation once.
+- Close first stops new work, rolls back an open lease, drains the overlay within the retry deadline, then shuts down the flusher before releasing the VFS.
+- Keep the named structures, algorithms, and invariants synchronized with the [deferred commits specification](sqlite/deferred-commits/SPEC.md).
+
 ## Read-mode/write-mode connection manager
 
 - The native connection manager is the SQLite read/write routing policy boundary. TypeScript and NAPI wrappers forward calls to native execution and must not decide routing from SQL text.

docs/content/docs/sqlite.mdx

@@ -111,7 +111,35 @@ const todoId = c.db.transactionSync((tx) => {
 });
 ```
 
-The callback must be synchronous and must use its `tx` value, which exposes only `executeSync(...)`. It must not return a promise. `{ name, timeout }` options are supported, matching `transaction(...)`. Synchronous operations are unavailable in WebAssembly runtimes.
+The callback must be synchronous and must use its `tx` value. The transaction client exposes `executeSync(...)` plus `commitSeq()`, `flushedSeq()`, `waitForFlush(...)`, and `flushError()`, delegating sequence and durability state to the base database. It must not return a promise. `{ name, timeout }` options are supported, matching `transaction(...)`. Synchronous operations are unavailable in WebAssembly runtimes.
+
+### Deferred commits
+
+Native SQLite normally waits for durable storage before a write returns. Integrations that need synchronous, turn-based storage can opt into deferred commits:
+
+```ts @nocheck
+db: db({
+  commitMode: "deferred",
+  onMigrate: async (db) => {
+    await db.execute("CREATE TABLE IF NOT EXISTS todos (id INTEGER PRIMARY KEY, title TEXT)");
+  },
+})
+```
+
+In deferred mode, a successful write is immediately visible to later queries in the same actor, but it may not be durable yet. Call `await c.db.waitForFlush()` before sending an output that depends on those writes. The call captures the current commit sequence synchronously, so writes started afterward are not part of that wait. You can also capture `c.db.commitSeq()` and pass it explicitly to `waitForFlush(sequence)`.
+
+`c.db.flushError()` returns the terminal durability error after the database has broken, or `null` while it is healthy. Transaction-scoped clients and synchronous transaction handles expose the same sequence, wait, and flush-error methods by delegating them to the base database. `execSync()` returns `{ readonly }`, matching the metadata available from `executeSyncRaw()`.
+
+`beginTransactionSync()` opens a synchronous handle that can stay open until a later event-loop turn. Run all synchronous SQL through the handle, then call `commitSync()` to receive its commit sequence, or `rollbackSync()` to discard it. Base-client synchronous calls throw while the handle is open; asynchronous calls queue behind it.
+
+```ts @nocheck
+const tx = c.db.beginTransactionSync({ name: "todo-turn" });
+tx.executeSync("INSERT INTO todos (title) VALUES (?)", title);
+const sequence = tx.commitSync();
+if (sequence !== null) await c.db.waitForFlush(sequence);
+```
+
+Deferred commits require local SQLite in the Node.js native runtime. Metadata-returning synchronous execution checks that capability before running SQL, so an unsupported remote statement is not executed and then rejected afterward. Actor shutdown drains staged commits. If durability becomes indeterminate, `waitForFlush()` rejects, `flushError()` reports the reason, and the actor generation stops rather than continuing with uncertain storage.
 
 ### Transactions
 

engine/packages/depot-client-embedded/src/lib.rs

@@ -11,7 +11,7 @@ use async_trait::async_trait;
 use depot::error::SqliteStorageError;
 use depot_client::{
 	database::{NativeDatabaseHandle, open_database_from_transport},
-	vfs::{SqliteTransport, SqliteVfsMetrics},
+	vfs::{CommitMode, SqliteTransport, SqliteVfsMetrics},
 };
 use rivet_envoy_protocol as protocol;
 use tokio::runtime::Handle;
@@ -39,6 +39,8 @@ pub async fn open_database_from_embedded_depot(
 		generation,
 		rt_handle,
 		metrics,
+		CommitMode::Awaited,
+		0,
 	)
 	.await
 }

engine/packages/depot-client-types/src/lib.rs

@@ -82,6 +82,7 @@ pub struct ExecResult {
 pub struct QueryResult {
 	pub columns: Vec<String>,
 	pub rows: Vec<Vec<ColumnValue>>,
+	pub readonly: Option<bool>,
 }
 
 #[derive(Clone, Debug, PartialEq)]
@@ -90,13 +91,16 @@ pub struct ExecuteResult {
 	pub rows: Vec<Vec<ColumnValue>>,
 	pub changes: i64,
 	pub last_insert_row_id: Option<i64>,
+	pub readonly: Option<bool>,
+	pub commit_seq: Option<u64>,
 }
 
 impl ExecuteResult {
 	pub fn into_query_result(self) -> QueryResult {
 		QueryResult {
 			columns: self.columns,
 			rows: self.rows,
+			readonly: self.readonly,
 		}
 	}
 
@@ -130,6 +134,8 @@ mod tests {
 			]],
 			changes: 3,
 			last_insert_row_id: Some(42),
+			readonly: Some(false),
+			commit_seq: Some(7),
 		};
 
 		assert_eq!(result.columns, vec!["id", "name"]);
@@ -151,6 +157,8 @@ mod tests {
 			rows: vec![vec![ColumnValue::Integer(9)]],
 			changes: 2,
 			last_insert_row_id: Some(10),
+			readonly: Some(false),
+			commit_seq: Some(8),
 		};
 
 		let query_result = result.clone().into_query_result();

engine/packages/depot-client/Cargo.toml

@@ -26,6 +26,7 @@ depot-client-types.workspace = true
 moka = { version = "0.12", default-features = false, features = ["sync"] }
 parking_lot.workspace = true
 scc.workspace = true
+thiserror.workspace = true
 
 [dev-dependencies]
 depot = { workspace = true, features = ["test-faults"] }

engine/packages/depot-client/src/database.rs

@@ -9,11 +9,14 @@ use tokio::runtime::Handle;
 use crate::{
 	query::{BindParam, ExecResult, ExecuteResult, QueryResult},
 	vfs::{
-		NativeVfsHandle, SqliteOpenPhase, SqliteTransportHandle, SqliteVfs, SqliteVfsMetrics,
-		SqliteVfsMetricsSnapshot, VfsConfig, VfsPreloadHintSnapshot,
-		fetch_initial_pages_for_registration,
+		CommitMode, DatabaseFailure, FlushError, NativeVfsHandle, SqliteOpenPhase,
+		SqliteTransportHandle, SqliteVfs, SqliteVfsMetrics, SqliteVfsMetricsSnapshot, VfsConfig,
+		VfsPreloadHintSnapshot, fetch_initial_pages_for_registration,
+	},
+	worker::{
+		SqliteWorkerCloseTimeoutError, SqliteWorkerFatalError, SqliteWorkerHandle,
+		SqliteWorkerResult,
 	},
-	worker::{SqliteWorkerFatalError, SqliteWorkerHandle, SqliteWorkerResult},
 };
 
 #[derive(Clone)]
@@ -80,10 +83,14 @@ pub async fn open_database_from_transport(
 	generation: u64,
 	rt_handle: Handle,
 	metrics: Option<Arc<dyn SqliteVfsMetrics>>,
+	commit_mode: CommitMode,
+	initial_commit_seq: u64,
 ) -> Result<NativeDatabaseHandle> {
 	let open_timer = SqliteOpenTimer::new(&metrics);
 	let vfs_name = vfs_name_for_actor_database(&actor_id, generation);
-	let config = VfsConfig::default();
+	let mut config = VfsConfig::default();
+	config.commit_mode = commit_mode;
+	config.initial_commit_seq = initial_commit_seq;
 	let transport: SqliteTransportHandle = Arc::new(GenerationFencedTransport {
 		inner: transport,
 		generation,
@@ -253,6 +260,7 @@ impl NativeDatabaseHandle {
 		self.execute(sql, params).await.map(|result| QueryResult {
 			columns: result.columns,
 			rows: result.rows,
+			readonly: result.readonly,
 		})
 	}
 
@@ -281,14 +289,79 @@ impl NativeDatabaseHandle {
 	}
 
 	pub async fn close(&self) -> Result<()> {
-		match self.worker.close().await {
-			Ok(()) => Ok(()),
-			Err(error) => Err(self.fatal_error().unwrap_or(error)),
+		self.close_with_timeouts(None, self.vfs.close_flush_timeout())
+			.await
+	}
+
+	async fn close_with_timeouts(
+		&self,
+		worker_timeout: Option<std::time::Duration>,
+		flush_timeout: std::time::Duration,
+	) -> Result<()> {
+		self.vfs.begin_close();
+		#[cfg(test)]
+		let worker_result = match worker_timeout {
+			Some(timeout) => self.worker.close_with_timeout_for_test(timeout).await,
+			None => self.worker.close().await,
+		};
+		#[cfg(not(test))]
+		let worker_result = {
+			let _ = worker_timeout;
+			self.worker.close().await
+		};
+		if worker_result.as_ref().err().is_some_and(|error| {
+			error
+				.downcast_ref::<SqliteWorkerCloseTimeoutError>()
+				.is_some()
+		}) {
+			self.vfs.abort_flusher_for_worker_timeout().await;
+			return worker_result;
+		}
+		let flush_result = self.vfs.drain_and_shutdown_flusher(flush_timeout).await;
+		match (worker_result, flush_result) {
+			(Ok(()), Ok(())) => Ok(()),
+			(_, Err(error)) => Err(anyhow!(error)),
+			(Err(error), Ok(())) => Err(self.fatal_error().unwrap_or(error)),
+		}
+	}
+
+	#[cfg(test)]
+	pub(crate) async fn close_with_timeouts_for_test(
+		&self,
+		worker_timeout: std::time::Duration,
+		flush_timeout: std::time::Duration,
+	) -> Result<()> {
+		self.close_with_timeouts(Some(worker_timeout), flush_timeout)
+			.await
+	}
+
+	pub async fn wait_for_failure(&self) -> DatabaseFailure {
+		tokio::select! {
+			reason = self.vfs.wait_for_failure() => reason,
+			failed = self.worker.wait_for_failure() => {
+				if failed {
+					DatabaseFailure::WorkerStopped
+				} else {
+					DatabaseFailure::Closed
+				}
+			}
 		}
 	}
 
-	pub async fn wait_for_worker_failure(&self) -> bool {
-		self.worker.wait_for_failure().await
+	pub fn commit_seq(&self) -> u64 {
+		self.vfs.commit_seq()
+	}
+
+	pub fn flushed_seq(&self) -> u64 {
+		self.vfs.flushed_seq()
+	}
+
+	pub fn flush_error(&self) -> Option<FlushError> {
+		self.vfs.flush_error()
+	}
+
+	pub async fn wait_for_flush(&self, seq: u64) -> std::result::Result<(), FlushError> {
+		self.vfs.wait_for_flush(seq).await
 	}
 
 	pub fn take_last_kv_error(&self) -> Option<String> {

engine/packages/depot-client/src/optimization_flags.rs

@@ -24,6 +24,10 @@ pub const VFS_PAGE_CACHE_CAPACITY_PAGES_ENV: &str =
 pub const VFS_PROTECTED_CACHE_PAGES_ENV: &str = "RIVETKIT_SQLITE_OPT_VFS_PROTECTED_CACHE_PAGES";
 pub const VFS_STAGING_CACHE_TTL_MS_ENV: &str = "RIVETKIT_SQLITE_OPT_VFS_STAGING_CACHE_TTL_MS";
 pub const PAGER_CACHE_SIZE_KIB_ENV: &str = "RIVETKIT_SQLITE_OPT_PAGER_CACHE_SIZE_KIB";
+pub const FLUSH_RETRY_DEADLINE_MS_ENV: &str = "RIVETKIT_SQLITE_OPT_FLUSH_RETRY_DEADLINE_MS";
+pub const FLUSH_RETRY_BACKOFF_MIN_MS_ENV: &str = "RIVETKIT_SQLITE_OPT_FLUSH_RETRY_BACKOFF_MIN_MS";
+pub const FLUSH_RETRY_BACKOFF_MAX_MS_ENV: &str = "RIVETKIT_SQLITE_OPT_FLUSH_RETRY_BACKOFF_MAX_MS";
+pub const MAX_UNFLUSHED_BYTES_ENV: &str = "RIVETKIT_SQLITE_OPT_MAX_UNFLUSHED_BYTES";
 
 pub const DEFAULT_STARTUP_PRELOAD_MAX_BYTES: usize = 2 * 1024 * 1024;
 pub const MAX_STARTUP_PRELOAD_MAX_BYTES: usize = 64 * 1024 * 1024;
@@ -37,6 +41,10 @@ pub const DEFAULT_VFS_STAGING_CACHE_TTL_MS: u64 = 30_000;
 pub const MAX_VFS_STAGING_CACHE_TTL_MS: u64 = 300_000;
 pub const DEFAULT_PAGER_CACHE_SIZE_KIB: u64 = 8 * 1024;
 pub const MAX_PAGER_CACHE_SIZE_KIB: u64 = 256 * 1024;
+pub const DEFAULT_FLUSH_RETRY_DEADLINE_MS: u64 = 30_000;
+pub const DEFAULT_FLUSH_RETRY_BACKOFF_MIN_MS: u64 = 50;
+pub const DEFAULT_FLUSH_RETRY_BACKOFF_MAX_MS: u64 = 2_000;
+pub const DEFAULT_MAX_UNFLUSHED_BYTES: usize = 64 * 1024 * 1024;
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum SqliteReadAheadMode {
@@ -110,6 +118,10 @@ pub struct SqliteOptimizationFlags {
 	pub vfs_protected_cache_pages: usize,
 	pub vfs_staging_cache_ttl_ms: u64,
 	pub pager_cache_size_kib: u64,
+	pub flush_retry_deadline_ms: u64,
+	pub flush_retry_backoff_min_ms: u64,
+	pub flush_retry_backoff_max_ms: u64,
+	pub max_unflushed_bytes: usize,
 }
 
 impl Default for SqliteOptimizationFlags {
@@ -138,6 +150,10 @@ impl Default for SqliteOptimizationFlags {
 			vfs_protected_cache_pages: DEFAULT_VFS_PROTECTED_CACHE_PAGES,
 			vfs_staging_cache_ttl_ms: DEFAULT_VFS_STAGING_CACHE_TTL_MS,
 			pager_cache_size_kib: DEFAULT_PAGER_CACHE_SIZE_KIB,
+			flush_retry_deadline_ms: DEFAULT_FLUSH_RETRY_DEADLINE_MS,
+			flush_retry_backoff_min_ms: DEFAULT_FLUSH_RETRY_BACKOFF_MIN_MS,
+			flush_retry_backoff_max_ms: DEFAULT_FLUSH_RETRY_BACKOFF_MAX_MS,
+			max_unflushed_bytes: DEFAULT_MAX_UNFLUSHED_BYTES,
 		}
 	}
 }
@@ -216,6 +232,22 @@ impl SqliteOptimizationFlags {
 				DEFAULT_PAGER_CACHE_SIZE_KIB,
 				MAX_PAGER_CACHE_SIZE_KIB,
 			),
+			flush_retry_deadline_ms: u64_by_default(
+				read_env(FLUSH_RETRY_DEADLINE_MS_ENV).as_deref(),
+				DEFAULT_FLUSH_RETRY_DEADLINE_MS,
+			),
+			flush_retry_backoff_min_ms: u64_by_default(
+				read_env(FLUSH_RETRY_BACKOFF_MIN_MS_ENV).as_deref(),
+				DEFAULT_FLUSH_RETRY_BACKOFF_MIN_MS,
+			),
+			flush_retry_backoff_max_ms: u64_by_default(
+				read_env(FLUSH_RETRY_BACKOFF_MAX_MS_ENV).as_deref(),
+				DEFAULT_FLUSH_RETRY_BACKOFF_MAX_MS,
+			),
+			max_unflushed_bytes: usize_by_default(
+				read_env(MAX_UNFLUSHED_BYTES_ENV).as_deref(),
+				DEFAULT_MAX_UNFLUSHED_BYTES,
+			),
 		}
 	}
 }
@@ -246,6 +278,18 @@ fn usize_bounded_by_default(value: Option<&str>, default: usize, max: usize) ->
 		.min(max)
 }
 
+fn usize_by_default(value: Option<&str>, default: usize) -> usize {
+	value
+		.and_then(|value| value.trim().parse::<usize>().ok())
+		.unwrap_or(default)
+}
+
+fn u64_by_default(value: Option<&str>, default: u64) -> u64 {
+	value
+		.and_then(|value| value.trim().parse::<u64>().ok())
+		.unwrap_or(default)
+}
+
 fn u64_bounded_by_default(value: Option<&str>, default: u64, max: u64) -> u64 {
 	value
 		.and_then(|value| value.trim().parse::<u64>().ok())