fix(pegboard-envoy): allow sqlite requests across lifecycle states

Author: NathanFlurry

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

@@ -305,30 +305,3 @@ async fn close_actor_on_shutdown(
 		}
 	}
 }
-
-pub async fn assert_sqlite_actor_active(
-	conn: &Conn,
-	actor_id: &str,
-	sqlite_generation: u64,
-) -> Result<ActiveActor> {
-	// Stopping is accepted in addition to Running: the actor still owns its sqlite
-	// generation until actor_stopped runs, and may flush a final commit while draining.
-	let active = conn
-		.active_actors
-		.get_async(actor_id)
-		.await
-		.map(|entry| entry.get().clone())
-		.context("sqlite actor is not active on envoy connection")?;
-
-	let active_sqlite_generation = active
-		.sqlite_generation
-		.context("sqlite actor is still starting")?;
-	ensure!(
-		active_sqlite_generation == sqlite_generation,
-		"sqlite request generation {} did not match active generation {}",
-		sqlite_generation,
-		active_sqlite_generation
-	);
-
-	Ok(active)
-}

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

@@ -703,8 +703,6 @@ async fn handle_sqlite_get_pages(
 ) -> Result<protocol::SqliteGetPagesResponse> {
 	validate_sqlite_get_pages_request(&request)?;
 	validate_sqlite_actor(ctx, conn, &request.actor_id).await?;
-	actor_lifecycle::assert_sqlite_actor_active(conn, &request.actor_id, request.generation)
-		.await?;
 
 	match conn
 		.sqlite_engine
@@ -749,8 +747,6 @@ async fn handle_sqlite_commit(
 	let decode_request_start = Instant::now();
 	validate_sqlite_dirty_pages("sqlite commit", &request.dirty_pages)?;
 	validate_sqlite_actor(ctx, conn, &request.actor_id).await?;
-	actor_lifecycle::assert_sqlite_actor_active(conn, &request.actor_id, request.generation)
-		.await?;
 	let decode_request_duration = decode_request_start.elapsed();
 	conn.sqlite_engine.metrics().observe_commit_phase(
 		"fast",
@@ -817,8 +813,6 @@ async fn handle_sqlite_commit_stage(
 	request: protocol::SqliteCommitStageRequest,
 ) -> Result<protocol::SqliteCommitStageResponse> {
 	validate_sqlite_actor(ctx, conn, &request.actor_id).await?;
-	actor_lifecycle::assert_sqlite_actor_active(conn, &request.actor_id, request.generation)
-		.await?;
 
 	match conn
 		.sqlite_engine
@@ -856,8 +850,6 @@ async fn handle_sqlite_commit_stage_begin(
 	request: protocol::SqliteCommitStageBeginRequest,
 ) -> Result<protocol::SqliteCommitStageBeginResponse> {
 	validate_sqlite_actor(ctx, conn, &request.actor_id).await?;
-	actor_lifecycle::assert_sqlite_actor_active(conn, &request.actor_id, request.generation)
-		.await?;
 
 	match conn
 		.sqlite_engine
@@ -892,8 +884,6 @@ async fn handle_sqlite_commit_finalize(
 ) -> Result<protocol::SqliteCommitFinalizeResponse> {
 	let decode_request_start = Instant::now();
 	validate_sqlite_actor(ctx, conn, &request.actor_id).await?;
-	actor_lifecycle::assert_sqlite_actor_active(conn, &request.actor_id, request.generation)
-		.await?;
 	conn.sqlite_engine.metrics().observe_commit_phase(
 		"slow",
 		"decode_request",

engine/sdks/rust/envoy-client/src/actor.rs

@@ -199,7 +199,8 @@ async fn actor_inner(
 		.await;
 
 	if let Err(error) = start_result {
-		tracing::error!(?error, "actor start failed");
+		let error_chain = error.chain().map(ToString::to_string).collect::<Vec<_>>();
+		tracing::error!(?error, error_chain = ?error_chain, "actor start failed");
 		send_event(
 			&mut ctx,
 			protocol::Event::EventActorStateUpdate(protocol::EventActorStateUpdate {

rivetkit-rust/packages/rivetkit-core/src/actor/task.rs

@@ -1398,7 +1398,7 @@ impl ActorTask {
 		let clean_exit = match outcome {
 			Ok(Ok(())) => true,
 			Ok(Err(error)) => {
-				tracing::error!(?error, "actor run handler failed");
+				log_actor_error(&error, "actor run handler failed");
 				false
 			}
 			Err(error) => {
@@ -1529,7 +1529,7 @@ impl ActorTask {
 		match (&mut run_handle).await {
 			Ok(Ok(())) => {}
 			Ok(Err(error)) => {
-				tracing::error!(?error, "actor run handler failed during shutdown");
+				log_actor_error(&error, "actor run handler failed during shutdown");
 			}
 			Err(error) => {
 				if !error.is_cancelled() {
@@ -2129,6 +2129,18 @@ fn clone_shutdown_result(result: &Result<()>) -> Result<()> {
 	}
 }
 
+fn log_actor_error(error: &anyhow::Error, log_message: &'static str) {
+	let structured = rivet_error::RivetError::extract(error);
+	tracing::error!(
+		?error,
+		group = structured.group(),
+		code = structured.code(),
+		message = %structured.message(),
+		metadata = ?structured.metadata(),
+		"{log_message}"
+	);
+}
+
 fn result_outcome<T>(result: &Result<T>) -> &'static str {
 	match result {
 		Ok(_) => "ok",

rivetkit-rust/packages/rivetkit-sqlite/src/vfs.rs

@@ -1308,6 +1308,15 @@ impl VfsContext {
 		{
 			Ok(outcome) => outcome,
 			Err(err) => {
+				tracing::error!(
+					actor_id = %self.actor_id,
+					generation = request.generation,
+					expected_head_txid = request.expected_head_txid,
+					new_db_size_pages = request.new_db_size_pages,
+					dirty_pages = request.dirty_pages.len(),
+					?err,
+					"sqlite flush commit failed"
+				);
 				mark_dead_for_non_fence_commit_error(self, &err);
 				return Err(err);
 			}
@@ -1387,6 +1396,15 @@ impl VfsContext {
 		{
 			Ok(outcome) => outcome,
 			Err(err) => {
+				tracing::error!(
+					actor_id = %self.actor_id,
+					generation = request.generation,
+					expected_head_txid = request.expected_head_txid,
+					new_db_size_pages = request.new_db_size_pages,
+					dirty_pages = request.dirty_pages.len(),
+					?err,
+					"sqlite atomic commit failed"
+				);
 				mark_dead_for_non_fence_commit_error(self, &err);
 				return Err(err);
 			}
@@ -1460,13 +1478,22 @@ fn assert_batch_atomic_probe(db: *mut sqlite3, vfs: &SqliteVfs) -> std::result::
 	";
 
 	if let Err(err) = sqlite_exec(db, probe_sql) {
+		tracing::error!(
+			%err,
+			last_error = ?vfs.clone_last_error(),
+			commit_atomic_before,
+			"sqlite batch atomic probe failed"
+		);
 		cleanup_batch_atomic_probe(db);
 		return Err(format!("batch atomic probe failed: {err}"));
 	}
 
 	let commit_atomic_after = vfs.commit_atomic_count();
 	if commit_atomic_after == commit_atomic_before {
 		tracing::error!(
+			commit_atomic_before,
+			commit_atomic_after,
+			last_error = ?vfs.clone_last_error(),
 			"batch atomic writes not active for sqlite, SQLITE_ENABLE_BATCH_ATOMIC_WRITE may be missing"
 		);
 		cleanup_batch_atomic_probe(db);
@@ -2215,6 +2242,12 @@ unsafe extern "C" fn io_sync(p_file: *mut sqlite3_file, _flags: c_int) -> c_int
 		match ctx.flush_dirty_pages() {
 			Ok(_) => SQLITE_OK,
 			Err(err) => {
+				tracing::error!(
+					actor_id = %ctx.actor_id,
+					last_error = ?ctx.clone_last_error(),
+					?err,
+					"sqlite sync failed"
+				);
 				mark_dead_from_fence_commit_error(ctx, &err);
 				SQLITE_IOERR_FSYNC
 			}
@@ -2280,6 +2313,12 @@ unsafe extern "C" fn io_file_control(
 					SQLITE_OK
 				}
 				Err(err) => {
+					tracing::error!(
+						actor_id = %ctx.actor_id,
+						last_error = ?ctx.clone_last_error(),
+						?err,
+						"sqlite atomic write file control failed"
+					);
 					mark_dead_from_fence_commit_error(ctx, &err);
 					SQLITE_IOERR
 				}
@@ -2532,6 +2571,10 @@ impl SqliteVfs {
 		unsafe { (*self.ctx_ptr).take_last_error() }
 	}
 
+	fn clone_last_error(&self) -> Option<String> {
+		unsafe { (*self.ctx_ptr).clone_last_error() }
+	}
+
 	fn register_with_transport(
 		name: &str,
 		transport: SqliteTransport,
@@ -2659,6 +2702,13 @@ pub fn open_database(
 	};
 	if rc != SQLITE_OK {
 		let message = sqlite_error_message(db);
+		tracing::error!(
+			file_name,
+			rc,
+			%message,
+			last_error = ?vfs.clone_last_error(),
+			"failed to open sqlite database with custom VFS"
+		);
 		if !db.is_null() {
 			unsafe {
 				sqlite3_close(db);
@@ -2676,6 +2726,13 @@ pub fn open_database(
 		"PRAGMA locking_mode = EXCLUSIVE;",
 	] {
 		if let Err(err) = sqlite_exec(db, pragma) {
+			tracing::error!(
+				file_name,
+				pragma,
+				%err,
+				last_error = ?vfs.clone_last_error(),
+				"failed to configure sqlite database"
+			);
 			unsafe {
 				sqlite3_close(db);
 			}
@@ -2684,6 +2741,12 @@ pub fn open_database(
 	}
 
 	if let Err(err) = assert_batch_atomic_probe(db, &vfs) {
+		tracing::error!(
+			file_name,
+			%err,
+			last_error = ?vfs.clone_last_error(),
+			"failed to verify sqlite batch atomic writes"
+		);
 		unsafe {
 			sqlite3_close(db);
 		}

rivetkit-typescript/packages/rivetkit-napi/src/actor_factory.rs

@@ -976,9 +976,11 @@ fn parse_bridge_rivet_error(reason: &str) -> Option<anyhow::Error> {
 			return None;
 		}
 	};
-	tracing::debug!(
+	tracing::warn!(
 		group = %payload.group.as_str(),
 		code = %payload.code.as_str(),
+		message = %payload.message.as_str(),
+		metadata = ?payload.metadata,
 		has_metadata = payload.metadata.is_some(),
 		public_ = ?payload.public_,
 		status_code = ?payload.status_code,
@@ -1001,8 +1003,16 @@ fn parse_bridge_rivet_error(reason: &str) -> Option<anyhow::Error> {
 }
 
 pub(crate) fn callback_error(callback_name: &str, error: napi::Error) -> anyhow::Error {
+	let status = error.status;
 	let reason = error.reason;
 	if let Some(error) = parse_bridge_rivet_error(&reason) {
+		let error_chain = error.chain().map(ToString::to_string).collect::<Vec<_>>();
+		tracing::warn!(
+			callback = callback_name,
+			status = ?status,
+			error_chain = ?error_chain,
+			"napi callback failed with structured bridge error"
+		);
 		return error;
 	}
 	if error.status == napi::Status::Closing {

rivetkit-typescript/packages/rivetkit-napi/src/lib.rs

@@ -43,6 +43,7 @@ pub(crate) struct NapiInvalidState {
 }
 
 pub(crate) fn napi_anyhow_error(error: anyhow::Error) -> napi::Error {
+	let error_chain = error.chain().map(ToString::to_string).collect::<Vec<_>>();
 	let bridge_context = error
 		.chain()
 		.find_map(|cause| cause.downcast_ref::<crate::actor_factory::BridgeRivetErrorContext>());
@@ -57,9 +58,12 @@ pub(crate) fn napi_anyhow_error(error: anyhow::Error) -> napi::Error {
 		"public": public_,
 		"statusCode": status_code,
 	});
-	tracing::debug!(
+	tracing::error!(
 		group = error.group(),
 		code = error.code(),
+		message = %error.message(),
+		metadata = ?error.metadata(),
+		error_chain = ?error_chain,
 		has_metadata = error.metadata().is_some(),
 		?public_,
 		?status_code,

rivetkit-typescript/packages/rivetkit/src/registry/native.ts

@@ -575,6 +575,17 @@ function encodeNativeCallbackError(error: unknown): Error {
 				bridge: "native_callback",
 			});
 
+	logger().warn({
+		msg: "native callback error encoded for bridge",
+		group: structuredError.group,
+		code: structuredError.code,
+		message: structuredError.message,
+		metadata: structuredError.metadata,
+		originalError: stringifyError(error),
+		stack: error instanceof Error ? error.stack : undefined,
+		bridge: "native_callback",
+	});
+
 	const bridgeError = new Error(encodeBridgeRivetError(structuredError), {
 		cause: error instanceof Error ? error : undefined,
 	});