fix(rivetkit): stop lost actor generations immediately

CLAUDE.md

@@ -262,6 +262,7 @@ When the user asks to track something in a note, store it in `~/.agents/notes/`
 - In `rivetkit-core` `ActorTask::run`, bind inbox `recv()` calls as raw `Option`s and log the closed channel before terminating. `Some(...) = recv()` plus `else => break` hides which inbox died.
 - In `rivetkit-typescript/packages/rivetkit/src/common/utils.ts::deconstructError`, only passthrough canonical structured errors (`instanceof RivetError` or tagged `__type: "RivetError"` with full fields). Plain-object lookalikes must still be classified and sanitized.
 - Actor-owned lifecycle / dispatch / lifecycle-event inbox producers use `try_reserve` helpers and return `actor.overloaded`. Do not await bounded `mpsc::Sender::send`.
+- A Lost actor generation must stop immediately: no user sleep hooks, no final save, and no storage writes after the lost signal fires. Never treat Lost as a graceful sleep.
 
 ## Performance
 

docs-internal/engine/rivetkit-core-internals.md

@@ -66,7 +66,11 @@ Legacy traces under `[7, 1, ...]` are intentionally skipped and are not copied t
 ## Registry + dispatch
 
 - Registry startup builds configured `ActorContext`s with `ActorContext::build(...)` so state, queue, and connection managers inherit the actor config before lifecycle startup runs. `ActorContext::build(...)` must seed owned queue, connection, and sleep config storage from its `ActorConfig`; do not initialize those fields with `ActorConfig::default()`.
-- Registry actor task handles live in one `actor_instances: SccHashMap<String, ActorInstanceState>`. Use `entry_async` for Active/Stopping transitions.
+- Registry records (`actor_instances`, `starting_instances`, `pending_stops`) are keyed by `(actor_id, generation)`, so a lost generation that is still tearing down stays reachable after the next generation starts. `current_generations` holds the newest registered generation per actor id, and dispatch routes only to it. Use `entry_async` for Active/Stopping transitions.
+- A stop for a starting generation parks under that generation's starting entry and is consumed by its own startup. A stop for an older generation with no record completes immediately, and registering a newer generation completes stops parked for older ones.
+- A starting generation marks older generations of the same actor on the runner, starting or registered, lost and waits up to 10s for them to finish (`ActorContext::wait_for_generation_finished`). A generation finishes once every `GenerationHold` is dropped: the actor task holds one, and a lost generation's storage cleanup holds another until a strict SQLite close returns. Runtime wrappers that spawn per-generation tasks outside the actor task take a hold for each and stop them on the lost signal. If one does not finish, the new start fails instead of overlapping it. The engine only starts a generation after giving up on the older ones.
+- A generation lost during startup stays registered as starting until its task finishes aborting, so envoy-client does not report it stopped early.
+- A stopped generation's record is removed only once the generation has finished, and a generation refuses to start if a newer generation of the same actor is already starting or registered, since start callbacks can arrive out of order.
 - `RegistryDispatcher::handle_fetch` owns framework HTTP routes `/metadata`, `/health`, `/inspector/*`, `/action/*`, and `/queue/*`. Prometheus metrics are process-wide and are never served by core itself; `metrics::render_prometheus_metrics()` only backs the NAPI `metrics()` binding and `Registry::prometheus_metrics()` so hosts can mount the rendered response on their own router. TypeScript NAPI callbacks keep action/queue schema validation and queue `canPublish`.
 - Raw `onRequest` HTTP fetches bypass `maxIncomingMessageSize` / `maxOutgoingMessageSize`. Those message-size guards apply only to `/action/*` and `/queue/*` framework routes, not unmatched user `onRequest` paths.
 - Framework HTTP error payloads omit absent `metadata` for JSON/CBOR responses so missing metadata stays `undefined`. Only explicit metadata `null` serializes as `null`.

docs-internal/engine/sleep-sequence.md

@@ -49,6 +49,15 @@ When the grace deadline elapses before `can_finalize_sleep()` returns true:
 - The NAPI `RunGracefulCleanup` task observes `shutdown_deadline_token()` via `tokio::select!` and aborts its in-flight `onSleep` / `onDestroy` call so SQLite and KV cleanup in `teardown_sleep_state` do not race against mid-commit user work.
 - Foreign-runtime adapters that run user cleanup callbacks must observe the shutdown deadline token the same way.
 
