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/commands.rs

@@ -18,11 +18,10 @@ pub async fn handle_commands(ctx: &mut EnvoyContext, commands: Vec<protocol::Com
 		);
 	}
 
-	// Collect actors with a stop in the raw batch before dedup, so a replayed
-	// (skipped) stop is still re-acked instead of being replayed forever.
-	let stopped_actors: Vec<(String, u32)> = 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<protocol::Com
 		}
 	}
 
-	// Ack stops immediately since their actors are removed before the periodic
-	// tick. Scope to just the stopped actors instead of a full-state ack, and do
-	// not clear dedup; the tick handles full re-acks, recovery, and clearing.
-	if !stopped_actors.is_empty() {
-		send_stop_command_acks(ctx, &stopped_actors).await;
+	// Ack the whole batch immediately. Anything left unacked stays in the
+	// engine's `ActorCommandKey` subspace, and `envoy_conn_prepare` re-streams
+	// that subspace on every reconnect, so a start that waits for the periodic
+	// tick can be replayed for up to `ACK_COMMANDS_INTERVAL_MS` and resurrect a
+	// stopped actor or replace a live one. Scope to just this batch's actors
+	// instead of a full-state ack, and do not clear dedup; the tick handles
+	// full re-acks, recovery, and clearing.
+	if !batch_actors.is_empty() {
+		send_batch_command_acks(ctx, &batch_actors).await;
 	}
 }
 
 /// Ack only the given actors' latest processed command index. Used for the
-/// immediate stop ack. Does not clear dedup (see the race note in
+/// immediate post-batch ack. Does not clear dedup (see the race note in
 /// `send_command_ack`); a failed send is retried by the replayed batch or tick.
-async fn send_stop_command_acks(ctx: &EnvoyContext, actors: &[(String, u32)]) {
+async fn send_batch_command_acks(ctx: &EnvoyContext, actors: &[(String, u32)]) {
 	let mut highest: HashMap<(String, u32), i64> = HashMap::new();
 	for key in actors {
 		if let Some(&index) = ctx.processed_command_idx.get(key) {

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

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<protocol::ActorCheckpoint> {
 	}
 }
 
+#[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();

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