From 7139815652850150a8bc6971067c3077e2a3b31c Mon Sep 17 00:00:00 2001 From: MasterPtato Date: Fri, 4 Sep 2026 12:17:00 -0700 Subject: [PATCH 1/4] fix(envoy-client): ack start commands immediately instead of waiting for the periodic tick --- engine/sdks/rust/envoy-client/src/commands.rs | 25 ++++++---- .../rust/envoy-client/tests/command_dedup.rs | 50 +++++++++++++++++++ 2 files changed, 64 insertions(+), 11 deletions(-) diff --git a/engine/sdks/rust/envoy-client/src/commands.rs b/engine/sdks/rust/envoy-client/src/commands.rs index ca35743e98..5d4180e4a8 100644 --- a/engine/sdks/rust/envoy-client/src/commands.rs +++ b/engine/sdks/rust/envoy-client/src/commands.rs @@ -18,11 +18,10 @@ pub async fn handle_commands(ctx: &mut EnvoyContext, commands: Vec = commands + // Collect every actor in the raw batch before dedup, so a replayed + // (skipped) command is still re-acked instead of being replayed forever. + let batch_actors: Vec<(String, u32)> = commands .iter() - .filter(|c| matches!(c.inner, protocol::Command::CommandStopActor(_))) .map(|c| (c.checkpoint.actor_id.clone(), c.checkpoint.generation)) .collect(); @@ -91,18 +90,22 @@ pub async fn handle_commands(ctx: &mut EnvoyContext, commands: Vec = HashMap::new(); for key in actors { if let Some(&index) = ctx.processed_command_idx.get(key) { diff --git a/engine/sdks/rust/envoy-client/tests/command_dedup.rs b/engine/sdks/rust/envoy-client/tests/command_dedup.rs index b058b25862..cac7b177dd 100644 --- a/engine/sdks/rust/envoy-client/tests/command_dedup.rs +++ b/engine/sdks/rust/envoy-client/tests/command_dedup.rs @@ -140,6 +140,26 @@ fn stop_command(actor_id: &str, generation: u32, index: i64) -> protocol::Comman } } +fn start_command(actor_id: &str, generation: u32, index: i64) -> protocol::CommandWrapper { + protocol::CommandWrapper { + checkpoint: protocol::ActorCheckpoint { + actor_id: actor_id.to_string(), + generation, + index, + }, + inner: protocol::Command::CommandStartActor(protocol::CommandStartActor { + config: protocol::ActorConfig { + name: actor_id.to_string(), + key: None, + create_ts: 0, + input: None, + }, + hibernating_requests: Vec::new(), + preloaded_kv: None, + }), + } +} + fn execute_request() -> protocol::SqliteExecuteRequest { protocol::SqliteExecuteRequest { namespace_id: "test".to_string(), @@ -289,6 +309,36 @@ fn decode_ack_checkpoints(msg: WsTxMessage) -> Vec { } } +#[tokio::test] +async fn start_command_is_acked_immediately() { + let mut ctx = new_envoy_context(); + let (ws_tx, mut ws_rx) = mpsc::unbounded_channel(); + *ctx.shared.ws_tx.lock().await = Some(ws_tx); + + handle_commands(&mut ctx, vec![start_command("actor-a", 1, 1)]).await; + + // A start left unacked stays in the engine's command subspace, which is + // re-streamed on every reconnect. Waiting for the periodic tick leaves a + // window of `ACK_COMMANDS_INTERVAL_MS` in which a reconnect replays the + // start and replaces the live actor. + let checkpoints = decode_ack_checkpoints( + ws_rx + .try_recv() + .expect("start should trigger an immediate ack"), + ); + assert_eq!(checkpoints.len(), 1); + assert_eq!(checkpoints[0].actor_id, "actor-a"); + assert_eq!(checkpoints[0].generation, 1); + assert_eq!(checkpoints[0].index, 1); + + // Dedup is retained so a replay can still be suppressed in-process until + // the tick clears it. + assert_eq!( + ctx.processed_command_idx.get(&("actor-a".to_string(), 1)), + Some(&1) + ); +} + #[tokio::test] async fn stop_command_is_acked_immediately() { let mut ctx = new_envoy_context(); From 3f6314f440f0a1dec90344e400039f92554a8b19 Mon Sep 17 00:00:00 2001 From: MasterPtato Date: Fri, 4 Sep 2026 12:17:00 -0700 Subject: [PATCH 2/4] fix(rivetkit-core): report actor crashes as a sleep intent instead of a stop intent --- .../engine/tests/common/test_envoy.rs | 4 +- engine/sdks/rust/envoy-client/src/handle.rs | 17 +++- .../rivetkit-core/src/actor/context.rs | 89 ++++++++++++----- .../packages/rivetkit-core/src/actor/sleep.rs | 28 ++++-- .../rivetkit-core/src/actor/sqlite/mod.rs | 5 +- .../packages/rivetkit-core/tests/sleep.rs | 97 +++++++++++++++++-- .../packages/rivetkit-core/tests/sqlite.rs | 4 +- .../packages/rivetkit-core/tests/task.rs | 30 +++--- 8 files changed, 210 insertions(+), 64 deletions(-) diff --git a/engine/packages/engine/tests/common/test_envoy.rs b/engine/packages/engine/tests/common/test_envoy.rs index 9a8a97460d..071bb5aa52 100644 --- a/engine/packages/engine/tests/common/test_envoy.rs +++ b/engine/packages/engine/tests/common/test_envoy.rs @@ -547,10 +547,10 @@ fn spawn_event_bridge(handle: EnvoyHandle, mut event_rx: mpsc::UnboundedReceiver rivet_runner_protocol::mk2::Event::EventActorIntent(intent) => { match intent.intent { rivet_runner_protocol::mk2::ActorIntent::ActorIntentSleep => { - handle.sleep_actor(event.actor_id, Some(event.generation)); + handle.sleep_actor(event.actor_id, Some(event.generation), None); } rivet_runner_protocol::mk2::ActorIntent::ActorIntentStop => { - handle.stop_actor(event.actor_id, Some(event.generation), None); + handle.stop_actor(event.actor_id, Some(event.generation)); } } } diff --git a/engine/sdks/rust/envoy-client/src/handle.rs b/engine/sdks/rust/envoy-client/src/handle.rs index 24c734a243..744e56e932 100644 --- a/engine/sdks/rust/envoy-client/src/handle.rs +++ b/engine/sdks/rust/envoy-client/src/handle.rs @@ -151,26 +151,35 @@ impl EnvoyHandle { Ok(()) } - pub fn sleep_actor(&self, actor_id: String, generation: Option) { + /// Reports a sleep intent for an actor. An `error` marks the sleep as a + /// crash: it surfaces as `StopCode::Error` on the eventual `Stopped` event, + /// which is what the engine records the crash from. The engine answers a + /// crashed stop by putting the actor back to sleep rather than destroying + /// it, so a crash belongs here rather than on [`Self::stop_actor`]. + pub fn sleep_actor(&self, actor_id: String, generation: Option, error: Option) { let _ = crate::envoy::send_to_envoy_tx( &self.shared, ToEnvoyMessage::ActorIntent { actor_id, generation, intent: protocol::ActorIntent::ActorIntentSleep, - error: None, + error, }, ); } - pub fn stop_actor(&self, actor_id: String, generation: Option, error: Option) { + /// Reports a stop intent for an actor. This is the deliberate-destruction + /// signal: the engine answers it by destroying the actor and its durable + /// state. It takes no error by construction, because a crash must not be + /// reported as an intent to destroy. + pub fn stop_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, + error: None, }, ); } diff --git a/rivetkit-rust/packages/rivetkit-core/src/actor/context.rs b/rivetkit-rust/packages/rivetkit-core/src/actor/context.rs index fde8e5e7e7..cfc312a148 100644 --- a/rivetkit-rust/packages/rivetkit-core/src/actor/context.rs +++ b/rivetkit-rust/packages/rivetkit-core/src/actor/context.rs @@ -551,11 +551,13 @@ impl ActorContext { 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. + /// Request a stop with an error attached. The envoy reports the stop to the + /// engine with `StopCode::Error` and the message, and the engine answers a + /// crash by putting the actor back to sleep rather than destroying it. The + /// local teardown therefore takes the sleep path too, so `onSleep` runs, + /// hibernatable connections are preserved, the persisted alarm stays armed + /// for the next generation, and incoming work is not rejected with + /// `Destroying` for an actor that is about to resume. pub fn stop_with_error(&self, message: impl Into) -> Result<()> { self.request_stop(Some(truncate_stop_error_message(message.into()))) } @@ -569,33 +571,60 @@ impl ActorContext { && !self.0.destroy_requested.load(Ordering::SeqCst) { return Err(ActorLifecycleError::Starting.build()) - .context("cannot request destroy before actor startup completes"); + .context("cannot request stop before actor startup completes"); } - if self.0.destroy_requested.swap(true, Ordering::SeqCst) { - 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; + + if let Some(error) = error { + // An errored stop is a crash report, not a destroy. A destroy that + // already won owns the teardown, so leave it alone. + if self.0.destroy_requested.load(Ordering::SeqCst) { + return Err(ActorLifecycleError::Stopping.build()) + .context("destroy already requested for this generation"); + } + // Record the error even when a sleep is already in flight. The + // envoy attaches it to the actor regardless of whether the intent + // itself is a duplicate, so the eventual `Stopped` still carries + // `StopCode::Error` and the engine still records the crash. + let queued_error = self.0.sleep.stop_error.lock().replace(error); + if !self.0.sleep_requested.swap(true, Ordering::SeqCst) { + self.mark_errored_stop_requested(); + } + // An errored stop is already queued and has not reached the envoy + // yet. It picks up the error recorded above, so sending a second + // intent would only race an error-less `stop_actor` against it: + // `request_stop_from_envoy` consumes the single error slot with + // `take`, and the loser reports the crash as a deliberate destroy. + // A sleep that is already in flight leaves the slot empty, so the + // `sleep()` -> `stop_with_error()` upgrade still reports here. + if queued_error.is_some() { + return Ok(()); + } + } else { + if self.0.destroy_requested.swap(true, Ordering::SeqCst) { + return Err(ActorLifecycleError::Stopping.build()) + .context("destroy already requested for this generation"); + } + // A destroy supersedes an errored stop that has not reached the + // envoy yet. Without clearing the slot, whichever request runs + // first consumes the error and reports this destroy as a sleep + // intent, leaving the actor alive. + *self.0.sleep.stop_error.lock() = None; + // 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 `store(true)` inside is harmless. + #[cfg(not(feature = "wasm-runtime"))] + self.mark_destroy_requested(); + #[cfg(feature = "wasm-runtime")] + self.mark_destroy_requested_without_spawn(); } - // 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 - // `store(true)` inside is harmless. - #[cfg(not(feature = "wasm-runtime"))] - self.mark_destroy_requested(); - #[cfg(feature = "wasm-runtime")] - self.mark_destroy_requested_without_spawn(); let ctx = self.clone(); if Handle::try_current().is_ok() { let tracked = self.track_shutdown_task(async move { ctx.record_user_task_started(UserTaskKind::DestroyRequest); let started_at = Instant::now(); - ctx.request_destroy_from_envoy(); + ctx.request_stop_from_envoy(); ctx.record_user_task_finished(UserTaskKind::DestroyRequest, started_at.elapsed()); }); if tracked { @@ -603,7 +632,7 @@ impl ActorContext { } } - self.request_destroy_from_envoy(); + self.request_stop_from_envoy(); Ok(()) } @@ -614,6 +643,16 @@ impl ActorContext { self.0.destroy_completed.store(false, Ordering::SeqCst); } + /// Teardown bookkeeping for an errored stop. Mirrors + /// `mark_destroy_requested` minus the destroy flags, since the actor is + /// heading for the sleep path. The state flush still runs so a crash + /// persists whatever the actor had before its teardown. + fn mark_errored_stop_requested(&self) { + self.cancel_sleep_timer(); + #[cfg(not(feature = "wasm-runtime"))] + self.flush_on_shutdown(); + } + #[cfg(feature = "wasm-runtime")] fn mark_destroy_requested_without_spawn(&self) { self.cancel_sleep_timer(); diff --git a/rivetkit-rust/packages/rivetkit-core/src/actor/sleep.rs b/rivetkit-rust/packages/rivetkit-core/src/actor/sleep.rs index f12270b2ce..e224e5127d 100644 --- a/rivetkit-rust/packages/rivetkit-core/src/actor/sleep.rs +++ b/rivetkit-rust/packages/rivetkit-core/src/actor/sleep.rs @@ -47,9 +47,10 @@ 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>, + // Forced-sync: set by an errored stop, then consumed when the intent is + // sent to the envoy. Its presence is what makes the intent a sleep rather + // than a destroy. + pub(super) stop_error: Mutex>, #[cfg(test)] sleep_request_count: TestAtomicUsize, #[cfg(test)] @@ -82,7 +83,7 @@ impl SleepState { envoy_handle: Mutex::new(None), generation: Mutex::new(None), http_request_counter: Mutex::new(None), - destroy_error: Mutex::new(None), + stop_error: Mutex::new(None), #[cfg(test)] sleep_request_count: TestAtomicUsize::new(0), #[cfg(test)] @@ -163,11 +164,11 @@ impl ActorContext { let envoy_handle = self.0.sleep.envoy_handle.lock().clone(); let generation = *self.0.sleep.generation.lock(); if let Some(envoy_handle) = envoy_handle { - envoy_handle.sleep_actor(self.actor_id().to_owned(), generation); + envoy_handle.sleep_actor(self.actor_id().to_owned(), generation, None); } } - pub(crate) fn request_destroy_from_envoy(&self) { + pub(crate) fn request_stop_from_envoy(&self) { #[cfg(test)] self.0 .sleep @@ -175,9 +176,18 @@ 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.stop_actor(self.actor_id().to_owned(), generation, error); + let error = self.0.sleep.stop_error.lock().take(); + let Some(envoy_handle) = envoy_handle else { + return; + }; + // A crash is reported as a sleep intent. The engine puts a crashed + // actor back to sleep rather than destroying it, and `StopIntent` is + // reserved for a deliberate destroy. + match error { + Some(error) => { + envoy_handle.sleep_actor(self.actor_id().to_owned(), generation, Some(error)) + } + None => envoy_handle.stop_actor(self.actor_id().to_owned(), generation), } } diff --git a/rivetkit-rust/packages/rivetkit-core/src/actor/sqlite/mod.rs b/rivetkit-rust/packages/rivetkit-core/src/actor/sqlite/mod.rs index 717a971c23..c5cc659d00 100644 --- a/rivetkit-rust/packages/rivetkit-core/src/actor/sqlite/mod.rs +++ b/rivetkit-rust/packages/rivetkit-core/src/actor/sqlite/mod.rs @@ -1227,7 +1227,10 @@ fn report_sqlite_worker_fatal(reported: &AtomicBool, config: SqliteRuntimeConfig // A dead worker means SQLite's sole native connection is no longer a valid // actor subsystem. Core reports that through envoy lifecycle instead of // letting the actor continue to serve requests with a broken database. - config.handle.stop_actor( + // This is a crash, not a deliberate destroy, so it goes out as a sleep + // intent: the next generation opens a fresh worker over the same durable + // state. + config.handle.sleep_actor( config.actor_id, config .generation diff --git a/rivetkit-rust/packages/rivetkit-core/tests/sleep.rs b/rivetkit-rust/packages/rivetkit-core/tests/sleep.rs index 89e6576345..821bd2df8a 100644 --- a/rivetkit-rust/packages/rivetkit-core/tests/sleep.rs +++ b/rivetkit-rust/packages/rivetkit-core/tests/sleep.rs @@ -669,7 +669,7 @@ mod moved_tests { async fn recv_stop_intent( rx: &mut mpsc::UnboundedReceiver, expected_actor_id: &str, - ) -> Option { + ) -> (protocol::ActorIntent, Option) { let message = tokio::time::timeout(Duration::from_secs(5), rx.recv()) .await .expect("timed out waiting for stop intent") @@ -678,14 +678,14 @@ mod moved_tests { ToEnvoyMessage::ActorIntent { actor_id, generation, - intent: protocol::ActorIntent::ActorIntentStop, + intent, error, } => { assert_eq!(actor_id, expected_actor_id); assert_eq!(generation, Some(3)); - error + (intent, error) } - _ => panic!("expected stop intent envoy message"), + _ => panic!("expected an intent envoy message"), } } @@ -698,7 +698,10 @@ mod moved_tests { ctx.destroy().expect("destroy should succeed after startup"); - let error = recv_stop_intent(&mut rx, "actor-destroy-intent").await; + // A deliberate destroy is the only thing that may report + // `ActorIntentStop`. + let (intent, error) = recv_stop_intent(&mut rx, "actor-destroy-intent").await; + assert!(matches!(intent, protocol::ActorIntent::ActorIntentStop)); assert_eq!(error, None); } @@ -712,13 +715,89 @@ mod moved_tests { 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; + // A crash is reported as a sleep intent so the engine resumes the + // actor instead of destroying it. The message still rides along and + // becomes `StopCode::Error` on the eventual `Stopped` event. + let (intent, error) = recv_stop_intent(&mut rx, "actor-stop-error-intent").await; + assert!(matches!(intent, protocol::ActorIntent::ActorIntentSleep)); assert_eq!( error.as_deref(), Some("child exited unexpectedly (exit status: 137)") ); } + async fn assert_no_further_intent(rx: &mut mpsc::UnboundedReceiver) { + // Paused time auto-advances once every task is idle, so this + // resolves as soon as the runtime has nothing left to run. + let extra = tokio::time::timeout(Duration::from_secs(1), rx.recv()).await; + assert!(extra.is_err(), "expected no further envoy message"); + } + + #[tokio::test(start_paused = true)] + async fn repeated_stop_with_error_sends_one_sleep_intent() { + let ctx = ActorContext::new_for_sleep_tests("actor-stop-error-repeated"); + let (handle, mut rx) = test_envoy_handle(); + ctx.configure_sleep_envoy(handle, Some(3)); + ctx.set_started(true); + + // Two reports for one crash is a real path: the run task reports + // eagerly and the event loop reports the same failure again at + // shutdown. The second report must not reach the envoy as a bare + // `ActorIntentStop`, which the engine answers by destroying the + // actor. + ctx.stop_with_error("first crash") + .expect("first stop_with_error should succeed after startup"); + ctx.stop_with_error("second crash") + .expect("second stop_with_error should succeed"); + + let (intent, error) = recv_stop_intent(&mut rx, "actor-stop-error-repeated").await; + assert!(matches!(intent, protocol::ActorIntent::ActorIntentSleep)); + assert_eq!(error.as_deref(), Some("second crash")); + assert_no_further_intent(&mut rx).await; + } + + #[tokio::test(start_paused = true)] + async fn destroy_after_stop_with_error_still_sends_stop_intent() { + let ctx = ActorContext::new_for_sleep_tests("actor-destroy-after-error"); + let (handle, mut rx) = test_envoy_handle(); + ctx.configure_sleep_envoy(handle, Some(3)); + ctx.set_started(true); + + // A destroy escalates an errored stop that has not been sent yet. + // It must report `ActorIntentStop` rather than inheriting the + // pending error and reporting a sleep, which would leave the actor + // alive. + ctx.stop_with_error("crash before destroy") + .expect("stop_with_error should succeed after startup"); + ctx.destroy() + .expect("destroy should succeed after an errored stop"); + + let (intent, error) = recv_stop_intent(&mut rx, "actor-destroy-after-error").await; + assert!(matches!(intent, protocol::ActorIntent::ActorIntentStop)); + assert_eq!(error, None); + } + + #[tokio::test(start_paused = true)] + async fn sleep_then_stop_with_error_reports_the_crash() { + let ctx = ActorContext::new_for_sleep_tests("actor-sleep-then-error"); + let (handle, mut rx) = test_envoy_handle(); + ctx.configure_sleep_envoy(handle, Some(3)); + ctx.set_started(true); + + // An in-flight sleep leaves the error slot empty, so upgrading it + // to a crash still has to reach the envoy. + ctx.sleep().expect("sleep should succeed after startup"); + let (intent, error) = recv_stop_intent(&mut rx, "actor-sleep-then-error").await; + assert!(matches!(intent, protocol::ActorIntent::ActorIntentSleep)); + assert_eq!(error, None); + + ctx.stop_with_error("crash during sleep") + .expect("stop_with_error should succeed while sleeping"); + let (intent, error) = recv_stop_intent(&mut rx, "actor-sleep-then-error").await; + assert!(matches!(intent, protocol::ActorIntent::ActorIntentSleep)); + assert_eq!(error.as_deref(), Some("crash during sleep")); + } + #[tokio::test(start_paused = true)] async fn stop_with_error_truncates_long_message() { let ctx = ActorContext::new_for_sleep_tests("actor-stop-error-truncated"); @@ -729,9 +808,9 @@ mod moved_tests { 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"); + let (intent, error) = recv_stop_intent(&mut rx, "actor-stop-error-truncated").await; + assert!(matches!(intent, protocol::ActorIntent::ActorIntentSleep)); + let error = error.expect("stop intent should carry the truncated message"); assert!( error.len() < 4096, "message must be capped: {}", diff --git a/rivetkit-rust/packages/rivetkit-core/tests/sqlite.rs b/rivetkit-rust/packages/rivetkit-core/tests/sqlite.rs index 8e2fe8bb9d..018e7da300 100644 --- a/rivetkit-rust/packages/rivetkit-core/tests/sqlite.rs +++ b/rivetkit-rust/packages/rivetkit-core/tests/sqlite.rs @@ -1551,7 +1551,9 @@ fn remote_head_fence_mismatch_stops_actor_once() { } => { assert_eq!(actor_id, "actor-a"); assert_eq!(generation, Some(7)); - assert!(matches!(intent, protocol::ActorIntent::ActorIntentStop)); + // A dead sqlite worker is a crash, not a deliberate destroy, so it + // is reported as a sleep intent carrying the error. + assert!(matches!(intent, protocol::ActorIntent::ActorIntentSleep)); assert!( error .expect("missing stop reason") diff --git a/rivetkit-rust/packages/rivetkit-core/tests/task.rs b/rivetkit-rust/packages/rivetkit-core/tests/task.rs index 587776925e..9e8a56b954 100644 --- a/rivetkit-rust/packages/rivetkit-core/tests/task.rs +++ b/rivetkit-rust/packages/rivetkit-core/tests/task.rs @@ -752,14 +752,14 @@ pub(crate) mod moved_tests { } fn detached_cleanup_after_failed_run_factory( - destroy_count: Arc, + cleanup_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 cleanup_count = cleanup_count.clone(); let run_returned_tx = run_returned_tx.clone(); let cleanup_tx = cleanup_tx.clone(); Box::pin(async move { @@ -771,9 +771,7 @@ pub(crate) mod moved_tests { reply.send(Ok(Vec::new())); } ActorEvent::RunGracefulCleanup { reason, reply } => { - if matches!(reason, ShutdownKind::Destroy) { - destroy_count.fetch_add(1, Ordering::SeqCst); - } + cleanup_count.fetch_add(1, Ordering::SeqCst); reply.send(Ok(())); if let Some(tx) = cleanup_tx .lock() @@ -4102,13 +4100,13 @@ pub(crate) mod moved_tests { "local", new_in_memory(), ); - let destroy_count = Arc::new(AtomicUsize::new(0)); + let cleanup_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(), + cleanup_count.clone(), run_returned_tx, cleanup_tx, ), @@ -4130,16 +4128,22 @@ pub(crate) mod moved_tests { 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. + // command still drives the grace hooks. assert_eq!(task.lifecycle, LifecycleState::Started); + // A crash reports a sleep intent, not a destroy, so the engine can + // resume the actor on a new generation. assert!( - ctx.is_destroy_requested(), + ctx.sleep_requested(), "failed run should request an errored stop" ); + assert!( + !ctx.is_destroy_requested(), + "a crash must not request a destroy" + ); let (stop_tx, stop_rx) = oneshot::channel(); task.handle_lifecycle(LifecycleCommand::Stop { - reason: ShutdownKind::Destroy, + reason: ShutdownKind::Sleep, reply: stop_tx, }) .await; @@ -4148,9 +4152,9 @@ pub(crate) mod moved_tests { .await .expect("grace cleanup should run after Stop") .expect("cleanup signal should send"), - ShutdownKind::Destroy + ShutdownKind::Sleep ); - assert_eq!(destroy_count.load(Ordering::SeqCst), 1); + assert_eq!(cleanup_count.load(Ordering::SeqCst), 1); timeout(Duration::from_secs(2), async { while ctx.core_dispatched_hook_count() != 0 { @@ -4169,7 +4173,7 @@ pub(crate) mod moved_tests { else { panic!("grace should transition to shutdown"); }; - assert_eq!(shutdown_reason, ShutdownKind::Destroy); + assert_eq!(shutdown_reason, ShutdownKind::Sleep); let result = task.run_shutdown(shutdown_reason).await; task.deliver_shutdown_reply(shutdown_reason, &result); task.transition_to(LifecycleState::Terminated); From db28891889c74568473b34eb3fc63d3f006bbe97 Mon Sep 17 00:00:00 2001 From: MasterPtato Date: Fri, 4 Sep 2026 12:42:00 -0700 Subject: [PATCH 3/4] fix(envoy-client): fail sent vfs sqlite requests on disconnect instead of stalling until expiry --- engine/sdks/rust/envoy-client/src/envoy.rs | 4 +- engine/sdks/rust/envoy-client/src/sqlite.rs | 31 ++++ .../tests/sqlite_vfs_disconnect_stall.rs | 139 ++++++------------ 3 files changed, 83 insertions(+), 91 deletions(-) diff --git a/engine/sdks/rust/envoy-client/src/envoy.rs b/engine/sdks/rust/envoy-client/src/envoy.rs index e910cf3533..dfa1c4e2d6 100644 --- a/engine/sdks/rust/envoy-client/src/envoy.rs +++ b/engine/sdks/rust/envoy-client/src/envoy.rs @@ -29,7 +29,8 @@ use crate::sqlite::{ RemoteSqliteRequest, RemoteSqliteRequestEntry, RemoteSqliteResponseEnvelope, SqliteRequest, SqliteRequestEntry, SqliteResponse, cleanup_old_remote_sqlite_requests, cleanup_old_sqlite_requests, fail_remote_sqlite_requests_with_shutdown, - fail_sent_remote_sqlite_requests_with_indeterminate_result, fail_sqlite_requests_with_shutdown, + fail_sent_remote_sqlite_requests_with_indeterminate_result, + fail_sent_sqlite_requests_with_indeterminate_result, fail_sqlite_requests_with_shutdown, handle_remote_sqlite_exec_response, handle_remote_sqlite_execute_batch_response, handle_remote_sqlite_execute_response, handle_remote_sqlite_request, handle_sqlite_commit_finalize_response, handle_sqlite_commit_response, @@ -473,6 +474,7 @@ async fn envoy_loop( } } fail_sent_remote_sqlite_requests_with_indeterminate_result(&mut ctx); + fail_sent_sqlite_requests_with_indeterminate_result(&mut ctx); lost_timeout = handle_conn_close(&ctx, lost_timeout); if evict { observe_envoy_loop_iteration(branch, iter_start); diff --git a/engine/sdks/rust/envoy-client/src/sqlite.rs b/engine/sdks/rust/envoy-client/src/sqlite.rs index e8c89f1695..254e53e6a7 100644 --- a/engine/sdks/rust/envoy-client/src/sqlite.rs +++ b/engine/sdks/rust/envoy-client/src/sqlite.rs @@ -518,6 +518,37 @@ pub fn fail_sqlite_requests_with_shutdown(ctx: &mut EnvoyContext) { } } +/// Fails every sent-but-unanswered VFS request after the websocket drops. +/// +/// Without this a sent request survives `ConnClose` and is only freed by the +/// `KV_EXPIRE_MS` cleanup tick. The VFS callback that is parked on the response +/// blocks the SQLite worker thread for that whole window, which stalls every +/// subsequent SQLite call on the actor. Unsent requests are left in place for +/// `process_unsent_sqlite_requests` to resend on reconnect. +pub fn fail_sent_sqlite_requests_with_indeterminate_result(ctx: &mut EnvoyContext) { + let request_ids: Vec = ctx + .sqlite_requests + .iter() + .filter(|(_, request)| request.sent) + .map(|(request_id, _)| *request_id) + .collect(); + + for request_id in request_ids { + if let Some(request) = ctx.sqlite_requests.remove(&request_id) { + METRICS.sqlite_requests_inflight.dec(); + let operation = request.request.kind(); + tracing::warn!( + request_id, + operation, + "sqlite response lost after websocket disconnect" + ); + let _ = request.response_tx.send(Err(anyhow::anyhow!( + RemoteSqliteIndeterminateResultError { operation } + ))); + } + } +} + pub fn fail_remote_sqlite_requests_with_shutdown(ctx: &mut EnvoyContext) { for (_id, request) in ctx.remote_sqlite_requests.drain() { METRICS.remote_sqlite_requests_inflight.dec(); diff --git a/engine/sdks/rust/envoy-client/tests/sqlite_vfs_disconnect_stall.rs b/engine/sdks/rust/envoy-client/tests/sqlite_vfs_disconnect_stall.rs index b45f8051c2..73b33ffe71 100644 --- a/engine/sdks/rust/envoy-client/tests/sqlite_vfs_disconnect_stall.rs +++ b/engine/sdks/rust/envoy-client/tests/sqlite_vfs_disconnect_stall.rs @@ -1,39 +1,31 @@ -//! Reproduction test for the actor-side VFS-SQLite-stall-on-disconnect bug. +//! Regression test for the actor-side VFS-SQLite-stall-on-disconnect bug. //! -//! Hypothesis under test (H6): +//! The bug: when the actor-engine WebSocket disconnected with sent-but-unanswered VFS +//! SQLite requests (`get_pages` / `commit`) in flight, the actor side did not fail those +//! requests. `ConnClose` only ran the *remote* exec/execute variant +//! (`fail_sent_remote_sqlite_requests_with_indeterminate_result`), so VFS requests +//! survived it and were only freed by the periodic cleanup (every 15s, 30s timeout). +//! During that window the SQLite VFS callback stayed parked in +//! `runtime.block_on(transport.get_pages(...))`, which blocked the SQLite worker thread +//! and every subsequent SQLite call on that actor. //! -//! When the actor-engine WebSocket disconnects with sent-but-unanswered VFS SQLite -//! requests (`get_pages` / `commit`) in flight, the actor side does NOT immediately -//! fail those requests. Instead, they sit in `ctx.sqlite_requests` until the periodic -//! cleanup (every 15s, 30s timeout) expires them. During that ~30s window, the SQLite -//! VFS callback is parked in `runtime.block_on(transport.get_pages(...))`, which blocks -//! the SQLite worker thread, which blocks all subsequent SQLite calls on that actor. +//! `fail_sent_sqlite_requests_with_indeterminate_result` closes that gap, and `ConnClose` +//! now runs both variants. //! -//! The disconnect handler is at -//! `engine/sdks/rust/envoy-client/src/envoy.rs:363-367` and only calls -//! `fail_sent_remote_sqlite_requests_with_indeterminate_result` — the *remote* exec/execute -//! variant. There is no symmetric `fail_sent_sqlite_requests_*` for VFS requests, so they -//! survive `ConnClose` and only get dropped by the 30s cleanup tick. +//! This test invokes the same code paths `ConnClose` runs on a synthetic `EnvoyContext` +//! holding one sent VFS get_pages request and one sent remote-execute request, and +//! asserts both resolve immediately with `RemoteSqliteIndeterminateResultError`. //! -//! This test does not change any production code. It directly invokes the same code paths -//! that `ConnClose` runs (`fail_sent_remote_sqlite_requests_with_indeterminate_result`, -//! `handle_conn_close`) on a synthetic `EnvoyContext` containing one sent VFS get_pages -//! request and one sent remote-execute request, then measures how long until each oneshot -//! response future resolves. -//! -//! Expected output: -//! - Remote exec/execute oneshot resolves IMMEDIATELY with -//! `RemoteSqliteIndeterminateResultError` (the existing disconnect path). -//! - VFS get_pages oneshot stays pending until `cleanup_old_sqlite_requests` runs and -//! finds the entry older than `KV_EXPIRE_MS`. We accelerate that by mutating the -//! entry's `timestamp` to a synthetic past value, demonstrating that the only escape -//! hatch for a sent VFS request is the timestamp-based expiry. With unmodified -//! timestamps the request would sit for 30s. +//! A lost `commit` reply stays genuinely ambiguous: the engine may have applied it. Failing +//! fast bounds the stall but does not resolve that, and the retry is still adjudicated by +//! the head fence (see `lost_commit_response_fails_later_on_head_fence_mismatch` in +//! `rivet-depot-client`). Reconciling instead would need writer identity in `DBHead`, which +//! it does not carry. //! //! Negative-control test (`unsent_vfs_request_quickly_resubmitted_after_reconnect`) //! flips one variable: the VFS request is queued *before* the WS goes up, so it never -//! reaches `sent=true`. The bug must NOT manifest in this case — `process_unsent_sqlite_requests` -//! must successfully re-send it on reconnect. +//! reaches `sent=true`. Unsent requests must be left alone for +//! `process_unsent_sqlite_requests` to re-send on reconnect. use std::collections::HashMap; use std::sync::Arc; @@ -46,10 +38,10 @@ use rivet_envoy_client::config::{ use rivet_envoy_client::context::{SharedContext, WsTxMessage}; use rivet_envoy_client::envoy::EnvoyContext; use rivet_envoy_client::handle::EnvoyHandle; -use rivet_envoy_client::kv::KV_EXPIRE_MS; use rivet_envoy_client::sqlite::{ RemoteSqliteRequest, SqliteRequest, cleanup_old_sqlite_requests, - fail_sent_remote_sqlite_requests_with_indeterminate_result, handle_remote_sqlite_request, + fail_sent_remote_sqlite_requests_with_indeterminate_result, + fail_sent_sqlite_requests_with_indeterminate_result, handle_remote_sqlite_request, handle_sqlite_request, process_unsent_sqlite_requests, }; use rivet_envoy_client::utils::{BufferMap, RemoteSqliteIndeterminateResultError}; @@ -211,7 +203,7 @@ fn init_tracing() { /// VFS oneshot resolves with the `sqlite request timed out` error. This is the same /// error path that fires in production at ~30s after the disconnect. #[tokio::test] -async fn sent_vfs_request_stalls_on_disconnect() { +async fn sent_vfs_request_fails_immediately_on_disconnect() { init_tracing(); let mut ctx = new_envoy_context(); @@ -265,9 +257,10 @@ async fn sent_vfs_request_stalls_on_disconnect() { tracing::info!("ws disconnected; running production ConnClose handler"); let before_handler = std::time::Instant::now(); fail_sent_remote_sqlite_requests_with_indeterminate_result(&mut ctx); + fail_sent_sqlite_requests_with_indeterminate_result(&mut ctx); tracing::info!( elapsed_us = before_handler.elapsed().as_micros() as u64, - "ran fail_sent_remote_sqlite_requests_with_indeterminate_result" + "ran both ConnClose sqlite failure handlers" ); // 1) Remote oneshot resolves NOW with indeterminate result. (Existing behavior.) @@ -288,71 +281,37 @@ async fn sent_vfs_request_stalls_on_disconnect() { "REMOTE: fails immediately on disconnect (existing correct behavior)" ); - // 2) VFS oneshot is STILL pending. The disconnect handler did not touch it. - // We give it a real 500ms wall-clock window to confirm it does not resolve. - let stall_probe = Duration::from_millis(500); - let stall_start = std::time::Instant::now(); - let vfs_immediate = tokio::time::timeout(stall_probe, &mut vfs_rx).await; - let stall_elapsed = stall_start.elapsed(); - assert!( - vfs_immediate.is_err(), - "BUG: VFS sqlite request is still pending after disconnect handler ran. \ - In production, this oneshot is what `runtime.block_on(transport.get_pages(...))` \ - is parked on inside the SQLite VFS callback." - ); - assert!( - ctx.sqlite_requests.contains_key(&0), - "VFS request must still be in the pending map" - ); - tracing::warn!( - stall_elapsed_ms = stall_elapsed.as_millis() as u64, - "BUG REPRODUCED: VFS sqlite_requests entry survives ConnClose; still pending after \ - {} ms. Only the 15s/30s cleanup tick can free it.", - stall_elapsed.as_millis() - ); - - // 3) Demonstrate that the timestamp-based cleanup is the only escape hatch. Backdate - // the entry's timestamp by KV_EXPIRE_MS+1ms and run the same `cleanup_old_sqlite_requests` - // function the periodic tick runs. - let backdate = Duration::from_millis(KV_EXPIRE_MS + 1); - let entry = ctx - .sqlite_requests - .get_mut(&0) - .expect("vfs request still pending"); - // Walk the Instant backward. - entry.timestamp = std::time::Instant::now() - .checked_sub(backdate) - .expect("instant subtraction"); - tracing::info!( - backdate_ms = backdate.as_millis() as u64, - "backdated timestamp to simulate KV_EXPIRE_MS elapsed" - ); - cleanup_old_sqlite_requests(&mut ctx); - + // 2) VFS oneshot resolves NOW too, with the same indeterminate error. + let vfs_resolve_start = std::time::Instant::now(); let vfs_result = tokio::time::timeout(Duration::from_millis(50), &mut vfs_rx) .await - .expect("VFS oneshot must resolve after cleanup") - .expect("VFS oneshot must complete"); + .expect("vfs oneshot must resolve immediately after disconnect") + .expect("vfs oneshot must complete"); + let vfs_elapsed = vfs_resolve_start.elapsed(); let vfs_err = match vfs_result { - Ok(_) => panic!("VFS request must fail with timed-out error"), + Ok(_) => panic!("vfs get_pages must fail with indeterminate"), Err(err) => err, }; - let msg = format!("{vfs_err:#}"); - assert!( - msg.contains("sqlite request timed out"), - "unexpected VFS error: {msg}" - ); + let vfs_indet = vfs_err + .downcast_ref::() + .expect("vfs get_pages must fail with RemoteSqliteIndeterminateResultError"); + assert_eq!(vfs_indet.operation, "get_pages"); assert!( !ctx.sqlite_requests.contains_key(&0), - "VFS request must be removed by cleanup" + "vfs request must be removed from the pending map by the disconnect handler" ); - tracing::warn!( - error = %msg, - stall_window_ms = KV_EXPIRE_MS, - "In production this is the only path that frees the VFS request. \ - During the entire {KV_EXPIRE_MS}ms window the SQLite VFS callback is blocked \ - on `runtime.block_on(transport.get_pages(...))`, blocking the SQLite worker \ - thread and any subsequent SQLite call on this actor.", + tracing::info!( + elapsed_us = vfs_elapsed.as_micros() as u64, + operation = %vfs_indet.operation, + "VFS: fails immediately on disconnect, matching the remote path" + ); + + // 3) The expiry tick is no longer the escape hatch for a sent request: there is + // nothing left for it to reap. + cleanup_old_sqlite_requests(&mut ctx); + assert!( + ctx.sqlite_requests.is_empty(), + "no sent vfs request should survive to the expiry tick" ); } From 81a708408587b60f872af317c1097b54ef7c2a01 Mon Sep 17 00:00:00 2001 From: MasterPtato Date: Fri, 4 Sep 2026 14:41:16 -0700 Subject: [PATCH 4/4] fix(rivetkit): surface native runtime load failures instead of masking them with wasm --- engine/artifacts/openapi.json | 2 +- .../packages/rivetkit/src/registry/native.ts | 28 ++++++- .../rivetkit/src/registry/runtime.test.ts | 74 +++++++++++++++++++ 3 files changed, 102 insertions(+), 2 deletions(-) diff --git a/engine/artifacts/openapi.json b/engine/artifacts/openapi.json index eb2e00c349..30d5ec455a 100644 --- a/engine/artifacts/openapi.json +++ b/engine/artifacts/openapi.json @@ -11,7 +11,7 @@ "name": "Apache-2.0", "identifier": "Apache-2.0" }, - "version": "2.4.0" + "version": "2.3.14" }, "paths": { "/actors": { diff --git a/rivetkit-typescript/packages/rivetkit/src/registry/native.ts b/rivetkit-typescript/packages/rivetkit/src/registry/native.ts index cee301a9e2..4d3a55b45e 100644 --- a/rivetkit-typescript/packages/rivetkit/src/registry/native.ts +++ b/rivetkit-typescript/packages/rivetkit/src/registry/native.ts @@ -224,10 +224,36 @@ export async function loadAutoRuntime( return (await loaders.loadWasm(config.wasm)).runtime; } + let nativeError: unknown; try { return (await loaders.loadNative()).runtime; - } catch { + } catch (error) { + nativeError = error; + // Native is the expected runtime on a node-like host, so this is the + // actionable error even when the wasm fallback goes on to succeed. + // Discarding it hides causes such as a platform binding that npm + // silently skipped, which then resurfaces as an unrelated wasm error. + logger().warn({ + msg: "native runtime failed to load; falling back to wasm", + error: stringifyError(error), + }); + } + + try { return (await loaders.loadWasm(config.wasm)).runtime; + } catch (wasmError) { + // Report both, native first. The wasm failure on a node-like host is + // usually just its loader fetching over `file://`, which says nothing + // about why native was unavailable. + throw new RivetError( + "config", + "runtime_unavailable", + `RivetKit could not load a core runtime. Native runtime: ${stringifyError(nativeError)} Wasm runtime: ${stringifyError(wasmError)}`, + { + public: true, + statusCode: 500, + }, + ); } } diff --git a/rivetkit-typescript/packages/rivetkit/src/registry/runtime.test.ts b/rivetkit-typescript/packages/rivetkit/src/registry/runtime.test.ts index 1921809190..2d11a51818 100644 --- a/rivetkit-typescript/packages/rivetkit/src/registry/runtime.test.ts +++ b/rivetkit-typescript/packages/rivetkit/src/registry/runtime.test.ts @@ -1,5 +1,8 @@ import { describe, expect, test } from "vitest"; +import type { RegistryConfig } from "./config"; +import { loadAutoRuntime, type RuntimeLoaders } from "./native"; import { + type CoreRuntime, normalizeRuntimeSqlExecuteResult, type RuntimeSqlBindParam, type RuntimeSqlBindParams, @@ -48,3 +51,74 @@ describe("runtime SQL boundary", () => { expect(normalizeRuntimeSqlExecuteResult(base)).toEqual(base); }); }); + +describe("loadAutoRuntime failure reporting", () => { + const wasmRuntime = { kind: "wasm" } as unknown as CoreRuntime; + const nativeRuntime = { kind: "napi" } as unknown as CoreRuntime; + const config = {} as RegistryConfig; + + function loaders(overrides: Partial): RuntimeLoaders { + return { + detectHost: () => "node-like", + loadNative: async () => ({ runtime: nativeRuntime }) as never, + loadWasm: async () => ({ runtime: wasmRuntime }) as never, + ...overrides, + }; + } + + test("prefers native when it loads", async () => { + const runtime = await loadAutoRuntime(config, loaders({})); + expect(runtime).toBe(nativeRuntime); + }); + + test("falls back to wasm when native fails", async () => { + const runtime = await loadAutoRuntime( + config, + loaders({ + loadNative: async () => { + throw new Error("missing platform binding"); + }, + }), + ); + expect(runtime).toBe(wasmRuntime); + }); + + test("reports the native cause when both runtimes fail", async () => { + // The native failure is the actionable one. Before this was reported, + // a skipped platform binding surfaced only as the wasm loader's + // unrelated `file://` fetch error. + const promise = loadAutoRuntime( + config, + loaders({ + loadNative: async () => { + throw new Error( + "Cannot find module '@rivetkit/rivetkit-napi-linux-x64-musl'", + ); + }, + loadWasm: async () => { + throw new Error("fetch failed"); + }, + }), + ); + await expect(promise).rejects.toThrow(/rivetkit-napi-linux-x64-musl/); + await expect(promise).rejects.toThrow(/fetch failed/); + }); + + test("uses wasm directly on an edge-like host without touching native", async () => { + let nativeCalls = 0; + const runtime = await loadAutoRuntime( + config, + loaders({ + detectHost: () => "edge-like", + loadNative: async () => { + nativeCalls += 1; + throw new Error( + "native must not be attempted on edge hosts", + ); + }, + }), + ); + expect(runtime).toBe(wasmRuntime); + expect(nativeCalls).toBe(0); + }); +});