+## Lost generations
+
+- `StopActor { reason: Lost }` or an envoy-side lost timeout means the engine has given up on the generation and may already be running the next one. envoy-client cancels a per-generation lost `CancellationToken` synchronously, before the stop is queued, and passes it to core through `on_actor_start_with_lost_signal`.
+- Core adopts the token as the context lost signal (`ActorContext::is_lost`). Once it fires, `SqliteDb` rejects every statement except `ROLLBACK` and `LegacyActorKv` rejects writes, so state, KV, queue, and workflow writes from the lost generation are revoked without waiting on the task.
+- `ActorTask` observes the token in `run_live`, in graceful shutdown, and in serialization waits, and jumps to `abort_lost()`. That path runs no user sleep hooks, skips the final save and alarm sync, dispatches runtime cleanup once, and bounds every wait by `LOST_SHUTDOWN_GRACE_PERIOD`. Cleanup that has not finished stays owned by its retained task.
+- The final `ActorStateStopped` for a lost generation is always `StopCode::Error` with "envoy connection lost", including when Lost escalates a pending graceful stop. `StopCode::Ok` would make the engine destroy the actor.
+- Already-submitted SQLite commits cannot be recalled. Depot generation checks are the backstop for those.
+- A lost generation starts no new user callback. Every runtime callback passes through `refuseCallbacksAfterLost` in `registry/native.ts` when JS enters it, except the `onSleep` and `onDestroy` cleanup wrappers, which release runtime state and skip the user hook themselves. Callbacks the framework starts on its own (workflow steps, run, inspector workflow calls, deferred `onStateChange`, WebSocket open and message listeners, database provider hooks) check the lost state first. Continuations inside a user call that is already running, and abort or close listeners, are not blocked.
+
 ## Guarding lifecycle requests
 
 - `ctx.sleep()` and `ctx.destroy()` return `Result<()>`. They fail with `actor/starting` if called before startup completes and `actor/stopping` if the request flag has already been swapped to true for this generation. An atomic `swap(true, ...)` on `sleep_requested` / `destroy_requested` enforces single-shot request semantics per generation.

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

