refactor(rivetkit): centralize internal sqlite storage

CLAUDE.md

@@ -139,6 +139,7 @@ docker-compose up -d
 ### RivetKit Test Fixtures
 
 - Core tests that touch the `_RIVET_TEST_INSPECTOR_TOKEN` env override must share a process-wide lock with startup tests that assert inspector-token initialization side effects; otherwise parallel `cargo test` runs can flip `init_inspector_token(...)` between the env-override no-op path and the stored-token initialization path.
+- Every new or modified internal SQLite `SELECT`, `UPDATE`, or `DELETE` in `rivetkit-core` must add or update a query-efficiency test using the production SQL. Use representative cardinality to reject full scans, temporary sorts, and automatic indexes where indexed access is expected; allowlist intentional scans in the catalog with a concrete reason and production bound. Simple inserts and schema DDL need correctness or migration coverage but no plan assertion. Assert semantic plan properties rather than exact `EXPLAIN QUERY PLAN` text, and run `cargo test -p rivetkit-core sql_efficiency --lib` after internal schema, index, or query changes. This policy excludes user-provided SQL.
 - For the fast static/http/bare driver verifier, pass only the files listed under `## Fast Tests` in `~/.agents/notes/driver-test-progress.md`; `tests/driver/*.test.ts` also pulls in slow-suite files and gives bogus gate failures.
 - Wasm host smoke tests can drive `buildNativeFactory` through `WasmCoreRuntime` fake bindings to cover actor callbacks, KV, state serialization, remote SQLite routing, and NAPI import boundaries without checked-in wasm-pack output.
 - When moving Rust inline tests out of `src/`, keep a tiny source-owned `#[cfg(test)] #[path = "..."] mod tests;` shim so the moved file still has private module access without widening runtime visibility. Prefer a dedicated moved-test file per source module; reusing stale shared `tests/modules/*.rs` files can silently rot against private APIs and explode once you wire them back in.

Cargo.toml

@@ -279,6 +279,12 @@ members = [
     version = "0.4.38"
     features = [ "now" ]
 
+    [workspace.dependencies.chrono-tz]
+    version = "0.10.4"
+
+    [workspace.dependencies.croner]
+    version = "2.2.0"
+
     [workspace.dependencies.clap]
     version = "4.3"
     features = [ "derive", "cargo" ]

engine/packages/depot-client/tests/inline/vfs.rs

@@ -58,7 +58,6 @@
 		vfs_page_cache_capacity_pages: DEFAULT_VFS_PAGE_CACHE_CAPACITY_PAGES / 2,
 		vfs_protected_cache_pages: DEFAULT_VFS_PROTECTED_CACHE_PAGES / 2,
 		vfs_staging_cache_ttl_ms: DEFAULT_VFS_STAGING_CACHE_TTL_MS / 2,
-		vfs_retain_read_cache: true,
 		pager_cache_size_kib: DEFAULT_PAGER_CACHE_SIZE_KIB,
 	};
 
@@ -5679,6 +5678,7 @@
 					request.now_ms,
 					CommitOptions {
 						expected_head_txid: request.expected_head_txid,
+						disable_size_cap: false,
 					},
 				)
 				.await
@@ -6175,6 +6175,77 @@
 	assert!(metrics.request_build_ns + metrics.transport_ns + metrics.state_update_ns > 0);
 }
 
