fix(rivetkit-core): report actor crashes as a sleep intent instead of a stop intent

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));
 						}
 					}
 				}

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<u32>) {
+	/// 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<u32>, error: Option<String>) {
 		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<u32>, error: Option<String>) {
+	/// 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<u32>) {
 		let _ = crate::envoy::send_to_envoy_tx(
 			&self.shared,
 			ToEnvoyMessage::ActorIntent {
 				actor_id,
 				generation,
 				intent: protocol::ActorIntent::ActorIntentStop,
-				error,
+				error: None,
 			},
 		);
 	}

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<String>) -> Result<()> {
 		self.request_stop(Some(truncate_stop_error_message(message.into())))
 	}
@@ -569,41 +571,68 @@ 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 {
 				return Ok(());
 			}
 		}
 
-		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();

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

@@ -47,9 +47,10 @@ pub(crate) struct SleepState {
 	pub(super) envoy_handle: Mutex<Option<EnvoyHandle>>,
 	pub(super) generation: Mutex<Option<u32>>,
 	pub(super) http_request_counter: Mutex<Option<Arc<AsyncCounter>>>,
-	// 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<Option<String>>,
+	// 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<Option<String>>,
 	#[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,21 +164,30 @@ 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
 			.destroy_request_count
 			.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),
 		}
 	}
 

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

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

@@ -669,7 +669,7 @@ mod moved_tests {
 		async fn recv_stop_intent(
 			rx: &mut mpsc::UnboundedReceiver<ToEnvoyMessage>,
 			expected_actor_id: &str,
-		) -> Option<String> {
+		) -> (protocol::ActorIntent, Option<String>) {
 			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<ToEnvoyMessage>) {
+			// 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: {}",