@@ -169,6 +169,13 @@ pub(crate) struct ActorContextInner {
 	destroy_requested: AtomicBool,
 	destroy_completed: AtomicBool,
 	destroy_completion_notify: Notify,
+	/// Outstanding holds on this generation. The actor task holds one, and a lost generation's
+	/// storage cleanup holds another until SQLite is released. The registry waits for zero
+	/// before starting a newer generation.
+	generation_holds: AtomicUsize,
+	generation_finished_notify: Notify,
+	// Forced-sync: read by synchronous accessors and bound before the actor task starts.
+	lost: Mutex<CancellationToken>,
 	abort_signal: Mutex<CancellationToken>,
 	shutdown_deadline: CancellationToken,
 	// Forced-sync: runtime wiring slots are configured through synchronous
@@ -429,6 +436,9 @@ impl ActorContext {
 		let max_schedules = config.max_schedules;
 		let abort_signal = CancellationToken::new();
 		let shutdown_deadline = CancellationToken::new();
+		let lost = CancellationToken::new();
+		sql.set_lost_signal(lost.clone());
+		legacy_kv.set_lost_signal(lost.clone());
 		let sleep = SleepState::new(config.clone());
 		let user_kv = ActorKv { sql: sql.clone() };
 		let inner = Arc::new(ActorContextInner {
@@ -508,6 +518,9 @@ impl ActorContext {
 			destroy_requested: AtomicBool::new(false),
 			destroy_completed: AtomicBool::new(false),
 			destroy_completion_notify: Notify::new(),
+			generation_holds: AtomicUsize::new(0),
+			generation_finished_notify: Notify::new(),
+			lost: Mutex::new(lost),
 			abort_signal: Mutex::new(abort_signal),
 			shutdown_deadline,
 			inspector: RwLock::new(None),
@@ -774,6 +787,31 @@ impl ActorContext {
 		Ok(())
 	}
 
+	/// Adopts the lost signal envoy-client created for this generation. envoy-client cancels it
+	/// synchronously when the engine or the envoy declares the generation lost, which revokes
+	/// SQLite writes immediately and wakes the actor task into its abort path.
+	pub(crate) fn configure_lost_signal(&self, lost: CancellationToken) {
+		self.sql().set_lost_signal(lost.clone());
+		self.0.legacy_kv.set_lost_signal(lost.clone());
+		*self.0.lost.lock() = lost;
+	}
+
+	/// Returns this generation's lost signal.
+	pub fn lost_signal(&self) -> CancellationToken {
+		self.0.lost.lock().clone()
+	}
+
+	/// Declares this generation lost. Idempotent.
+	pub(crate) fn mark_lost(&self) {
+		self.0.lost.lock().cancel();
+	}
+
+	/// Returns whether this generation was declared lost. Foreign-runtime shutdown hooks must
+	/// skip user callbacks and state saves when this is set, and only release runtime resources.
+	pub fn is_lost(&self) -> bool {
+		self.0.lost.lock().is_cancelled()
+	}
+
 	pub fn mark_destroy_requested(&self) {
 		self.cancel_sleep_timer();
 		self.flush_on_shutdown();
@@ -1531,6 +1569,31 @@ impl ActorContext {
 		self.destroy_requested()
 	}
 
+	/// Keeps this generation from counting as finished until the returned hold is dropped.
+	/// Runtime wrappers hold it for tasks they spawn outside the actor task.
+	#[doc(hidden)]
+	pub fn hold_generation(&self) -> GenerationHold {
+		self.0.generation_holds.fetch_add(1, Ordering::SeqCst);
+		GenerationHold(self.clone())
+	}
+
+	pub(crate) fn is_generation_finished(&self) -> bool {
+		self.0.generation_holds.load(Ordering::SeqCst) == 0
+	}
+
+	pub(crate) async fn wait_for_generation_finished(&self) {
+		loop {
+			let notified = self.0.generation_finished_notify.notified();
+			tokio::pin!(notified);
+			// Register before checking so a finish between the check and the await is not missed.
+			notified.as_mut().enable();
+			if self.is_generation_finished() {
+				return;
+			}
+			notified.await;
+		}
+	}
+
 	pub(crate) async fn wait_for_destroy_completion(&self) {
 		if self.0.destroy_completed.load(Ordering::SeqCst) {
 			return;
@@ -2172,6 +2235,17 @@ impl std::fmt::Debug for ActorContext {
 }
 
 // Test shim keeps moved tests in crate-root tests/ with private-module access.
+/// A hold on an actor generation. See [`ActorContext::hold_generation`].
+pub struct GenerationHold(ActorContext);
+
+impl Drop for GenerationHold {
+	fn drop(&mut self) {
+		if self.0.0.generation_holds.fetch_sub(1, Ordering::SeqCst) == 1 {
+			self.0.0.generation_finished_notify.notify_waiters();
+		}
+	}
+}
+
 #[cfg(test)]
 #[path = "../../tests/context.rs"]
 pub(crate) mod tests;

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

@@ -1,13 +1,15 @@
 #[cfg(test)]
 use std::collections::BTreeMap;
-#[cfg(test)]
 use std::sync::Arc;
 
-use anyhow::Result;
+use anyhow::{Context, Result};
+use parking_lot::Mutex;
 #[cfg(test)]
-use parking_lot::{Mutex, RwLock};
+use parking_lot::RwLock;
 use rivet_envoy_client::handle::EnvoyHandle;
+use tokio_util::sync::CancellationToken;
 
+use crate::error::SqliteRuntimeError;
 use crate::types::ListOpts;
 
 /// Narrow access to the actor's pre-SQLite KV namespace.
@@ -18,6 +20,8 @@ use crate::types::ListOpts;
 #[derive(Clone)]
 pub(crate) struct LegacyActorKv {
 	backend: LegacyActorKvBackend,
+	/// The owning generation's lost signal. Writes are revoked once it fires.
+	lost: Arc<Mutex<Option<CancellationToken>>>,
 }
 
 #[derive(Clone)]
@@ -45,6 +49,7 @@ impl LegacyActorKv {
 				handle,
 				actor_id: actor_id.into(),
 			},
+			lost: Default::default(),
 		}
 	}
 
@@ -57,9 +62,27 @@ impl LegacyActorKv {
 				list_limit_cap: Mutex::new(None),
 				range_start_inclusive: Mutex::new(true),
 			})),
+			lost: Default::default(),
 		}
 	}
 
+	pub(crate) fn set_lost_signal(&self, lost: CancellationToken) {
+		*self.lost.lock() = Some(lost);
+	}
+
+	fn ensure_not_lost(&self) -> Result<()> {
+		let lost = self
+			.lost
+			.lock()
+			.as_ref()
+			.is_some_and(CancellationToken::is_cancelled);
+		if lost {
+			return Err(SqliteRuntimeError::Closed.build())
+				.context("actor generation was declared lost; storage writes are revoked");
+		}
+		Ok(())
+	}
+
 	#[cfg(test)]
 	pub(crate) async fn get(&self, key: &[u8]) -> Result<Option<Vec<u8>>> {
 		Ok(self.batch_get(&[key]).await?.pop().flatten())
@@ -88,17 +111,20 @@ impl LegacyActorKv {
 	}
 
 	pub(crate) async fn batch_put(&self, entries: &[(&[u8], &[u8])]) -> Result<()> {
+		self.ensure_not_lost()?;
 		match &self.backend {
 			LegacyActorKvBackend::Envoy { handle, actor_id } => {
-				handle
-					.kv_put(
-						actor_id.clone(),
-						entries
-							.iter()
-							.map(|(key, value)| (key.to_vec(), value.to_vec()))
-							.collect(),
-					)
-					.await
+				let entries = entries
+					.iter()
+					.map(|(key, value)| (key.to_vec(), value.to_vec()))
+					.collect();
+				// The write can sit in envoy's queue, so envoy drops it if the generation is
+				// declared lost before it is sent.
+				let lost = self.lost.lock().clone();
+				match lost {
+					Some(lost) => handle.kv_put_fenced(actor_id.clone(), entries, lost).await,
+					None => handle.kv_put(actor_id.clone(), entries).await,
+				}
 			}
 			#[cfg(test)]
 			LegacyActorKvBackend::InMemory(store) => {

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

@@ -294,6 +294,18 @@ impl ActorContext {
 			&& self.pending_disconnect_count() == 0
 	}
 
+	/// Waits until every core-dispatched hook has replied, or `deadline` passes. Waits on the hook
+	/// counter itself. The sleep activity signal only fires when its dirty flag flips, which the
+	/// lost abort path never acknowledges, so it can miss a hook completing.
+	pub(crate) async fn wait_for_core_dispatched_hooks(&self, deadline: Instant) -> bool {
+		self.0
+			.sleep
+			.work
+			.core_dispatched_hooks
+			.wait_zero(deadline)
+			.await
+	}
+
 	/// Spawn the fallback sleep timer used by `ActorContext`s that are not
 	/// bound to an `ActorTask`.
 	///
@@ -530,6 +542,13 @@ impl ActorContext {
 		}
 
 		let policy = kind.policy();
+		if policy.aborts_at_shutdown_deadline && self.is_lost() {
+			tracing::warn!(
+				kind = kind.label(),
+				"actor work spawned after the generation was lost; dropping it"
+			);
+			return false;
+		}
 		if policy.aborts_at_shutdown_deadline {
 			let mut shutdown_tasks = self.0.sleep.work.shutdown_tasks.lock();
 			if self.0.sleep.work.teardown_started.load(Ordering::Acquire) {
@@ -582,7 +601,19 @@ impl ActorContext {
 			let _region = region;
 			if policy.aborts_at_shutdown_deadline {
 				let shutdown_deadline = ctx.shutdown_deadline_token();
+				// A lost generation runs no more user code, so its work is dropped as soon as the
+				// engine gives up on it. Loss wins over a future that is also ready.
+				let lost = ctx.lost_signal();
 				tokio::select! {
+					biased;
+					_ = lost.cancelled() => {
+						tracing::debug!(
+							actor_id = %ctx.actor_id(),
+							kind = kind.label(),
+							reason = "generation_lost",
+							"actor work cancelled because the generation was lost"
+						);
+					}
 					_ = fut => {}
 					_ = shutdown_deadline.cancelled() => {
 						tracing::warn!(
@@ -636,6 +667,9 @@ impl ActorContext {
 				let task = async move {
 					let _region = region;
 					if policy.aborts_at_shutdown_deadline {
+						// Wasm work wraps JS promises that loss cannot stop, so it keeps only the
+						// shutdown-deadline policy. Lost wasm generations are fenced by storage
+						// revocation and the generation hold instead.
 						let shutdown_deadline = ctx_for_task.shutdown_deadline_token();
 						tokio::select! {
 							_ = fut => {}