+#[test]
+fn partial_status_index_reduces_storage_and_cold_page_fetches() {
+	let runtime = direct_runtime();
+	let harness = DirectEngineHarness::new();
+	let engine = runtime.block_on(harness.open_engine());
+	let relaxed = std::sync::atomic::Ordering::Relaxed;
+
+	let measure = |actor_id: &str, index_sql: &str| {
+		let db = harness.open_db_on_engine(
+			&runtime,
+			engine.clone(),
+			actor_id,
+			VfsConfig::default(),
+		);
+		sqlite_exec(
+			db.as_ptr(),
+			"CREATE TABLE history (id INTEGER PRIMARY KEY, result INTEGER NOT NULL, payload BLOB NOT NULL);",
+		)
+		.expect("create history table");
+		sqlite_exec(db.as_ptr(), index_sql).expect("create history index");
+		sqlite_exec(
+			db.as_ptr(),
+			"WITH RECURSIVE seq(id) AS (SELECT 1 UNION ALL SELECT id + 1 FROM seq WHERE id < 4000) INSERT INTO history (id, result, payload) SELECT id, CASE WHEN id = 1 THEN 0 ELSE 1 END, randomblob(256) FROM seq;",
+		)
+		.expect("populate history table");
+		let page_count = sqlite_query_i64(db.as_ptr(), "PRAGMA page_count;")
+			.expect("read database page count");
+		drop(db);
+
+		let reopened = harness.open_db_on_engine(
+			&runtime,
+			engine.clone(),
+			actor_id,
+			VfsConfig::default(),
+		);
+		let ctx = direct_vfs_ctx(&reopened);
+		ctx.resolve_pages_fetches.store(0, relaxed);
+		ctx.pages_fetched_total.store(0, relaxed);
+		sqlite_exec(
+			reopened.as_ptr(),
+			"UPDATE history SET payload = randomblob(256) WHERE result = 0;",
+		)
+		.expect("update running history row");
+		(
+			page_count,
+			ctx.resolve_pages_fetches.load(relaxed),
+			ctx.pages_fetched_total.load(relaxed),
+		)
+	};
+
+	let full = measure(
+		&next_test_name("sqlite-full-status-index"),
+		"CREATE INDEX history_result ON history (result);",
+	);
+	let partial = measure(
+		&next_test_name("sqlite-partial-status-index"),
+		"CREATE INDEX history_running ON history (result) WHERE result = 0;",
+	);
+
+	assert!(partial.0 < full.0, "partial index should persist fewer pages");
+	assert!(partial.1 > 0, "cold update should fetch pages from storage");
+	assert!(
+		partial.1 <= full.1,
+		"partial index should not add cold storage round trips: partial={partial:?}, full={full:?}"
+	);
+	assert!(
+		partial.2 <= full.2,
+		"partial index should not fetch more cold pages: partial={partial:?}, full={full:?}"
+	);
+}
+
 #[test]
 fn profile_large_tx_insert_5mb() {
 	// 5MB = 1280 rows x 4KB blobs in one transaction

engine/packages/pegboard-envoy/src/actor_remote_sqlite_task.rs

@@ -28,6 +28,7 @@ impl Key {
 pub(super) enum Message {
 	Exec(protocol::ToRivetSqliteExecRequest),
 	Execute(protocol::ToRivetSqliteExecuteRequest),
+	ExecuteBatch(protocol::ToRivetSqliteExecuteBatchRequest),
 }
 
 pub(super) async fn task(
@@ -52,6 +53,18 @@ pub(super) async fn task(
 				ws_to_tunnel_task::send_sqlite_execute_response(&conn, req.request_id, response)
 					.await?;
 			}
+			Ok(Some(Message::ExecuteBatch(req))) => {
+				let response = ws_to_tunnel_task::handle_remote_sqlite_execute_batch_response(
+					&ctx, &conn, req.data,
+				)
+				.await;
+				ws_to_tunnel_task::send_sqlite_execute_batch_response(
+					&conn,
+					req.request_id,
+					response,
+				)
+				.await?;
+			}
 			Ok(None) | Err(_) => return Ok(TaskExit::RemoteSqlite(key)),
 		}
 	}

engine/packages/pegboard-envoy/src/ws_to_tunnel_task.rs

@@ -609,6 +609,7 @@
 		protocol::ToRivet::ToRivetSqliteCommitRequest(_) => "sqlite_commit",
 		protocol::ToRivet::ToRivetSqliteExecRequest(_) => "sqlite_exec",
 		protocol::ToRivet::ToRivetSqliteExecuteRequest(_) => "sqlite_execute",
+		protocol::ToRivet::ToRivetSqliteExecuteBatchRequest(_) => "sqlite_execute_batch",
 		protocol::ToRivet::ToRivetTunnelMessage(_) => "tunnel_message",
 		protocol::ToRivet::ToRivetMetadata(_) => "metadata",
 		protocol::ToRivet::ToRivetEvents(_) => "events",
@@ -720,6 +721,14 @@
 			task_manager
 				.enqueue_remote_sqlite(key, actor_remote_sqlite_task::Message::Execute(req))?;
 		}
+		protocol::ToRivet::ToRivetSqliteExecuteBatchRequest(req) => {
+			let key =
+				actor_remote_sqlite_task::Key::new(req.data.actor_id.clone(), req.data.generation);
+			task_manager.enqueue_remote_sqlite(
+				key,
+				actor_remote_sqlite_task::Message::ExecuteBatch(req),
+			)?;
+		}
 		protocol::ToRivet::ToRivetTunnelMessage(tunnel_msg) => {
 			let inner_data_len = tunnel_message_inner_data_len(&tunnel_msg.message_kind);
 			if inner_data_len > ctx.config().pegboard().envoy_max_response_payload_size() {
@@ -1140,6 +1149,47 @@
 	response
 }
 
+pub(super) async fn handle_remote_sqlite_execute_batch_response(
+	ctx: &StandaloneCtx,
+	conn: &Conn,
+	request: protocol::SqliteExecuteBatchRequest,
+) -> protocol::SqliteExecuteBatchResponse {
+	let start = Instant::now();
+	let actor_id = request.actor_id.clone();
+	let request_bytes = request
+		.statements
+		.iter()
+		.map(|statement| statement.sql.len() + bind_params_bytes(statement.params.as_ref()))
+		.sum();
+	let response = match handle_remote_sqlite_execute_batch(ctx, conn, request).await {
+		Ok(results) => protocol::SqliteExecuteBatchResponse::SqliteExecuteBatchOk(
+			protocol::SqliteExecuteBatchOk {
+				results: results.into_iter().map(protocol_execute_result).collect(),
+			},
+		),
+		Err(err) => {
+			tracing::error!(actor_id = %actor_id, ?err, "remote sqlite execute batch request failed");
+			protocol::SqliteExecuteBatchResponse::SqliteErrorResponse(sqlite_error_response(&err))
+		}
+	};
+	record_sqlite_request_metrics(
+		conn,
+		"execute_batch",
+		sqlite_execute_batch_response_kind(&response),
+		start,
+	);
+	record_sqlite_payload_bytes(conn, "execute_batch", "request", request_bytes);
+	if let protocol::SqliteExecuteBatchResponse::SqliteExecuteBatchOk(ok) = &response {
+		record_sqlite_payload_bytes(
+			conn,
+			"execute_batch",
+			"response",
+			ok.results.iter().map(execute_result_bytes).sum(),
+		);
+	}
+	response
+}
+
 pub(super) async fn ack_commands(
 	ctx: &StandaloneCtx,
 	namespace_id: Id,
@@ -1552,6 +1602,15 @@
 	}
 }
 
