fix(sqlite-native): surface kv errors from native sqlite

Author: NathanFlurry

rivetkit-typescript/packages/rivetkit-native/index.d.ts

@@ -59,6 +59,7 @@ export declare function startEnvoySyncJs(config: JsEnvoyConfig, eventCallback: (
 export declare function startEnvoyJs(config: JsEnvoyConfig, eventCallback: (event: any) => void): JsEnvoyHandle
 /** Native SQLite database handle exposed to JavaScript. */
 export declare class JsNativeDatabase {
+  takeLastKvError(): string | null
   run(sql: string, params?: Array<JsBindParam> | undefined | null): Promise<ExecuteResult>
   query(sql: string, params?: Array<JsBindParam> | undefined | null): Promise<QueryResult>
   exec(sql: string): Promise<QueryResult>

rivetkit-typescript/packages/rivetkit-native/src/database.rs

@@ -34,6 +34,10 @@ impl EnvoyKv {
 
 #[async_trait]
 impl SqliteKv for EnvoyKv {
+	fn on_error(&self, actor_id: &str, error: &SqliteKvError) {
+		tracing::error!(%actor_id, %error, "native sqlite kv operation failed");
+	}
+
 	async fn on_open(&self, _actor_id: &str) -> Result<(), SqliteKvError> {
 		Ok(())
 	}
@@ -116,6 +120,14 @@ impl JsNativeDatabase {
 			.and_then(|guard| guard.as_ref().map(NativeDatabase::as_ptr))
 			.unwrap_or(ptr::null_mut())
 	}
+
+	fn take_last_kv_error_inner(&self) -> Option<String> {
+		self
+			.db
+			.lock()
+			.ok()
+			.and_then(|guard| guard.as_ref().and_then(NativeDatabase::take_last_kv_error))
+	}
 }
 
 #[napi(object)]
@@ -140,6 +152,11 @@ pub struct QueryResult {
 
 #[napi]
 impl JsNativeDatabase {
+	#[napi]
+	pub fn take_last_kv_error(&self) -> Option<String> {
+		self.take_last_kv_error_inner()
+	}
+
 	#[napi]
 	pub async fn run(
 		&self,

rivetkit-typescript/packages/rivetkit-native/wrapper.js

@@ -264,6 +264,20 @@ function mapRows(rows, columns) {
 	});
 }
 
+async function wrapNativeStorageError(nativeDb, error) {
+	const lastKvError =
+		typeof nativeDb.takeLastKvError === "function"
+			? await nativeDb.takeLastKvError()
+			: null;
+	if (!lastKvError) {
+		throw error;
+	}
+	throw new Error(
+		`Database query failed because the underlying storage is no longer available (${lastKvError}). This usually means the actor is stopping. Use c.abortSignal to cancel long-running work before the actor shuts down.`,
+		{ cause: error instanceof Error ? error : undefined },
+	);
+}
+
 async function openRawDatabaseFromEnvoy(handle, actorId) {
 	const nativeDb = await openDatabaseFromEnvoy(handle, actorId);
 	let closed = false;
@@ -288,15 +302,29 @@ async function openRawDatabaseFromEnvoy(handle, actorId) {
 					/\bRETURNING\b/i.test(query);
 
 				if (returnsRows) {
-					const result = await nativeDb.query(query, bindings);
+					let result;
+					try {
+						result = await nativeDb.query(query, bindings);
+					} catch (error) {
+						await wrapNativeStorageError(nativeDb, error);
+					}
 					return mapRows(result.rows, result.columns);
 				}
 
-				await nativeDb.run(query, bindings);
+				try {
+					await nativeDb.run(query, bindings);
+				} catch (error) {
+					await wrapNativeStorageError(nativeDb, error);
+				}
 				return [];
 			}
 
-			const result = await nativeDb.exec(query);
+			let result;
+			try {
+				result = await nativeDb.exec(query);
+			} catch (error) {
+				await wrapNativeStorageError(nativeDb, error);
+			}
 			return mapRows(result.rows, result.columns);
 		},
 		close: async () => {

rivetkit-typescript/packages/sqlite-native/src/sqlite_kv.rs

@@ -68,6 +68,10 @@ pub struct KvGetResult {
 /// at a higher level.
 #[async_trait]
 pub trait SqliteKv: Send + Sync {
+	/// Called when a KV operation fails inside a VFS callback before the
+	/// original error is collapsed into a generic SQLite IO error code.
+	fn on_error(&self, _actor_id: &str, _error: &SqliteKvError) {}
+
 	/// Called when an actor's database is opened.
 	async fn on_open(&self, _actor_id: &str) -> Result<(), SqliteKvError> {
 		Ok(())

rivetkit-typescript/packages/sqlite-native/src/vfs.rs

@@ -8,13 +8,13 @@ use std::ffi::{c_char, c_int, c_void, CStr, CString};
 use std::ptr;
 use std::slice;
 use std::sync::atomic::{AtomicU64, Ordering};
-use std::sync::{Arc, OnceLock};
+use std::sync::{Arc, Mutex, OnceLock};
 
 use libsqlite3_sys::*;
 use tokio::runtime::Handle;
 
 use crate::kv;
-use crate::sqlite_kv::{KvGetResult, SqliteKv};
+use crate::sqlite_kv::{KvGetResult, SqliteKv, SqliteKvError};
 
 // MARK: Panic Guard
 
@@ -158,12 +158,32 @@ struct VfsContext {
 	actor_id: String,
 	main_file_name: String,
 	read_cache_enabled: bool,
+	last_error: Mutex<Option<String>>,
 	rt_handle: Handle,
 	io_methods: Box<sqlite3_io_methods>,
 	vfs_metrics: Arc<VfsMetrics>,
 }
 
 impl VfsContext {
+	fn clear_last_error(&self) {
+		if let Ok(mut last_error) = self.last_error.lock() {
+			*last_error = None;
+		}
+	}
+
+	fn set_last_error(&self, message: String) {
+		if let Ok(mut last_error) = self.last_error.lock() {
+			*last_error = Some(message);
+		}
+	}
+
+	fn report_kv_error(&self, err: SqliteKvError) -> String {
+		let message = err.to_string();
+		self.set_last_error(message.clone());
+		self.kv.on_error(&self.actor_id, &err);
+		message
+	}
+
 	fn resolve_file_tag(&self, path: &str) -> Option<u8> {
 		if path == self.main_file_name {
 			return Some(kv::FILE_TAG_MAIN);
@@ -187,7 +207,10 @@ impl VfsContext {
 		let result = self
 			.rt_handle
 			.block_on(self.kv.batch_get(&self.actor_id, keys))
-			.map_err(|e| e.to_string());
+			.map_err(|err| self.report_kv_error(err));
+		if result.is_ok() {
+			self.clear_last_error();
+		}
 		let elapsed = start.elapsed();
 		tracing::debug!(
 			op = %format_args!("get({key_count}keys)"),
@@ -208,7 +231,10 @@ impl VfsContext {
 		let result = self
 			.rt_handle
 			.block_on(self.kv.batch_put(&self.actor_id, keys, values))
-			.map_err(|e| e.to_string());
+			.map_err(|err| self.report_kv_error(err));
+		if result.is_ok() {
+			self.clear_last_error();
+		}
 		let elapsed = start.elapsed();
 		tracing::debug!(
 			op = %format_args!("put({key_count}keys)"),
@@ -229,7 +255,10 @@ impl VfsContext {
 		let result = self
 			.rt_handle
 			.block_on(self.kv.batch_delete(&self.actor_id, keys))
-			.map_err(|e| e.to_string());
+			.map_err(|err| self.report_kv_error(err));
+		if result.is_ok() {
+			self.clear_last_error();
+		}
 		let elapsed = start.elapsed();
 		tracing::debug!(
 			op = %format_args!("del({key_count}keys)"),
@@ -249,7 +278,10 @@ impl VfsContext {
 		let result = self
 			.rt_handle
 			.block_on(self.kv.delete_range(&self.actor_id, start, end))
-			.map_err(|e| e.to_string());
+			.map_err(|err| self.report_kv_error(err));
+		if result.is_ok() {
+			self.clear_last_error();
+		}
 		let elapsed = start_time.elapsed();
 		tracing::debug!(
 			op = "delRange",
@@ -1180,11 +1212,29 @@ unsafe extern "C" fn kv_vfs_current_time(_p_vfs: *mut sqlite3_vfs, p_time_out: *
 }
 
 unsafe extern "C" fn kv_vfs_get_last_error(
-	_p_vfs: *mut sqlite3_vfs,
-	_n_byte: c_int,
-	_z_err_msg: *mut c_char,
+	p_vfs: *mut sqlite3_vfs,
+	n_byte: c_int,
+	z_err_msg: *mut c_char,
 ) -> c_int {
-	vfs_catch_unwind!(SQLITE_IOERR, SQLITE_OK)
+	vfs_catch_unwind!(SQLITE_IOERR, {
+		if n_byte <= 0 || z_err_msg.is_null() {
+			return 0;
+		}
+
+		let ctx = get_vfs_ctx(p_vfs);
+		let last_error = ctx.last_error.lock().ok().and_then(|guard| guard.clone());
+		let Some(message) = last_error else {
+			*z_err_msg = 0;
+			return 0;
+		};
+
+		let bytes = message.as_bytes();
+		let max_len = (n_byte as usize).saturating_sub(1);
+		let copy_len = bytes.len().min(max_len);
+		ptr::copy_nonoverlapping(bytes.as_ptr(), z_err_msg.cast::<u8>(), copy_len);
+		*z_err_msg.add(copy_len) = 0;
+		0
+	})
 }
 
 // MARK: KvVfs
@@ -1199,6 +1249,16 @@ unsafe impl Send for KvVfs {}
 unsafe impl Sync for KvVfs {}
 
 impl KvVfs {
+	fn take_last_kv_error(&self) -> Option<String> {
+		unsafe {
+			(*self.ctx_ptr)
+				.last_error
+				.lock()
+				.ok()
+				.and_then(|mut last_error| last_error.take())
+		}
+	}
+
 	pub fn register(
 		name: &str,
 		kv: Arc<dyn SqliteKv>,
@@ -1226,6 +1286,7 @@ impl KvVfs {
 			actor_id: actor_id.clone(),
 			main_file_name: actor_id,
 			read_cache_enabled: read_cache_enabled(),
+			last_error: Mutex::new(None),
 			rt_handle,
 			io_methods: Box::new(io_methods),
 			vfs_metrics,
@@ -1295,6 +1356,10 @@ impl NativeDatabase {
 	pub fn as_ptr(&self) -> *mut sqlite3 {
 		self.db
 	}
+
+	pub fn take_last_kv_error(&self) -> Option<String> {
+		self._vfs.take_last_kv_error()
+	}
 }
 
 impl Drop for NativeDatabase {
@@ -1307,6 +1372,18 @@ impl Drop for NativeDatabase {
 	}
 }
 
+fn sqlite_error_message(db: *mut sqlite3) -> String {
+	unsafe {
+		if db.is_null() {
+			"unknown sqlite error".to_string()
+		} else {
+			CStr::from_ptr(sqlite3_errmsg(db))
+				.to_string_lossy()
+				.into_owned()
+		}
+	}
+}
+
 pub fn open_database(vfs: KvVfs, file_name: &str) -> Result<NativeDatabase, String> {
 	let c_name = CString::new(file_name).map_err(|err| err.to_string())?;
 	let mut db: *mut sqlite3 = ptr::null_mut();
@@ -1320,12 +1397,13 @@ pub fn open_database(vfs: KvVfs, file_name: &str) -> Result<NativeDatabase, Stri
 		)
 	};
 	if rc != SQLITE_OK {
+		let message = sqlite_error_message(db);
 		if !db.is_null() {
 			unsafe {
 				sqlite3_close(db);
 			}
 		}
-		return Err(format!("sqlite3_open_v2 failed with code {rc}"));
+		return Err(format!("sqlite3_open_v2 failed with code {rc}: {message}"));
 	}
 
 	for pragma in &[
@@ -1340,10 +1418,11 @@ pub fn open_database(vfs: KvVfs, file_name: &str) -> Result<NativeDatabase, Stri
 		let rc =
 			unsafe { sqlite3_exec(db, c_sql.as_ptr(), None, ptr::null_mut(), ptr::null_mut()) };
 		if rc != SQLITE_OK {
+			let message = sqlite_error_message(db);
 			unsafe {
 				sqlite3_close(db);
 			}
-			return Err(format!("{pragma} failed with code {rc}"));
+			return Err(format!("{pragma} failed with code {rc}: {message}"));
 		}
 	}