From 64c5c46452ce67f33554b9dce4db8170349b18d1 Mon Sep 17 00:00:00 2001 From: MasterPtato Date: Fri, 4 Sep 2026 12:17:00 -0700 Subject: [PATCH] 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);