+fn sqlite_execute_batch_response_kind(
+	response: &protocol::SqliteExecuteBatchResponse,
+) -> &'static str {
+	match response {
+		protocol::SqliteExecuteBatchResponse::SqliteExecuteBatchOk(_) => "ok",
+		protocol::SqliteExecuteBatchResponse::SqliteErrorResponse(_) => "error",
+	}
+}
+
 fn record_sqlite_request_metrics(
 	conn: &Conn,
 	request_type: &'static str,
@@ -1658,6 +1717,58 @@
 	database.execute(request.sql, params).await
 }
 
+async fn handle_remote_sqlite_execute_batch(
+	ctx: &StandaloneCtx,
+	conn: &Conn,
+	request: protocol::SqliteExecuteBatchRequest,
+) -> Result<Vec<ExecuteResult>> {
+	validate_remote_sqlite_actor(
+		ctx,
+		conn,
+		&request.namespace_id,
+		&request.actor_id,
+		request.generation,
+	)
+	.await?;
+	validate_remote_sqlite_batch(&request.statements)?;
+
+	let actor_db = actor_db(ctx, conn, request.actor_id.clone()).await?;
+	let database = remote_sqlite_executor_from_parts(
+		&conn.remote_sqlite_executors,
+		actor_db,
+		&request.actor_id,
+		request.generation,
+	)
+	.await?;
+	database.exec("BEGIN".to_owned()).await?;
+	let mut results = Vec::with_capacity(request.statements.len());
+	for statement in request.statements {
+		let params = statement
+			.params
+			.map(|params| params.into_iter().map(bind_param_from_protocol).collect());
+		match database.execute(statement.sql, params).await {
+			Ok(result) => results.push(result),
+			Err(error) => {
+				return match database.exec("ROLLBACK".to_owned()).await {
+					Ok(_) => Err(error.context("execute remote sqlite batch statement")),
+					Err(rollback_error) => Err(error
+						.context("execute remote sqlite batch statement")
+						.context(rollback_error.context("rollback remote sqlite batch"))),
+				};
+			}
+		}
+	}
+	if let Err(error) = database.exec("COMMIT".to_owned()).await {
+		return match database.exec("ROLLBACK".to_owned()).await {
+			Ok(_) => Err(error.context("commit remote sqlite batch")),
+			Err(rollback_error) => Err(error
+				.context("commit remote sqlite batch")
+				.context(rollback_error.context("rollback remote sqlite batch after failed commit"))),
+		};
+	}
+	Ok(results)
+}
+
 async fn validate_remote_sqlite_actor(
 	ctx: &StandaloneCtx,
 	conn: &Conn,
@@ -1848,6 +1959,13 @@
 	Ok(())
 }
 
+fn validate_remote_sqlite_batch(statements: &[protocol::SqliteBatchStatement]) -> Result<()> {
+	for statement in statements {
+		validate_remote_sqlite_params(statement.params.as_ref())?;
+	}
+	Ok(())
+}
+
 fn bind_params_bytes(params: Option<&Vec<protocol::SqliteBindParam>>) -> usize {
 	params.map_or(0, |params| {
 		params.iter().map(bind_param_bytes).sum::<usize>()
@@ -2176,6 +2294,21 @@
 	.await
 }
 
+pub(super) async fn send_sqlite_execute_batch_response(
+	conn: &Conn,
+	request_id: u32,
+	data: protocol::SqliteExecuteBatchResponse,
+) -> Result<()> {
+	send_to_envoy(
+		conn,
+		protocol::ToEnvoy::ToEnvoySqliteExecuteBatchResponse(
+			protocol::ToEnvoySqliteExecuteBatchResponse { request_id, data },
+		),
+		"sqlite execute batch response",
+	)
+	.await
+}
+
 async fn send_to_envoy(conn: &Conn, msg: protocol::ToEnvoy, description: &str) -> Result<()> {
 	let serialized = versioned::ToEnvoy::wrap_latest(msg)
 		.serialize(conn.protocol_version)