fix(rivetkit-core): preserve UserError message and metadata

rivetkit-rust/packages/rivetkit-core/src/error.rs

@@ -4,6 +4,7 @@ use serde_json::Value as JsonValue;
 
 pub fn public_error_status_code(group: &str, code: &str) -> Option<u16> {
 	match (group, code) {
+		("user", _) => Some(400),
 		("auth", "forbidden") => Some(403),
 		("actor", "action_not_found") => Some(404),
 		("actor", "action_timed_out") => Some(408),

rivetkit-rust/packages/rivetkit-core/tests/context.rs

@@ -327,6 +327,7 @@ mod moved_tests {
 			)),
 			protocol_metadata: Arc::new(tokio::sync::Mutex::new(None)),
 			shutting_down: std::sync::atomic::AtomicBool::new(false),
+			last_ping_ts: std::sync::atomic::AtomicI64::new(now_timestamp_ms()),
 			stopped_tx: tokio::sync::watch::channel(true).0,
 		});
 		shared
@@ -373,6 +374,7 @@ mod moved_tests {
 			)),
 			protocol_metadata: Arc::new(tokio::sync::Mutex::new(None)),
 			shutting_down: std::sync::atomic::AtomicBool::new(false),
+			last_ping_ts: std::sync::atomic::AtomicI64::new(now_timestamp_ms()),
 			stopped_tx: tokio::sync::watch::channel(true).0,
 		});
 		EnvoyHandle::from_shared(shared)

rivetkit-rust/packages/rivetkit-core/tests/modules/action_dispatch_error.rs

@@ -29,6 +29,29 @@ fn preserves_public_error_message_for_client_boundary() {
 	assert_eq!(error.client_message(), "action `missing` was not found");
 }
 
+#[test]
+fn preserves_user_error_message_and_metadata_for_client_boundary() {
+	let metadata = serde_json::json!({
+		"limit": 20,
+		"attempted": 25,
+	});
+	let error = ActionDispatchError::from_anyhow(anyhow::Error::new(RivetError {
+		kind: RivetErrorKind::Dynamic {
+			group: "user".to_owned(),
+			code: "quota_exceeded".to_owned(),
+			default_message: "quota exceeded".to_owned(),
+		},
+		meta: serde_json::value::to_raw_value(&metadata).ok(),
+		message: None,
+		actor: None,
+	}));
+
+	assert_eq!(error.group, "user");
+	assert_eq!(error.code, "quota_exceeded");
+	assert_eq!(error.client_message(), "quota exceeded");
+	assert_eq!(error.client_metadata(), Some(&metadata));
+}
+
 #[test]
 fn masks_private_structured_message_at_client_boundary() {
 	static TEST_ERROR: RivetErrorSchema = RivetErrorSchema {

rivetkit-rust/packages/rivetkit-core/tests/schedule.rs

@@ -96,6 +96,7 @@ mod moved_tests {
 			)),
 			protocol_metadata: Arc::new(tokio::sync::Mutex::new(None)),
 			shutting_down: AtomicBool::new(false),
+			last_ping_ts: std::sync::atomic::AtomicI64::new(now_timestamp_ms()),
 			stopped_tx: tokio::sync::watch::channel(true).0,
 		});
 

rivetkit-rust/packages/rivetkit-core/tests/sqlite.rs

@@ -1,7 +1,8 @@
 use std::collections::HashMap;
 use std::sync::Arc;
-use std::sync::atomic::AtomicBool;
 use std::sync::Mutex as StdMutex;
+use std::sync::atomic::AtomicBool;
+use std::time::{SystemTime, UNIX_EPOCH};
 
 use super::*;
 use depot_client_types::{HEAD_FENCE_MISMATCH_CODE, HEAD_FENCE_MISMATCH_GROUP};
@@ -34,6 +35,13 @@ struct SqliteOperationLog {
 	error_message: Option<String>,
 }
 
+fn now_timestamp_ms() -> i64 {
+	let duration = SystemTime::now()
+		.duration_since(UNIX_EPOCH)
+		.unwrap_or_default();
+	i64::try_from(duration.as_millis()).unwrap_or(i64::MAX)
+}
+
 #[derive(Clone)]
 struct SqliteOperationLogLayer {
 	records: Arc<StdMutex<Vec<SqliteOperationLog>>>,
@@ -173,6 +181,7 @@ fn test_envoy_handle() -> (EnvoyHandle, mpsc::UnboundedReceiver<ToEnvoyMessage>)
 		ws_tx: Arc::new(AsyncMutex::new(None::<mpsc::UnboundedSender<WsTxMessage>>)),
 		protocol_metadata: Arc::new(AsyncMutex::new(None)),
 		shutting_down: AtomicBool::new(false),
+		last_ping_ts: std::sync::atomic::AtomicI64::new(now_timestamp_ms()),
 		stopped_tx: tokio::sync::watch::channel(true).0,
 	});
 
@@ -318,7 +327,10 @@ async fn remote_execute_logs_operation_context_at_source() {
 	let result = db
 		.execute(
 			"SELECT ?",
-			Some(vec![BindParam::Integer(1), BindParam::Text("two".to_owned())]),
+			Some(vec![
+				BindParam::Integer(1),
+				BindParam::Text("two".to_owned()),
+			]),
 		)
 		.await;
 

rivetkit-rust/packages/rivetkit-core/tests/task.rs

@@ -6,7 +6,7 @@ mod moved_tests {
 	use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
 	use std::sync::{Mutex, OnceLock};
 	use std::task::Poll;
-	use std::time::{Duration, Instant};
+	use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
 
 	use futures::{FutureExt, poll};
 	use rivet_envoy_client::config::{
@@ -52,6 +52,13 @@ mod moved_tests {
 	use crate::{ActorConfig, ActorContext, ActorFactory};
 	use rivet_envoy_client::utils::EnvoyShutdownError;
 
+	fn now_timestamp_ms() -> i64 {
+		let duration = SystemTime::now()
+			.duration_since(UNIX_EPOCH)
+			.unwrap_or_default();
+		i64::try_from(duration.as_millis()).unwrap_or(i64::MAX)
+	}
+
 	fn test_hook_lock() -> &'static AsyncMutex<()> {
 		static LOCK: OnceLock<AsyncMutex<()>> = OnceLock::new();
 		LOCK.get_or_init(|| AsyncMutex::new(()))
@@ -254,6 +261,7 @@ mod moved_tests {
 			)),
 			protocol_metadata: Arc::new(tokio::sync::Mutex::new(None)),
 			shutting_down: AtomicBool::new(false),
+			last_ping_ts: std::sync::atomic::AtomicI64::new(now_timestamp_ms()),
 			stopped_tx: tokio::sync::watch::channel(true).0,
 		});