From 4c4a58c50836d9342f8b51f673626f6429736012 Mon Sep 17 00:00:00 2001 From: ABCxFF <79597906+abcxff@users.noreply.github.com> Date: Tue, 14 Jul 2026 04:58:17 +0000 Subject: [PATCH 1/6] feat(rivetkit): report actor run() errors to the engine as errored stops --- engine/sdks/rust/envoy-client/src/handle.rs | 12 - .../packages/rivetkit-core/CLAUDE.md | 2 + .../rivetkit-core/src/actor/context.rs | 37 +++ .../packages/rivetkit-core/src/actor/sleep.rs | 7 +- .../packages/rivetkit-core/src/actor/task.rs | 39 ++- .../packages/rivetkit-core/tests/sleep.rs | 210 ++++++++++++++++ .../packages/rivetkit-core/tests/task.rs | 136 +++++++++++ .../packages/rivetkit/src/context.rs | 9 + rivetkit-rust/packages/rivetkit/src/start.rs | 229 +++++++++++++++++- 9 files changed, 661 insertions(+), 20 deletions(-) diff --git a/engine/sdks/rust/envoy-client/src/handle.rs b/engine/sdks/rust/envoy-client/src/handle.rs index 5285c12412..0df27d4de3 100644 --- a/engine/sdks/rust/envoy-client/src/handle.rs +++ b/engine/sdks/rust/envoy-client/src/handle.rs @@ -161,18 +161,6 @@ impl EnvoyHandle { ); } - pub fn destroy_actor(&self, actor_id: String, generation: Option) { - let _ = crate::envoy::send_to_envoy_tx( - &self.shared, - ToEnvoyMessage::ActorIntent { - actor_id, - generation, - intent: protocol::ActorIntent::ActorIntentStop, - error: None, - }, - ); - } - pub async fn get_actor(&self, actor_id: &str, generation: Option) -> Option { let (tx, rx) = tokio::sync::oneshot::channel(); crate::envoy::send_to_envoy_tx( diff --git a/rivetkit-rust/packages/rivetkit-core/CLAUDE.md b/rivetkit-rust/packages/rivetkit-core/CLAUDE.md index 7c072d2b98..92a50ffc76 100644 --- a/rivetkit-rust/packages/rivetkit-core/CLAUDE.md +++ b/rivetkit-rust/packages/rivetkit-core/CLAUDE.md @@ -11,6 +11,8 @@ - `ActorContext::set_prevent_sleep(...)` / `prevent_sleep()` are deprecated no-ops kept for NAPI bridge compatibility. Use `keep_awake(future)` (holds counter while awaited) or `wait_until(future)` (tracked shutdown task) instead. Do not reintroduce a `prevent_sleep` field, a `CanSleep::PreventSleep` variant, or branches that read it. - Runtime-owned promises that must drain during shutdown should use `ActorContext::register_task(...)`, not public `wait_until(...)`, so metrics and runtime intent stay distinct. Registered tasks must race user work against `shutdown_deadline_token()` so shutdown cannot hang forever. - `ctx.sleep()` and `ctx.destroy()` return `Result<()>`. They error with `ActorLifecycleError::Starting` when called before startup completes and `ActorLifecycleError::Stopping` if the requested flag has already been set this generation (atomic `swap(true, ...)`). Internal idle-timer paths log and suppress the already-requested error. +- `ctx.stop_with_error(message)` shares `destroy()`'s request flow but attaches a (length-capped) error that the envoy reports as `StopCode::Error`, so the engine records a crash and applies its crash handling instead of unconditionally destroying. +- A failed `run` handler while `Started` requests `stop_with_error` and keeps the generation alive awaiting the engine `Stop`; transitioning to `Terminated` there would drop the lifecycle inbox before the answering stop command arrives. - The grace deadline path (`on_sleep_grace_deadline`) aborts the user `run` handle and cancels `shutdown_deadline_token()`. Foreign-runtime adapters running `onSleep` / `onDestroy` must observe that token via `tokio::select!` so SQLite teardown does not race user cleanup work. - Counter `register_zero_notify(&idle_notify)` hooks only drive shutdown drain waits. They are not a substitute for the activity-dirty notification, so any new sleep-affecting counter must also notify on transitions that change `can_sleep`. - A clean `run` exit while `Started` is not terminal. Keep the generation alive until the guaranteed `Stop` drives `SleepGrace` or `DestroyGrace`, and only treat `Terminated` as "grace hooks already completed." diff --git a/rivetkit-rust/packages/rivetkit-core/src/actor/context.rs b/rivetkit-rust/packages/rivetkit-core/src/actor/context.rs index 89e245de11..78ab767b13 100644 --- a/rivetkit-rust/packages/rivetkit-core/src/actor/context.rs +++ b/rivetkit-rust/packages/rivetkit-core/src/actor/context.rs @@ -464,6 +464,19 @@ impl ActorContext { } pub fn destroy(&self) -> Result<()> { + self.request_stop(None) + } + + /// Request a stop with an error attached. Behaves like [`Self::destroy`] + /// locally (destroy grace hooks run for this generation), but the envoy + /// reports the stop to the engine with `StopCode::Error` and the message, + /// so the engine records the crash and applies its crash handling instead + /// of unconditionally destroying the actor. + pub fn stop_with_error(&self, message: impl Into) -> Result<()> { + self.request_stop(Some(truncate_stop_error_message(message.into()))) + } + + fn request_stop(&self, error: Option) -> Result<()> { // See `sleep` for why the request flags disambiguate `started=false`. // destroy() is allowed after sleep() has been requested because // destroy is a stronger signal that escalates an in-flight sleep. @@ -478,6 +491,12 @@ impl ActorContext { return Err(ActorLifecycleError::Stopping.build()) .context("destroy already requested for this generation"); } + // Winning the swap above makes this the only writer of the error slot + // for this generation. The slot is consumed by + // `request_destroy_from_envoy` when the stop intent is sent. + if error.is_some() { + *self.0.sleep.destroy_error.lock() = error; + } // Reuse the shared teardown sequence used by the registry shutdown path // so future changes to `mark_destroy_requested` cannot drift. // `destroy_requested` is already true from the swap above. The redundant @@ -1569,6 +1588,24 @@ impl ActorContext { } } +/// Cap on the stop error message forwarded to the engine. The message is +/// persisted in engine workflow state and rendered in the dashboard, so +/// unbounded anyhow chains must be truncated at this boundary. +const MAX_STOP_ERROR_MESSAGE_LEN: usize = 2048; + +fn truncate_stop_error_message(mut message: String) -> String { + if message.len() <= MAX_STOP_ERROR_MESSAGE_LEN { + return message; + } + let mut end = MAX_STOP_ERROR_MESSAGE_LEN; + while !message.is_char_boundary(end) { + end -= 1; + } + message.truncate(end); + message.push_str("... (truncated)"); + message +} + fn now_timestamp_ms() -> i64 { let duration = SystemTime::now() .duration_since(UNIX_EPOCH) diff --git a/rivetkit-rust/packages/rivetkit-core/src/actor/sleep.rs b/rivetkit-rust/packages/rivetkit-core/src/actor/sleep.rs index e7841fcce3..332a0c3a7f 100644 --- a/rivetkit-rust/packages/rivetkit-core/src/actor/sleep.rs +++ b/rivetkit-rust/packages/rivetkit-core/src/actor/sleep.rs @@ -47,6 +47,9 @@ pub(crate) struct SleepState { pub(super) envoy_handle: Mutex>, pub(super) generation: Mutex>, pub(super) http_request_counter: Mutex>>, + // Forced-sync: written once by whichever caller wins the destroy-request + // swap, then consumed when the stop intent is sent to the envoy. + pub(super) destroy_error: Mutex>, #[cfg(test)] sleep_request_count: TestAtomicUsize, #[cfg(test)] @@ -79,6 +82,7 @@ impl SleepState { envoy_handle: Mutex::new(None), generation: Mutex::new(None), http_request_counter: Mutex::new(None), + destroy_error: Mutex::new(None), #[cfg(test)] sleep_request_count: TestAtomicUsize::new(0), #[cfg(test)] @@ -171,8 +175,9 @@ impl ActorContext { .fetch_add(1, Ordering::SeqCst); let envoy_handle = self.0.sleep.envoy_handle.lock().clone(); let generation = *self.0.sleep.generation.lock(); + let error = self.0.sleep.destroy_error.lock().take(); if let Some(envoy_handle) = envoy_handle { - envoy_handle.destroy_actor(self.actor_id().to_owned(), generation); + envoy_handle.stop_actor(self.actor_id().to_owned(), generation, error); } } diff --git a/rivetkit-rust/packages/rivetkit-core/src/actor/task.rs b/rivetkit-rust/packages/rivetkit-core/src/actor/task.rs index c87602edc5..1dfacdabf1 100644 --- a/rivetkit-rust/packages/rivetkit-core/src/actor/task.rs +++ b/rivetkit-rust/packages/rivetkit-core/src/actor/task.rs @@ -1455,15 +1455,19 @@ impl ActorTask { outcome: std::result::Result, JoinError>, ) -> Option { self.run_handle = None; - let clean_exit = match outcome { - Ok(Ok(())) => true, + let (clean_exit, crash_message) = match outcome { + Ok(Ok(())) => (true, None), Ok(Err(error)) => { log_actor_error(&error, "actor run handler failed"); - false + (false, Some(format!("{error:#}"))) } Err(error) => { tracing::error!(?error, "actor run handler join failed"); - false + // Deliberate cancellations are not crashes and must not be + // reported to the engine as one. + let message = + (!error.is_cancelled()).then(|| "actor run handler panicked".to_string()); + (false, message) } }; @@ -1476,6 +1480,33 @@ impl ActorTask { } if self.lifecycle == LifecycleState::Started { + // A failed run handler while `Started` must reach the engine as an + // errored stop instead of dying silently until the lost timeout. + // Keep the generation alive so the engine's answering `Stop` + // command still drives the destroy grace hooks; transitioning to + // `Terminated` here would drop the lifecycle inbox before that + // command arrives. + if let Some(message) = crash_message { + match self.ctx.stop_with_error(message.clone()) { + Ok(()) => return None, + Err(error) => { + if self.ctx.destroy_requested() { + tracing::debug!( + ?error, + actor_id = %self.ctx.actor_id(), + "run handler failed while a stop was already requested" + ); + return None; + } + tracing::error!( + ?error, + dropped_message = %message, + actor_id = %self.ctx.actor_id(), + "failed to report run handler error to engine" + ); + } + } + } self.transition_to(LifecycleState::Terminated); } diff --git a/rivetkit-rust/packages/rivetkit-core/tests/sleep.rs b/rivetkit-rust/packages/rivetkit-core/tests/sleep.rs index 6625c3870d..72c7c56c3d 100644 --- a/rivetkit-rust/packages/rivetkit-core/tests/sleep.rs +++ b/rivetkit-rust/packages/rivetkit-core/tests/sleep.rs @@ -527,6 +527,216 @@ mod moved_tests { assert_eq!(rivet_err.code(), "stopping"); } + #[tokio::test(start_paused = true)] + async fn stop_with_error_before_started_errors_with_actor_starting() { + let ctx = ActorContext::new_for_sleep_tests("actor-stop-error-before-started"); + + let err = ctx + .stop_with_error("boom") + .expect_err("stop_with_error should fail before started is set"); + let rivet_err = rivet_error::RivetError::extract(&err); + assert_eq!(rivet_err.group(), "actor"); + assert_eq!(rivet_err.code(), "starting"); + } + + #[tokio::test(start_paused = true)] + async fn stop_with_error_after_destroy_errors_with_actor_stopping() { + let ctx = ActorContext::new_for_sleep_tests("actor-stop-error-after-destroy"); + ctx.set_started(true); + + ctx.destroy() + .expect("destroy call should be accepted after startup"); + + let err = ctx + .stop_with_error("boom") + .expect_err("stop_with_error should fail once destroy is already requested"); + let rivet_err = rivet_error::RivetError::extract(&err); + assert_eq!(rivet_err.group(), "actor"); + assert_eq!(rivet_err.code(), "stopping"); + } + + mod stop_intent { + use std::collections::HashMap; + use std::sync::Mutex as EnvoySharedMutex; + use std::sync::atomic::AtomicBool; + use std::time::Duration; + + use rivet_envoy_client::config::{ + BoxFuture, EnvoyCallbacks, EnvoyConfig, HttpRequest, HttpResponse, WebSocketHandler, + WebSocketSender, + }; + use rivet_envoy_client::context::{SharedContext, WsTxMessage}; + use rivet_envoy_client::envoy::ToEnvoyMessage; + use rivet_envoy_client::handle::EnvoyHandle; + use rivet_envoy_client::protocol; + use tokio::sync::mpsc; + + use std::sync::Arc; + + use crate::actor::context::ActorContext; + + struct IdleEnvoyCallbacks; + + impl EnvoyCallbacks for IdleEnvoyCallbacks { + fn on_actor_start( + &self, + _handle: EnvoyHandle, + _actor_id: String, + _generation: u32, + _config: protocol::ActorConfig, + _preloaded_kv: Option, + ) -> BoxFuture> { + Box::pin(async { Ok(()) }) + } + + fn on_shutdown(&self) {} + + fn fetch( + &self, + _handle: EnvoyHandle, + _actor_id: String, + _gateway_id: protocol::GatewayId, + _request_id: protocol::RequestId, + _request: HttpRequest, + ) -> BoxFuture> { + Box::pin(async { anyhow::bail!("fetch should not run in sleep tests") }) + } + + fn websocket( + &self, + _handle: EnvoyHandle, + _actor_id: String, + _gateway_id: protocol::GatewayId, + _request_id: protocol::RequestId, + _request: HttpRequest, + _path: String, + _headers: HashMap, + _is_hibernatable: bool, + _is_restoring_hibernatable: bool, + _sender: WebSocketSender, + ) -> BoxFuture> { + Box::pin(async { anyhow::bail!("websocket should not run in sleep tests") }) + } + + fn can_hibernate( + &self, + _actor_id: &str, + _gateway_id: &protocol::GatewayId, + _request_id: &protocol::RequestId, + _request: &HttpRequest, + ) -> BoxFuture> { + Box::pin(async { Ok(false) }) + } + } + + fn test_envoy_handle() -> (EnvoyHandle, mpsc::UnboundedReceiver) { + let (envoy_tx, envoy_rx) = mpsc::unbounded_channel(); + let shared = Arc::new(SharedContext { + config: EnvoyConfig { + version: 1, + endpoint: "http://127.0.0.1:1".to_string(), + token: None, + namespace: "test".to_string(), + pool_name: "test".to_string(), + prepopulate_actor_names: HashMap::new(), + metadata: None, + not_global: true, + debug_latency_ms: None, + callbacks: Arc::new(IdleEnvoyCallbacks), + }, + envoy_key: "test-envoy".to_string(), + envoy_tx, + actors: Arc::new(EnvoySharedMutex::new(HashMap::new())), + actors_notify: Arc::new(tokio::sync::Notify::new()), + live_tunnel_requests: Arc::new(EnvoySharedMutex::new(HashMap::new())), + pending_hibernation_restores: Arc::new(EnvoySharedMutex::new(HashMap::new())), + ws_tx: Arc::new(tokio::sync::Mutex::new( + None::>, + )), + protocol_metadata: Arc::new(tokio::sync::Mutex::new(None)), + shutting_down: AtomicBool::new(false), + last_ping_ts: std::sync::atomic::AtomicI64::new(i64::MAX), + stopped_tx: tokio::sync::watch::channel(true).0, + }); + + (EnvoyHandle::from_shared(shared), envoy_rx) + } + + async fn recv_stop_intent( + rx: &mut mpsc::UnboundedReceiver, + expected_actor_id: &str, + ) -> Option { + let message = tokio::time::timeout(Duration::from_secs(5), rx.recv()) + .await + .expect("timed out waiting for stop intent") + .expect("envoy channel closed before stop intent"); + match message { + ToEnvoyMessage::ActorIntent { + actor_id, + generation, + intent: protocol::ActorIntent::ActorIntentStop, + error, + } => { + assert_eq!(actor_id, expected_actor_id); + assert_eq!(generation, Some(3)); + error + } + _ => panic!("expected stop intent envoy message"), + } + } + + #[tokio::test(start_paused = true)] + async fn destroy_sends_stop_intent_without_error() { + let ctx = ActorContext::new_for_sleep_tests("actor-destroy-intent"); + let (handle, mut rx) = test_envoy_handle(); + ctx.configure_sleep_envoy(handle, Some(3)); + ctx.set_started(true); + + ctx.destroy().expect("destroy should succeed after startup"); + + let error = recv_stop_intent(&mut rx, "actor-destroy-intent").await; + assert_eq!(error, None); + } + + #[tokio::test(start_paused = true)] + async fn stop_with_error_sends_stop_intent_with_message() { + let ctx = ActorContext::new_for_sleep_tests("actor-stop-error-intent"); + let (handle, mut rx) = test_envoy_handle(); + ctx.configure_sleep_envoy(handle, Some(3)); + ctx.set_started(true); + + ctx.stop_with_error("child exited unexpectedly (exit status: 137)") + .expect("stop_with_error should succeed after startup"); + + let error = recv_stop_intent(&mut rx, "actor-stop-error-intent").await; + assert_eq!( + error.as_deref(), + Some("child exited unexpectedly (exit status: 137)") + ); + } + + #[tokio::test(start_paused = true)] + async fn stop_with_error_truncates_long_message() { + let ctx = ActorContext::new_for_sleep_tests("actor-stop-error-truncated"); + let (handle, mut rx) = test_envoy_handle(); + ctx.configure_sleep_envoy(handle, Some(3)); + ctx.set_started(true); + + ctx.stop_with_error("x".repeat(1024 * 1024)) + .expect("stop_with_error should succeed after startup"); + + let error = recv_stop_intent(&mut rx, "actor-stop-error-truncated") + .await + .expect("stop intent should carry the truncated message"); + assert!( + error.len() < 4096, + "message must be capped: {}", + error.len() + ); + assert!(error.ends_with("... (truncated)")); + } + } + // `set_prevent_sleep` is a deprecated no-op kept for NAPI bridge // compatibility. The exhaustive `CanSleep` match below is a build-time // guard against reintroducing a `PreventSleep` enum variant. diff --git a/rivetkit-rust/packages/rivetkit-core/tests/task.rs b/rivetkit-rust/packages/rivetkit-core/tests/task.rs index 2a48d31128..298d1cc287 100644 --- a/rivetkit-rust/packages/rivetkit-core/tests/task.rs +++ b/rivetkit-rust/packages/rivetkit-core/tests/task.rs @@ -381,6 +381,55 @@ mod moved_tests { })) } + fn detached_cleanup_after_failed_run_factory( + destroy_count: Arc, + run_returned_tx: oneshot::Sender<()>, + cleanup_tx: oneshot::Sender, + ) -> Arc { + let run_returned_tx = Arc::new(Mutex::new(Some(run_returned_tx))); + let cleanup_tx = Arc::new(Mutex::new(Some(cleanup_tx))); + Arc::new(ActorFactory::new(ActorConfig::default(), move |start| { + let destroy_count = destroy_count.clone(); + let run_returned_tx = run_returned_tx.clone(); + let cleanup_tx = cleanup_tx.clone(); + Box::pin(async move { + let mut events = start.events; + tokio::spawn(async move { + while let Some(event) = events.recv().await { + match event { + ActorEvent::SerializeState { reply, .. } => { + reply.send(Ok(Vec::new())); + } + ActorEvent::RunGracefulCleanup { reason, reply } => { + if matches!(reason, ShutdownKind::Destroy) { + destroy_count.fetch_add(1, Ordering::SeqCst); + } + reply.send(Ok(())); + if let Some(tx) = cleanup_tx + .lock() + .expect("cleanup sender lock poisoned") + .take() + { + let _ = tx.send(reason); + } + break; + } + _ => {} + } + } + }); + if let Some(tx) = run_returned_tx + .lock() + .expect("run returned sender lock poisoned") + .take() + { + let _ = tx.send(()); + } + anyhow::bail!("run handler failed under test") + }) + })) + } + fn sleep_grace_factory( config: ActorConfig, begin_sleep_count: Arc, @@ -3490,6 +3539,93 @@ mod moved_tests { .expect("stop should succeed"); } + #[tokio::test] + async fn failed_run_exit_requests_errored_stop_and_still_dispatches_cleanup() { + let ctx = new_with_kv( + "actor-failed-run-stop", + "task-failed-run", + Vec::new(), + "local", + new_in_memory(), + ); + let destroy_count = Arc::new(AtomicUsize::new(0)); + let (run_returned_tx, run_returned_rx) = oneshot::channel(); + let (cleanup_tx, cleanup_rx) = oneshot::channel(); + let mut task = new_task_with_factory( + ctx.clone(), + detached_cleanup_after_failed_run_factory( + destroy_count.clone(), + run_returned_tx, + cleanup_tx, + ), + ); + + let (start_tx, start_rx) = oneshot::channel(); + task.handle_lifecycle(LifecycleCommand::Start { reply: start_tx }) + .await; + start_rx + .await + .expect("start reply should send") + .expect("start should succeed"); + timeout(Duration::from_secs(2), run_returned_rx) + .await + .expect("failed run should return") + .expect("run returned signal should send"); + + let outcome = ActorTask::wait_for_run_handle(task.run_handle.as_mut()).await; + assert!(task.handle_run_handle_outcome(outcome).is_none()); + // The failed run must not terminate the generation locally: the + // errored stop request goes to the engine and the answering Stop + // command still drives the destroy grace hooks. + assert_eq!(task.lifecycle, LifecycleState::Started); + assert!( + ctx.is_destroy_requested(), + "failed run should request an errored stop" + ); + + let (stop_tx, stop_rx) = oneshot::channel(); + task.handle_lifecycle(LifecycleCommand::Stop { + reason: ShutdownKind::Destroy, + reply: stop_tx, + }) + .await; + assert_eq!( + timeout(Duration::from_secs(2), cleanup_rx) + .await + .expect("grace cleanup should run after Stop") + .expect("cleanup signal should send"), + ShutdownKind::Destroy + ); + assert_eq!(destroy_count.load(Ordering::SeqCst), 1); + + timeout(Duration::from_secs(2), async { + while ctx.core_dispatched_hook_count() != 0 { + yield_now().await; + } + }) + .await + .expect("core-dispatched cleanup hook should complete"); + + let exit = task + .try_finish_grace() + .expect("grace should finish after cleanup hook completes"); + let LiveExit::Shutdown { + reason: shutdown_reason, + } = exit + else { + panic!("grace should transition to shutdown"); + }; + assert_eq!(shutdown_reason, ShutdownKind::Destroy); + let result = task.run_shutdown(shutdown_reason).await; + task.deliver_shutdown_reply(shutdown_reason, &result); + task.transition_to(LifecycleState::Terminated); + result.expect("shutdown should succeed"); + stop_rx + .await + .expect("stop reply should send") + .expect("stop should succeed"); + } + #[tokio::test] async fn self_initiated_sleep_waits_for_stop_reply_before_shutdown_finishes() { let _hook_lock = test_hook_lock().lock().await; diff --git a/rivetkit-rust/packages/rivetkit/src/context.rs b/rivetkit-rust/packages/rivetkit/src/context.rs index 8775ad2c98..9d426fe070 100644 --- a/rivetkit-rust/packages/rivetkit/src/context.rs +++ b/rivetkit-rust/packages/rivetkit/src/context.rs @@ -333,6 +333,15 @@ impl Ctx { self.inner.destroy() } + /// Requests a stop with an error attached. Behaves like `destroy` locally, + /// but the engine records the message as a crash (`StopCode::Error`) and + /// applies its crash handling instead of unconditionally destroying the + /// actor. Returning an `Err` from `Actor::run` reports the error this way + /// automatically. + pub fn stop_with_error(&self, message: impl Into) -> Result<()> { + self.inner.stop_with_error(message) + } + #[deprecated(note = "no-op: use `keep_awake` or `wait_until` instead")] pub fn set_prevent_sleep(&self, enabled: bool) { #[allow(deprecated)] diff --git a/rivetkit-rust/packages/rivetkit/src/start.rs b/rivetkit-rust/packages/rivetkit/src/start.rs index 964a48ad6f..5807af193c 100644 --- a/rivetkit-rust/packages/rivetkit/src/start.rs +++ b/rivetkit-rust/packages/rivetkit/src/start.rs @@ -1,10 +1,12 @@ use std::any::{Any, TypeId}; use std::io::Cursor; use std::marker::PhantomData; +use std::panic::AssertUnwindSafe; use std::sync::Arc; use std::time::Duration; use anyhow::{Context, Result}; +use futures::FutureExt; use rivetkit_core::actor::ShutdownKind; use rivetkit_core::error::{ActorLifecycle, ActorRuntime}; use rivetkit_core::{ActorEvent, ActorEvents, ActorStart, QueueSendResult, QueueSendStatus, Reply}; @@ -226,13 +228,45 @@ fn spawn_run_task( cancel: CancellationToken, ) -> JoinHandle> { tokio::spawn(async move { - tokio::select! { - _ = cancel.cancelled() => Ok(()), - result = actor.run(ctx) => result, + let result = tokio::select! { + _ = cancel.cancelled() => return Ok(()), + result = AssertUnwindSafe(actor.run(ctx.clone())).catch_unwind() => result, + }; + let result = result.unwrap_or_else(|_| Err(anyhow::anyhow!("actor run handler panicked"))); + if let Err(error) = &result { + report_run_error(&ctx, error); } + result }) } +/// Reports a `run` failure to the engine as an errored stop. The event loop in +/// `run_actor` does not observe the run task until shutdown, so the report +/// must happen here for the engine to learn about the crash promptly. Errors +/// that surface after the abort signal fires are shutdown-induced (cancelled +/// waits during sleep or destroy) and are not crashes. +fn report_run_error(ctx: &Ctx, error: &anyhow::Error) { + if ctx.abort_signal().is_cancelled() { + tracing::debug!(?error, "actor run task failed during shutdown"); + return; + } + if let Err(stop_error) = ctx.stop_with_error(format!("{error:#}")) { + if ctx.inner().is_destroy_requested() { + tracing::debug!( + ?stop_error, + run_error = ?error, + "actor run failed while a stop was already requested" + ); + } else { + tracing::error!( + ?stop_error, + run_error = ?error, + "failed to report actor run error as errored stop" + ); + } + } +} + async fn stop_run_task( ctx: &Ctx, cancel: CancellationToken, @@ -1070,6 +1104,85 @@ mod tests { assert!(ctx.state().log.iter().any(|entry| entry == "on_destroy")); } + #[tokio::test] + async fn run_actor_error_requests_errored_stop() { + let (tx, rx) = unbounded_channel(); + let (start, ctx) = unit_start_with_ctx::(rx.into()); + // The harness has no ActorTask to complete startup, so mark the + // lifecycle started directly; stop requests are rejected before then. + ctx.inner().set_started(true); + let actor = tokio::spawn(run_actor::(start)); + + // The errored stop is requested inside the spawned run task right + // after `run` returns, so yielding the current-thread runtime drives + // it to completion deterministically. + for _ in 0..1000 { + if ctx.inner().is_destroy_requested() { + break; + } + tokio::task::yield_now().await; + } + assert!( + ctx.inner().is_destroy_requested(), + "run error should request an errored stop" + ); + + drop(tx); + let error = actor + .await + .expect("join run_actor") + .expect_err("run_actor should propagate the run error"); + assert!(format!("{error:#}").contains("boom in run")); + } + + #[tokio::test] + async fn run_actor_panic_requests_errored_stop() { + let (tx, rx) = unbounded_channel(); + let (start, ctx) = unit_start_with_ctx::(rx.into()); + ctx.inner().set_started(true); + let actor = tokio::spawn(run_actor::(start)); + + // See run_actor_error_requests_errored_stop for why yielding drives + // the spawned run task deterministically. + for _ in 0..1000 { + if ctx.inner().is_destroy_requested() { + break; + } + tokio::task::yield_now().await; + } + assert!( + ctx.inner().is_destroy_requested(), + "run panic should request an errored stop" + ); + + drop(tx); + let error = actor + .await + .expect("join run_actor") + .expect_err("run_actor should propagate the panic as an error"); + assert!(format!("{error:#}").contains("panicked")); + } + + #[tokio::test] + async fn run_actor_shutdown_error_is_not_reported_as_crash() { + let (tx, rx) = unbounded_channel(); + let (start, ctx) = unit_start_with_ctx::(rx.into()); + ctx.inner().set_started(true); + let actor = tokio::spawn(run_actor::(start)); + + // Closing the events channel starts the shutdown join path, which + // cancels the abort signal before joining the run task. + drop(tx); + actor + .await + .expect("join run_actor") + .expect_err("run error should still propagate during shutdown"); + assert!( + !ctx.inner().is_destroy_requested(), + "shutdown-induced run errors must not be reported as crashes" + ); + } + #[tokio::test] async fn run_actor_dispatches_typed_actions_by_arg_shape() { let (tx, rx) = unbounded_channel(); @@ -1614,6 +1727,116 @@ mod tests { } } + struct FailingRunActor; + + #[async_trait] + impl Actor for FailingRunActor { + type State = (); + type Input = (); + type Actions = (); + type Events = (); + type Queue = (); + type ConnParams = (); + type ConnState = (); + type Action = action::Raw; + + async fn create_state(_ctx: &Ctx, (): Self::Input) -> Result { + Ok(()) + } + + async fn create(_ctx: &Ctx) -> Result { + Ok(Self) + } + + async fn run(self: Arc, _ctx: Ctx) -> Result<()> { + anyhow::bail!("boom in run") + } + } + + struct PanickingRunActor; + + #[async_trait] + impl Actor for PanickingRunActor { + type State = (); + type Input = (); + type Actions = (); + type Events = (); + type Queue = (); + type ConnParams = (); + type ConnState = (); + type Action = action::Raw; + + async fn create_state(_ctx: &Ctx, (): Self::Input) -> Result { + Ok(()) + } + + async fn create(_ctx: &Ctx) -> Result { + Ok(Self) + } + + async fn run(self: Arc, _ctx: Ctx) -> Result<()> { + panic!("run task panic under test"); + } + } + + struct ShutdownErrRunActor; + + #[async_trait] + impl Actor for ShutdownErrRunActor { + type State = (); + type Input = (); + type Actions = (); + type Events = (); + type Queue = (); + type ConnParams = (); + type ConnState = (); + type Action = action::Raw; + + async fn create_state(_ctx: &Ctx, (): Self::Input) -> Result { + Ok(()) + } + + async fn create(_ctx: &Ctx) -> Result { + Ok(Self) + } + + async fn run(self: Arc, ctx: Ctx) -> Result<()> { + ctx.abort_signal().cancelled().await; + anyhow::bail!("wait cancelled by shutdown") + } + } + + fn unit_start_with_ctx(rx: ActorEvents) -> (Start, Ctx) { + let ctx = Ctx::new(rivetkit_core::ActorContext::new( + "actor-id", + "test", + Vec::new(), + "local", + )); + + let start = Start { + ctx: ctx.clone(), + input: Input { + bytes: None, + _p: PhantomData, + }, + is_new: true, + snapshot: Snapshot { + is_new: true, + bytes: None, + }, + hibernated: Vec::new(), + events: Events { + ctx: ctx.clone(), + rx, + _p: PhantomData, + }, + startup_ready: None, + }; + + (start, ctx) + } + fn cbor(value: &T) -> Vec { let mut encoded = Vec::new(); ciborium::into_writer(value, &mut encoded).expect("encode test value as cbor"); From ba2f32f9ee4333c6d386d8b85a54467308d626fd Mon Sep 17 00:00:00 2001 From: ABCxFF <79597906+abcxff@users.noreply.github.com> Date: Tue, 14 Jul 2026 04:58:17 +0000 Subject: [PATCH 2/6] fix(container-runner): report crashed child exits as errored stops --- container-runner/src/actor.rs | 17 +++++++---- container-runner/src/child.rs | 55 +++++++++++++++++++++++++---------- 2 files changed, 51 insertions(+), 21 deletions(-) diff --git a/container-runner/src/actor.rs b/container-runner/src/actor.rs index 8f7df3957d..ff9c53f621 100644 --- a/container-runner/src/actor.rs +++ b/container-runner/src/actor.rs @@ -158,24 +158,29 @@ impl Actor for GameServer { /// Watchdog: waits for the child to exit. Deliberate stops remove the /// child from the global registry first, so winning the `remove` race - /// means the exit was unexpected and the actor must be torn down. + /// means the exit was unexpected and the actor must be torn down. A clean + /// exit (code 0) destroys the actor; any other exit returns an error so + /// the framework reports an errored stop and the engine records the crash. async fn run(self: Arc, ctx: Ctx) -> Result<()> { let Some(child) = self.child.lock().await.clone() else { anyhow::bail!("run: child process was never spawned"); }; - let status = child.wait_exit().await; + let exit = child.wait_exit().await; let actor_id = ctx.actor_id().to_string(); if children().remove_async(&actor_id).await.is_some() { release_child_port(child.child_port).await; let prefix = log_prefix(&actor_id, child.key.as_deref()); + if !exit.success { + println!( + "{prefix} runner: child exited unexpectedly ({exit}), reporting errored stop" + ); + anyhow::bail!("child exited unexpectedly ({exit})"); + } println!( - "{prefix} runner: child exited unexpectedly ({status}), reporting actor stopped" + "{prefix} runner: child exited unexpectedly ({exit}), reporting actor stopped" ); - // TODO: `ctx.destroy()` reports a clean stop, while the old envoy - // path reported an errored stop that fed the engine's crash - // policy. The rivetkit wrapper has no errored-stop API yet. if let Err(err) = ctx.destroy() { // The actor may already be stopping if the engine beat us to it. tracing::debug!(error = ?err, actor_id = %actor_id, "destroy after child exit failed"); diff --git a/container-runner/src/child.rs b/container-runner/src/child.rs index 01c54cac50..64c4785074 100644 --- a/container-runner/src/child.rs +++ b/container-runner/src/child.rs @@ -19,6 +19,22 @@ use tokio::process::Command; use tokio::sync::watch; use tokio::time::{Instant, sleep}; +/// Terminal state of a child process. +#[derive(Clone)] +pub struct ChildExit { + /// True when the child exited on its own with code 0. Signal terminations, + /// nonzero exits, and wait errors are all failures. + pub success: bool, + /// Human-readable status, e.g. "exit status: 1" or "signal: 9 (SIGKILL)". + pub status: String, +} + +impl std::fmt::Display for ChildExit { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(&self.status) + } +} + /// A spawned child process, its listening port, and the actor identity used to /// prefix its logs. pub struct ChildProcess { @@ -26,8 +42,8 @@ pub struct ChildProcess { pub key: Option, pub child_port: u16, pid: u32, - /// `None` while running, `Some(status_string)` once the child has exited. - exited_rx: watch::Receiver>, + /// `None` while running, `Some(exit)` once the child has exited. + exited_rx: watch::Receiver>, } /// Everything needed to launch the child. @@ -100,16 +116,22 @@ impl ChildProcess { println!("{prefix} runner: spawned `{program}` (pid={pid}) on child port {child_port}"); // Reaper task owns the Child and reports its exit status. - let (exited_tx, exited_rx) = watch::channel::>(None); + let (exited_tx, exited_rx) = watch::channel::>(None); { let prefix = prefix.clone(); tokio::spawn(async move { - let status = match child.wait().await { - Ok(status) => status.to_string(), - Err(e) => format!("wait error: {e}"), + let exit = match child.wait().await { + Ok(status) => ChildExit { + success: status.success(), + status: status.to_string(), + }, + Err(e) => ChildExit { + success: false, + status: format!("wait error: {e}"), + }, }; - println!("{prefix} runner: child process exited (status: {status})"); - let _ = exited_tx.send(Some(status)); + println!("{prefix} runner: child process exited (status: {exit})"); + let _ = exited_tx.send(Some(exit)); }); } @@ -138,8 +160,8 @@ impl ChildProcess { let addr = (Ipv4Addr::LOCALHOST, self.child_port); let prefix = log_prefix(&self.actor_id, self.key.as_deref()); loop { - if let Some(status) = self.exited_rx.borrow().clone() { - anyhow::bail!("child exited before becoming ready (status: {status})"); + if let Some(exit) = self.exited_rx.borrow().clone() { + anyhow::bail!("child exited before becoming ready (status: {exit})"); } if TcpStream::connect(addr).await.is_ok() { println!( @@ -163,15 +185,18 @@ impl ChildProcess { self.exited_rx.borrow().is_some() } - /// Wait until the child exits (on its own or via `stop`), returning its status. - pub async fn wait_exit(&self) -> String { + /// Wait until the child exits (on its own or via `stop`), returning its exit state. + pub async fn wait_exit(&self) -> ChildExit { let mut rx = self.exited_rx.clone(); loop { - if let Some(status) = rx.borrow().clone() { - return status; + if let Some(exit) = rx.borrow().clone() { + return exit; } if rx.changed().await.is_err() { - return "reaper task ended".to_string(); + return ChildExit { + success: false, + status: "reaper task ended".to_string(), + }; } } } From 83a9044abd70a5a349e8ab97a62e416ae3bc190e Mon Sep 17 00:00:00 2001 From: ABCxFF <79597906+abcxff@users.noreply.github.com> Date: Tue, 14 Jul 2026 16:23:50 +0000 Subject: [PATCH 3/6] docs(compute): remove beta label from rivet compute --- frontend/packages/shared-data/src/deploy.ts | 44 ++++++++++++------- .../src/components/docs/ComputePricing.tsx | 7 ++- .../src/content/docs/deploy/rivet-compute.mdx | 4 -- website/src/data/compute-pricing.ts | 7 +-- 4 files changed, 34 insertions(+), 28 deletions(-) diff --git a/frontend/packages/shared-data/src/deploy.ts b/frontend/packages/shared-data/src/deploy.ts index 33f6f20129..74bc866356 100644 --- a/frontend/packages/shared-data/src/deploy.ts +++ b/frontend/packages/shared-data/src/deploy.ts @@ -12,9 +12,22 @@ import { faVercel, } from "@rivet-gg/icons"; +export type Provider = + | "rivet" + | "vercel" + | "cloudflare-workers" + | "supabase-functions" + | "railway" + | "kubernetes" + | "aws-ecs" + | "gcp-cloud-run" + | "hetzner" + | "custom" + | "custom-platform"; + export interface DeployOption { displayName: string; - name: string; + name: Provider; shortTitle?: string; href: string; description: string; @@ -24,19 +37,18 @@ export interface DeployOption { specializedPlatform?: boolean; } -export const deployOptions = [ +export const deployOptions: DeployOption[] = [ { displayName: "Rivet Compute", - name: "rivet" as const, + name: "rivet", href: "/docs/deploy/rivet-compute", description: "Deploy to Rivet's managed compute platform", icon: faRivet as any, - badge: "Beta", }, { displayName: "Vercel", - name: "vercel" as const, + name: "vercel", href: "/docs/deploy/vercel", description: "Deploy Next.js + RivetKit apps to Vercel's edge network", icon: faVercel as any, @@ -44,7 +56,7 @@ export const deployOptions = [ { displayName: "Cloudflare Workers", shortTitle: "Cloudflare", - name: "cloudflare-workers" as const, + name: "cloudflare-workers", href: "/docs/deploy/cloudflare", description: "Run RivetKit on Cloudflare Workers with the WebAssembly runtime", @@ -54,7 +66,7 @@ export const deployOptions = [ { displayName: "Supabase Functions", shortTitle: "Supabase", - name: "supabase-functions" as const, + name: "supabase-functions", href: "/docs/deploy/supabase", description: "Run RivetKit on Supabase Edge Functions with the WebAssembly runtime", @@ -63,14 +75,14 @@ export const deployOptions = [ }, { displayName: "Railway", - name: "railway" as const, + name: "railway", href: "/docs/deploy/railway", description: "Deploy containers to Railway's managed infrastructure", icon: faRailway as any, }, { displayName: "Kubernetes", - name: "kubernetes" as const, + name: "kubernetes", href: "/docs/deploy/kubernetes", description: "Deploy to any Kubernetes cluster with container images", icon: faKubernetes as any, @@ -78,7 +90,7 @@ export const deployOptions = [ { displayName: "AWS ECS", shortTitle: "AWS", - name: "aws-ecs" as const, + name: "aws-ecs", href: "/docs/deploy/aws-ecs", description: "Run containerized workloads on Amazon Elastic Container Service", @@ -87,21 +99,21 @@ export const deployOptions = [ { displayName: "Google Cloud Run", shortTitle: "GCP", - name: "gcp-cloud-run" as const, + name: "gcp-cloud-run", href: "/docs/deploy/gcp-cloud-run", description: "Deploy containers to Google Cloud Run for auto-scaling", icon: faGoogleCloud, }, { displayName: "Hetzner", - name: "hetzner" as const, + name: "hetzner", href: "/docs/deploy/hetzner", description: "Deploy to Hetzner's cost-effective cloud infrastructure", icon: faHetznerH as any, }, { displayName: "VM & Bare Metal", - name: "custom" as const, + name: "custom", shortTitle: "VM", href: "/docs/deploy/vm-and-bare-metal", description: @@ -110,12 +122,10 @@ export const deployOptions = [ }, { displayName: "Custom Platform", - name: "custom-platform" as const, + name: "custom-platform", href: "/docs/deploy/custom", description: "Integrate RivetKit with any other hosting platform of your choice", icon: faRocket as any, }, -] satisfies DeployOption[]; - -export type Provider = (typeof deployOptions)[number]["name"]; +]; diff --git a/website/src/components/docs/ComputePricing.tsx b/website/src/components/docs/ComputePricing.tsx index 565e41e80e..7ddfc95959 100644 --- a/website/src/components/docs/ComputePricing.tsx +++ b/website/src/components/docs/ComputePricing.tsx @@ -22,9 +22,8 @@ export function ComputePricing({ model = defaultModel }: ComputePricingProps) { return (

- Rivet Compute is in beta and per-unit rates are being - finalized ahead of general availability. You are billed on - the resources your running instances use: + Per-unit rates for Rivet Compute are being finalized. You + are billed on the resources your running instances use:

    {model.dimensions.map((d) => ( @@ -39,7 +38,7 @@ export function ComputePricing({ model = defaultModel }: ComputePricingProps) {

{model.billingGranularity} Contact the Rivet team for current - beta pricing. + pricing.

); diff --git a/website/src/content/docs/deploy/rivet-compute.mdx b/website/src/content/docs/deploy/rivet-compute.mdx index 9a9984ba3e..cb518b4431 100644 --- a/website/src/content/docs/deploy/rivet-compute.mdx +++ b/website/src/content/docs/deploy/rivet-compute.mdx @@ -4,10 +4,6 @@ description: "Run your backend on Rivet Compute." skill: true --- - -Rivet Compute is currently in beta. - - Using an AI coding agent? Open **Connect** on the [Rivet dashboard](https://dashboard.rivet.dev), select **Rivet Cloud**, and paste the one-shot prompt into your agent and have it connect with Rivet Compute for you. diff --git a/website/src/data/compute-pricing.ts b/website/src/data/compute-pricing.ts index 7be472a4d5..84d3dc1b0e 100644 --- a/website/src/data/compute-pricing.ts +++ b/website/src/data/compute-pricing.ts @@ -8,8 +8,8 @@ // // Fill in the real `memoryRate` / `cpuRate` coefficients and any included // allotment below, then flip `finalized` to true. While `finalized` is false -// (or any rate is left null), the docs render a beta notice instead of a rate -// table so no unverified numbers ship. +// (or any rate is left null), the docs render a pending-pricing notice instead +// of a rate table so no unverified numbers ship. export interface ComputePriceDimension { /** Human-readable dimension name, e.g. "Memory". */ @@ -32,7 +32,8 @@ export interface ComputeIncludedAllotment { export interface ComputePricingModel { /** * Flip to true once every `rate` below is filled in with a real value. While - * false, the docs page renders a beta notice rather than asserting rates. + * false, the docs page renders a pending-pricing notice rather than + * asserting rates. */ finalized: boolean; /** Currency symbol prefixed to rendered rates. */ From e0de75af4e1c57cd160374258258963ebd46135b Mon Sep 17 00:00:00 2001 From: ABCxFF <79597906+abcxff@users.noreply.github.com> Date: Tue, 14 Jul 2026 17:51:06 +0000 Subject: [PATCH 4/6] feat(dashboard): add compute tab to namespace settings drawer --- frontend/src/app/settings-drawer.tsx | 155 +++- .../app/settings-pages/namespace-compute.tsx | 700 ++++++++++++++++++ .../src/components/lib/create-schema-form.tsx | 8 + 3 files changed, 846 insertions(+), 17 deletions(-) create mode 100644 frontend/src/app/settings-pages/namespace-compute.tsx diff --git a/frontend/src/app/settings-drawer.tsx b/frontend/src/app/settings-drawer.tsx index b47c79f858..ed2c853ce9 100644 --- a/frontend/src/app/settings-drawer.tsx +++ b/frontend/src/app/settings-drawer.tsx @@ -5,6 +5,7 @@ import { faClose, faCreditCard, faGear, + faRivet, faSliders, faSparkles, Icon, @@ -19,6 +20,7 @@ import { } from "@tanstack/react-router"; import { type ReactNode, useEffect, useState } from "react"; import { cn, VisuallyHidden } from "@/components"; +import { useHasManagedPool } from "@/components/actors/actor-details-shared"; import { useCloudNamespaceDataProvider, useCloudProjectDataProvider, @@ -26,6 +28,7 @@ import { import { features } from "@/lib/features"; import { BillingUsageGauge } from "./billing/billing-usage-gauge"; import { BillingPanel } from "./settings-pages/billing-panel"; +import { NamespaceComputeContent } from "./settings-pages/namespace-compute"; import { NamespaceAdvancedContent, NamespaceSettingsContent, @@ -38,6 +41,7 @@ import { WhatsNewPanel } from "./settings-pages/whats-new-panel"; export type SettingsTab = | "profile" | "settings" + | "compute" | "advanced" | "billing" | "organization" @@ -59,6 +63,7 @@ const NAV_SECTIONS: Array<{ label: "Namespace", items: [ { key: "settings", label: "Settings", icon: faGear }, + { key: "compute", label: "Compute", icon: faRivet }, { key: "advanced", label: "Advanced", icon: faSliders }, ], }, @@ -89,6 +94,10 @@ const TAB_META: Record = { description: "Connect your RivetKit application to Rivet Cloud. Use your cloud of choice to run Rivet Actors.", }, + compute: { + title: "Compute", + description: "Manage the Rivet Compute deployment for this namespace.", + }, advanced: { title: "Advanced", description: @@ -197,23 +206,41 @@ export function SettingsDrawer({
{section.label}
- {section.items.map((item) => ( - - switchTab(item.key) - } - trailing={ - item.key === "billing" && - activeTab !== "billing" ? ( - - ) : null - } - /> - ))} + {section.items.map((item) => + item.key === "compute" ? ( + + switchTab(item.key) + } + /> + ) : ( + + switchTab(item.key) + } + trailing={ + item.key === + "billing" && + activeTab !== + "billing" ? ( + + ) : null + } + /> + ), + )} ))} @@ -272,6 +299,34 @@ function NavItem({ ); } +interface ComputeNavItemProps { + icon: IconProp; + label: string; + active: boolean; + onClick: () => void; +} + +// The Compute nav item only shows when the current namespace actively uses +// Rivet Compute. Guarded with `shouldThrow: false` plus a `loaderData` check +// because the drawer can render while the namespace match tree is +// mid-transition; the inner managed-pool hooks read the namespace data +// provider from loader data and would crash on undefined. The outer/inner +// split keeps hook order stable while the route match changes. +function ComputeNavItem(props: ComputeNavItemProps) { + const match = useMatch({ + from: "/_context/orgs/$organization/projects/$project/ns/$namespace", + shouldThrow: false, + }); + if (!features.compute || !match || !match.loaderData) return null; + return ; +} + +function ComputeNavItemInner(props: ComputeNavItemProps) { + const hasManagedPool = useHasManagedPool(); + if (!hasManagedPool) return null; + return ; +} + // Renders the usage gauge next to the Billing nav item. Guarded with // `shouldThrow: false` plus a `loaderData` check because the drawer can render // while the project match tree is mid-transition; without it the inner billing @@ -298,6 +353,8 @@ function TabContent({ tab }: { tab: SettingsTab }) { return ; case "settings": return ; + case "compute": + return ; case "advanced": return ; case "organization": @@ -458,6 +515,66 @@ function AdvancedTabBody() { return ; } +// Rivet Compute is cloud-platform-only, so there is no engine variant. The +// feature check is a backstop; `settingsParamToTab` already keeps the tab +// unreachable on flavors without the compute flag. +function ComputeTabBody() { + if (!features.compute) { + return null; + } + return ; +} + +function CloudComputeTabBody() { + const namespaceMatch = useMatch({ + from: "/_context/orgs/$organization/projects/$project/ns/$namespace", + shouldThrow: false, + }); + + if (!namespaceMatch) { + return ( + + ); + } + if (!namespaceMatch.loaderData) { + return ; + } + return ; +} + +// Deep links can land on the Compute tab for a namespace that does not use +// Rivet Compute. Redirect to the regular Settings tab once the managed-pool +// check resolves, and show the skeleton until then. +function CloudComputeTabBodyInner() { + const navigate = useNavigate(); + const provider = useCloudNamespaceDataProvider(); + const { data: hasManagedPool, isPending } = useQuery({ + ...provider.currentNamespaceHasManagedPoolQueryOptions(), + staleTime: Infinity, + }); + + useEffect(() => { + if (isPending || hasManagedPool) return; + navigate({ + to: ".", + search: (old) => ({ + ...(old as Record), + settings: "settings", + }), + }); + }, [isPending, hasManagedPool, navigate]); + + if (!hasManagedPool) { + return ; + } + return ; +} + function CloudAdvancedTabBody() { const namespaceMatch = useMatch({ from: "/_context/orgs/$organization/projects/$project/ns/$namespace", @@ -491,6 +608,10 @@ export function settingsParamToTab( case "organization": case "whats-new": return param; + // Compute is gated by the compute feature flag; degrade deep links to + // the regular Settings tab on flavors without it. + case "compute": + return features.compute ? "compute" : "settings"; // Legacy: members lived in its own tab before being merged into // Organization. Keep the deep link working. case "members": diff --git a/frontend/src/app/settings-pages/namespace-compute.tsx b/frontend/src/app/settings-pages/namespace-compute.tsx new file mode 100644 index 0000000000..634d6a66fd --- /dev/null +++ b/frontend/src/app/settings-pages/namespace-compute.tsx @@ -0,0 +1,700 @@ +import type { Rivet } from "@rivet-gg/cloud"; +import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query"; +import { type ReactNode, useEffect } from "react"; +import { useFormState } from "react-hook-form"; +import z from "zod"; +import { + Code, + cn, + createSchemaForm, + FormControl, + FormField, + FormItem, + FormLabel, + FormMessage, + Input, + Skeleton, + SmallText, +} from "@/components"; +import { useCloudNamespaceDataProvider } from "@/components/actors"; +import { Slider } from "@/components/ui/slider"; +import { Switch } from "@/components/ui/switch"; +import { SettingsCard } from "./settings-card"; + +const MEMORY_RE = /^(\d+)(Mi|Gi)$/; + +function parseMemoryMib(memory: string): number | null { + const match = MEMORY_RE.exec(memory); + if (!match) return null; + return Number(match[1]) * (match[2] === "Gi" ? 1024 : 1); +} + +// Formats pool args for the editable field, quoting entries that contain +// whitespace, quotes, or backslashes so the round-trip through `parseArgs` +// is lossless. +function formatArgs(args: string[]): string { + return args + .map((arg) => + arg === "" || /[\s"'\\]/.test(arg) + ? `"${arg.replace(/[\\"]/g, "\\$&")}"` + : arg, + ) + .join(" "); +} + +// Splits an args string on whitespace with shell-style quoting: single +// quotes group a literal token, double quotes group a token with `\"` and +// `\\` escapes, and a backslash outside quotes escapes the next character. +// Returns null on an unterminated quote so the schema can reject the input. +function parseArgs(input: string): string[] | null { + const args: string[] = []; + let current = ""; + let hasToken = false; + let i = 0; + while (i < input.length) { + const char = input[i]; + if (/\s/.test(char)) { + if (hasToken) { + args.push(current); + current = ""; + hasToken = false; + } + i++; + } else if (char === "'") { + const end = input.indexOf("'", i + 1); + if (end === -1) return null; + current += input.slice(i + 1, end); + hasToken = true; + i = end + 1; + } else if (char === '"') { + i++; + let closed = false; + while (i < input.length) { + if ( + input[i] === "\\" && + (input[i + 1] === '"' || input[i + 1] === "\\") + ) { + current += input[i + 1]; + i += 2; + } else if (input[i] === '"') { + closed = true; + i++; + break; + } else { + current += input[i]; + i++; + } + } + if (!closed) return null; + hasToken = true; + } else if (char === "\\" && i + 1 < input.length) { + current += input[i + 1]; + hasToken = true; + i += 2; + } else { + current += char; + hasToken = true; + i++; + } + } + if (hasToken) args.push(current); + return args; +} + +// A CPU value the deploy API accepts: the whole vCPU stops 1, 2, 4, and 8, +// or a fractional value below 1 with at most two decimals. +function isCatalogCpu(cpu: number): boolean { + if (cpu === 1 || cpu === 2 || cpu === 4 || cpu === 8) return true; + const scaled = cpu * 100; + return cpu < 1 && Math.abs(scaled - Math.round(scaled)) < 1e-9; +} + +function isValidCpu(cpu: number): boolean { + return Number.isFinite(cpu) && cpu >= 0.08 && cpu <= 8 && isCatalogCpu(cpu); +} + +// Constraints mirror the client-side validation of `rivet deploy` in +// engine/packages/cli/src/util.rs::build_resources and +// resolve_max_concurrent_actors. +const computeFormSchema = z + .object({ + maxConcurrentActors: z.coerce + .number() + .int("Must be a whole number") + .min(1, "Must be between 1 and 50,000") + .max(50000, "Must be between 1 and 50,000"), + drainGracePeriod: z.coerce + .number() + .int("Must be a whole number") + .min(5, "Must be between 5 and 3,200") + .max(3200, "Must be between 5 and 3,200"), + drainOnVersionUpgrade: z.boolean(), + command: z.string(), + args: z + .string() + .refine((args) => parseArgs(args) !== null, "Unterminated quote"), + minScale: z.coerce + .number() + .int("Must be a whole number") + .min(0, "Must be between 0 and 100") + .max(100, "Must be between 0 and 100"), + maxScale: z.coerce + .number() + .int("Must be a whole number") + .min(1, "Must be between 1 and 5,000") + .max(5000, "Must be between 1 and 5,000"), + cpu: z.coerce + .number() + .refine( + (cpu) => Number.isFinite(cpu) && cpu >= 0.08 && cpu <= 8, + "Must be between 0.08 and 8", + ) + .refine( + isCatalogCpu, + "Must be 1, 2, 4, or 8, or a value below 1 with at most two decimals", + ), + memory: z + .string() + .refine( + (memory) => MEMORY_RE.test(memory), + "Must match Mi or Gi, e.g. 512Mi", + ) + .refine((memory) => { + const mib = parseMemoryMib(memory); + return mib != null && mib >= 512 && mib <= 4096; + }, "Must be between 512Mi and 4Gi"), + instanceRequestConcurrency: z.coerce + .number() + .int("Must be a whole number") + .min(1, "Must be between 1 and 500") + .max(500, "Must be between 1 and 500"), + }) + .superRefine((values, ctx) => { + if (values.minScale > values.maxScale) { + const message = "Min scale must not exceed max scale"; + ctx.addIssue({ code: "custom", path: ["minScale"], message }); + ctx.addIssue({ code: "custom", path: ["maxScale"], message }); + } + }); + +type ComputeFormValues = z.infer; + +const { + Form, + Submit, + Reset, + useContext: useComputeForm, +} = createSchemaForm(computeFormSchema); + +// The namespace data provider can briefly resolve to `undefined` when the +// user switches namespace/project from the nav while this drawer is open. +// Bail until the new route's loader has populated `dataProvider`. The +// outer/inner split keeps the inner component's hook order stable while the +// route match changes, mirroring `ComputeNavItem` in settings-drawer.tsx. +export function NamespaceComputeContent() { + const dataProvider = useCloudNamespaceDataProvider(); + if (!dataProvider) { + return null; + } + return ; +} + +function NamespaceComputeContentInner() { + const dataProvider = useCloudNamespaceDataProvider(); + + const { + data: pool, + isPending, + isError, + } = useQuery({ + ...dataProvider.currentNamespaceManagedPoolQueryOptions({ + pool: "default", + }), + // Poll while a deploy is in flight so the footer button tracks the + // pool status; back off entirely once the pool settles. + refetchInterval: (query) => + isPoolBusy(query.state.data?.status) ? 3_000 : false, + }); + + if (isPending) { + return ( +
+ + +
+ ); + } + + if (isError || !pool?.config) { + return ( + + + Failed to load the Rivet Compute deployment for this + namespace. + + + ); + } + + return ; +} + +function isPoolBusy( + status: Rivet.ManagedPoolsGetResponse.ManagedPool.Status | undefined, +) { + return ( + status === "initializing" || + status === "allocating" || + status === "deploying" || + status === "binding" || + status === "destroying" + ); +} + +function ComputeForm({ + config, + status, +}: { + config: Rivet.ManagedPoolsGetResponse.ManagedPool.Config; + status: Rivet.ManagedPoolsGetResponse.ManagedPool.Status; +}) { + const dataProvider = useCloudNamespaceDataProvider(); + const queryClient = useQueryClient(); + const { mutateAsync } = useMutation( + dataProvider.upsertCurrentNamespaceManagedPoolMutationOptions(), + ); + + const resources = config.resources; + const runnerConfig = config.runnerConfig; + + // Unset fields fall back to the server-side defaults so submitting an + // untouched field pins the value the pool already effectively runs with. + // The deprecated top-level maxConcurrentActors seeds the runner config + // value for pools that predate the runnerConfig object. + const defaultValues: ComputeFormValues = { + maxConcurrentActors: + runnerConfig?.maxConcurrentActors ?? + config.maxConcurrentActors ?? + 1000, + drainGracePeriod: runnerConfig?.drainGracePeriod ?? 1800, + drainOnVersionUpgrade: runnerConfig?.drainOnVersionUpgrade ?? true, + command: config.command ?? "", + args: formatArgs(config.args ?? []), + minScale: resources?.minScale ?? 0, + maxScale: resources?.maxScale ?? 1, + cpu: resources?.cpu ?? 1, + memory: resources?.memory ?? "512Mi", + instanceRequestConcurrency: resources?.instanceRequestConcurrency ?? 80, + }; + + return ( +
{ + // The upsert merges field-wise: an omitted field keeps the + // pool's current value and an explicit null clears it. The + // command/args overrides are therefore omitted when unchanged + // and sent as null when the user cleared them. `image` is + // always omitted so the pool keeps its current build, + // matching `rivet deploy --reuse-image`. + const command = values.command.trim(); + const args = parseArgs(values.args); + if (args === null) { + throw new Error( + "args failed to parse after schema validation", + ); + } + const initialArgs = config.args ?? []; + const overrides: { + command?: string | null; + args?: string[] | null; + } = {}; + if (command !== (config.command ?? "").trim()) { + overrides.command = command === "" ? null : command; + } + if ( + args.length !== initialArgs.length || + args.some((arg, index) => arg !== initialArgs[index]) + ) { + overrides.args = args.length === 0 ? null : args; + } + await mutateAsync({ + pool: "default", + displayName: config.displayName || "Default", + runnerConfig: { + maxConcurrentActors: values.maxConcurrentActors, + drainGracePeriod: values.drainGracePeriod, + drainOnVersionUpgrade: values.drainOnVersionUpgrade, + }, + // The generated SDK request type does not model the + // clear-on-null semantics, so the null-capable overrides + // are cast back to the generated field types. + ...(overrides as { command?: string; args?: string[] }), + resources: { + minScale: values.minScale, + maxScale: values.maxScale, + cpu: values.cpu, + memory: values.memory, + instanceRequestConcurrency: + values.instanceRequestConcurrency, + }, + }); + await queryClient.invalidateQueries( + dataProvider.currentNamespaceManagedPoolQueryOptions({ + pool: "default", + }), + ); + form.reset(values); + }} + > +
+ + Image

} + > + {config.image ? ( + + {config.image.repository}:{config.image.tag} + + ) : ( + + None + + )} + + + + + + + + + + + + + + + + + +
+
+ ); +} + +// Renders the discard/deploy controls once the form has edits or a deploy is +// in flight, so the tab reads as a plain settings sheet otherwise. While the +// pool works through a deploy the submit button is locked and tracks the +// polled pool status. +function DeployFooter({ + status, +}: { + status: Rivet.ManagedPoolsGetResponse.ManagedPool.Status; +}) { + const { isDirty } = useFormState(); + const busy = isPoolBusy(status); + if (!isDirty && !busy) return null; + return ( +
+ {isDirty ? ( + + Discard + + ) : null} + + {busy ? "Deploying..." : "Deploy changes"} + +
+ ); +} + +const FIELD_LABEL_CLASS = "pt-2 text-sm font-normal text-muted-foreground"; + +function FieldRow({ + label, + children, +}: { + label: ReactNode; + children: ReactNode; +}) { + return ( +
+ {label} +
+ {children} +
+
+ ); +} + +const FIELD_ITEM_CLASS = + "flex flex-row items-start justify-between gap-4 space-y-0 px-5 py-3"; + +function TextField({ + name, + label, + placeholder, + narrow = false, +}: { + name: "command" | "args" | "memory"; + label: string; + placeholder?: string; + narrow?: boolean; +}) { + const { control } = useComputeForm(); + return ( + ( + + {label} +
+ + + + +
+ + )} + /> + ); +} + +function NumberField({ + name, + label, +}: { + name: + | "maxConcurrentActors" + | "drainGracePeriod" + | "minScale" + | "maxScale" + | "instanceRequestConcurrency"; + label: string; +}) { + const { control } = useComputeForm(); + return ( + ( + + {label} +
+ + + + +
+ + )} + /> + ); +} + +function SwitchField({ + name, + label, +}: { + name: "drainOnVersionUpgrade"; + label: string; +}) { + const { control } = useComputeForm(); + return ( + ( + + {label} + + + + + )} + /> + ); +} + +// The CPU slider walks the fractional range [0.08, 1.00] in 0.01 steps on +// the left of the track, then jumps between the whole vCPU values 1, 2, 4, +// and 8 as magnetic stops on the right, mirroring the values `rivet deploy` +// accepts. Positions are integer indices so the fractional zone stays evenly +// spaced while the whole stops get wider spacing than one step each. +const CPU_FRACTIONAL_MAX_INDEX = 92; // (1.00 - 0.08) / 0.01 +const CPU_WHOLE_STOPS = [ + { index: 92, cpu: 1 }, + { index: 122, cpu: 2 }, + { index: 152, cpu: 4 }, + { index: 182, cpu: 8 }, +]; +const CPU_SLIDER_MAX = 182; +const CPU_TICKS = [ + { label: "0.08", index: 0 }, + { label: "1", index: 92 }, + { label: "2", index: 122 }, + { label: "4", index: 152 }, + { label: "8", index: 182 }, +]; + +function sliderIndexToCpu(index: number): number { + if (index <= CPU_FRACTIONAL_MAX_INDEX) { + return (index + 8) / 100; + } + let nearest = CPU_WHOLE_STOPS[0]; + for (const stop of CPU_WHOLE_STOPS) { + if (Math.abs(stop.index - index) < Math.abs(nearest.index - index)) { + nearest = stop; + } + } + return nearest.cpu; +} + +function cpuToSliderIndex(cpu: number): number { + if (!Number.isFinite(cpu)) { + return CPU_FRACTIONAL_MAX_INDEX; + } + if (cpu <= 1) { + return Math.min( + Math.max(Math.round(cpu * 100) - 8, 0), + CPU_FRACTIONAL_MAX_INDEX, + ); + } + let nearest = CPU_WHOLE_STOPS[0]; + for (const stop of CPU_WHOLE_STOPS) { + if (Math.abs(stop.cpu - cpu) < Math.abs(nearest.cpu - cpu)) { + nearest = stop; + } + } + return nearest.index; +} + +function formatCpu(cpu: number): string { + if (!Number.isFinite(cpu)) { + return "1"; + } + return cpu < 1 ? cpu.toFixed(2) : String(cpu); +} + +function CpuField() { + const { control, getValues, trigger } = useComputeForm(); + // An out-of-catalog CPU value from the server cannot be represented by a + // slider position (the thumb snaps to the nearest stop), so surface the + // validation error immediately instead of waiting for the first + // interaction with the field. + useEffect(() => { + if (!isValidCpu(Number(getValues("cpu")))) { + void trigger("cpu"); + } + }, [getValues, trigger]); + return ( + { + const cpu = Number(field.value); + return ( + + CPU +
+ + {formatCpu(cpu)} vCPU + + + + field.onChange(sliderIndexToCpu(next)) + } + /> + +
+ {CPU_TICKS.map((tick) => ( + + {tick.label} + + ))} +
+ +
+ + ); + }} + /> + ); +} diff --git a/frontend/src/components/lib/create-schema-form.tsx b/frontend/src/components/lib/create-schema-form.tsx index f3916c12c3..8dbd1cd798 100644 --- a/frontend/src/components/lib/create-schema-form.tsx +++ b/frontend/src/components/lib/create-schema-form.tsx @@ -20,6 +20,12 @@ interface FormProps onSubmit: SubmitHandler; mode?: UseFormProps["mode"]; revalidateMode?: UseFormProps["reValidateMode"]; + /** + * Debounce error display by this many milliseconds. Only useful together + * with an eager `mode` like "onChange", where it batches per-keystroke + * error updates into one. + */ + delayError?: UseFormProps["delayError"]; defaultValues: DefaultValues; errors?: UseFormProps["errors"]; values?: FormValues; @@ -39,6 +45,7 @@ export const createSchemaForm = >( defaultValues, mode, revalidateMode = "onSubmit", + delayError, values, children, onSubmit, @@ -52,6 +59,7 @@ export const createSchemaForm = >( ), mode, reValidateMode: revalidateMode, + delayError, defaultValues, values, errors, From 661376df0bab5c479da47cdfc1698cca3d361733 Mon Sep 17 00:00:00 2001 From: ABCxFF <79597906+abcxff@users.noreply.github.com> Date: Wed, 15 Jul 2026 14:19:38 +0000 Subject: [PATCH 5/6] feat(dashboard): allow typing cpu value in compute tab --- .../app/settings-pages/namespace-compute.tsx | 47 ++++++++++--------- 1 file changed, 25 insertions(+), 22 deletions(-) diff --git a/frontend/src/app/settings-pages/namespace-compute.tsx b/frontend/src/app/settings-pages/namespace-compute.tsx index 634d6a66fd..f66ad1d00f 100644 --- a/frontend/src/app/settings-pages/namespace-compute.tsx +++ b/frontend/src/app/settings-pages/namespace-compute.tsx @@ -574,7 +574,10 @@ function SwitchField({ // the left of the track, then jumps between the whole vCPU values 1, 2, 4, // and 8 as magnetic stops on the right, mirroring the values `rivet deploy` // accepts. Positions are integer indices so the fractional zone stays evenly -// spaced while the whole stops get wider spacing than one step each. +// spaced while the whole stops get wider spacing than one step each. The +// value can also be typed directly into the input next to the slider; while +// a typed value is out of catalog the thumb snaps to the nearest stop and +// the field shows a validation error. const CPU_FRACTIONAL_MAX_INDEX = 92; // (1.00 - 0.08) / 0.01 const CPU_WHOLE_STOPS = [ { index: 92, cpu: 1 }, @@ -623,13 +626,6 @@ function cpuToSliderIndex(cpu: number): number { return nearest.index; } -function formatCpu(cpu: number): string { - if (!Number.isFinite(cpu)) { - return "1"; - } - return cpu < 1 ? cpu.toFixed(2) : String(cpu); -} - function CpuField() { const { control, getValues, trigger } = useComputeForm(); // An out-of-catalog CPU value from the server cannot be represented by a @@ -651,20 +647,27 @@ function CpuField() { CPU
- - {formatCpu(cpu)} vCPU - - - - field.onChange(sliderIndexToCpu(next)) - } - /> - +
+ + + + + vCPU + +
+ + field.onChange(sliderIndexToCpu(next)) + } + />
{CPU_TICKS.map((tick) => ( Date: Tue, 14 Jul 2026 19:25:31 +0000 Subject: [PATCH 6/6] feat(dashboard): add connection token issuing ui --- .../data-providers/cloud-data-provider.tsx | 47 +++++++++++-- .../app/settings-pages/namespace-settings.tsx | 2 + .../ns.$namespace/tokens.tsx | 66 +++++++++++++++++++ 3 files changed, 108 insertions(+), 7 deletions(-) diff --git a/frontend/src/app/data-providers/cloud-data-provider.tsx b/frontend/src/app/data-providers/cloud-data-provider.tsx index fec53499e1..e66e8789f6 100644 --- a/frontend/src/app/data-providers/cloud-data-provider.tsx +++ b/frontend/src/app/data-providers/cloud-data-provider.tsx @@ -30,13 +30,11 @@ function createClient() { delete args.headers?.[key]; } }); - return await fetcher( - { - ...args, - maxRetries: 1, - withCredentials: true, - }, - ); + return await fetcher({ + ...args, + maxRetries: 1, + withCredentials: true, + }); }, }); } @@ -1184,6 +1182,41 @@ export const createNamespaceContext = ({ }, }); }, + createConnectionTokenMutationOptions() { + return mutationOptions({ + mutationKey: [ + { + namespace, + project: parent.project, + organization: parent.organization, + }, + "tokens", + "connection", + "create", + ], + mutationFn: async () => { + // The pinned cloud SDK does not expose the connection + // token endpoint yet, so this calls cloud-api directly + // with the session cookie. Replace with + // client.namespaces.createConnectionToken once the SDK + // ships it. + const url = `${cloudEnv().VITE_APP_CLOUD_API_URL}/projects/${encodeURIComponent(parent.project)}/namespaces/${encodeURIComponent(namespace)}/tokens/connection?org=${encodeURIComponent(parent.organization)}`; + const response = await fetch(url, { + method: "POST", + credentials: "include", + }); + if (!response.ok) { + throw new Error( + `Failed to issue connection token (${response.status})`, + ); + } + return (await response.json()) as { + token: string; + expiresAt?: string; + }; + }, + }); + }, publishableTokenQueryOptions() { return queryOptions({ staleTime: 5 * 60 * 1000, // 5 minutes diff --git a/frontend/src/app/settings-pages/namespace-settings.tsx b/frontend/src/app/settings-pages/namespace-settings.tsx index 01e9c7d420..a75967fa7d 100644 --- a/frontend/src/app/settings-pages/namespace-settings.tsx +++ b/frontend/src/app/settings-pages/namespace-settings.tsx @@ -20,6 +20,7 @@ import { NoProvidersAlert } from "@/components/actors/no-providers-alert"; import { features } from "@/lib/features"; import { CloudApiTokens, + ConnectionTokens, PublishableToken, SecretToken, } from "@/routes/_context/orgs.$organization/projects.$project/ns.$namespace/tokens"; @@ -49,6 +50,7 @@ export function NamespaceAdvancedContent() {
+ {features.auth ? : null} {features.auth ? : null} {features.dangerZone ? : null} diff --git a/frontend/src/routes/_context/orgs.$organization/projects.$project/ns.$namespace/tokens.tsx b/frontend/src/routes/_context/orgs.$organization/projects.$project/ns.$namespace/tokens.tsx index 18b65d9066..e132d9091b 100644 --- a/frontend/src/routes/_context/orgs.$organization/projects.$project/ns.$namespace/tokens.tsx +++ b/frontend/src/routes/_context/orgs.$organization/projects.$project/ns.$namespace/tokens.tsx @@ -356,6 +356,72 @@ registry.startEnvoy();`; ); } +// The namespace data provider can briefly resolve to `undefined` when the +// user switches namespace/project from the nav while this drawer is open. +// The outer/inner split keeps the inner component's hook order stable while +// the route match changes. +export function ConnectionTokens() { + const dataProvider = useCloudNamespaceDataProvider(); + if (!dataProvider) { + return null; + } + return ; +} + +function ConnectionTokensInner() { + const dataProvider = useCloudNamespaceDataProvider(); + + const { + mutate: issueToken, + data, + isPending, + error, + } = useMutation(dataProvider.createConnectionTokenMutationOptions()); + + return ( + + Connection Tokens + Beta + + } + description="Connection tokens can only open connections to existing actors. Use them for clients that must not create actors or manage the namespace." + action={ + + } + > + {data ? ( +
+ +

+ {data.expiresAt + ? `Expires ${new Date(data.expiresAt).toLocaleString()}.` + : "Does not expire."}{" "} + This token is only shown once, copy it now. +

+
+ ) : error ? ( +

+ Failed to issue a connection token. {error.message} +

+ ) : ( +

+ Issue a token to reveal it here. +

+ )} + + ); +} + export function CloudApiTokens() { const { dataProvider } = useRouteContext({ from: "/_context/orgs/$organization/projects/$project",