feat(container-runner): self-destroy actors that restart after stopping

container-runner/Cargo.toml

@@ -18,6 +18,7 @@ async-trait.workspace = true
 clap = { workspace = true, features = ["env"] }
 futures-util.workspace = true
 nix = { workspace = true, features = ["process"] }
+rand.workspace = true
 reqwest = { workspace = true, features = ["stream"] }
 rivetkit.workspace = true
 scc.workspace = true

container-runner/src/actor.rs

@@ -7,20 +7,47 @@
 //! `on_destroy` stops the child while the instance stays warm for the next
 //! placement.
 
-use std::sync::Arc;
+use std::future::Future;
+use std::pin::Pin;
+use std::sync::{Arc, LazyLock};
+use std::time::Duration;
 
 use anyhow::{Context, Result};
 use async_trait::async_trait;
-use rivetkit::{Actor, ActorKeySegment, Ctx, Request, Response, WebSocket, action};
+use rivetkit::{Action, Actor, ActorKeySegment, Ctx, Handles, Request, Response, WebSocket, action};
+use serde::{Deserialize, Serialize};
 use tokio::sync::Mutex as TokioMutex;
 
 use crate::child::{ChildProcess, SpawnSpec, log_prefix};
-use crate::input::ActorInput;
+use crate::input::{ActorInput, ActorState};
 use crate::{
 	children, effective_stop_grace, release_child_port, reserve_child_port,
 	runner_config,
 };
 
+/// Upper bound, in milliseconds, of the random delay before a stopped container
+/// is woken to destroy itself. Spreading the wakes keeps a wave of containers
+/// that stopped together from waking the engine in a single burst.
+const REAP_MAX_JITTER_MS: f64 = 8_000.0;
+
+/// Action scheduled on stop purely to wake the actor so it can self-destroy.
+/// Carries no data; the real work happens in `on_start`, which sees `did_start`
+/// and destroys before this would run.
+#[derive(Debug, Serialize, Deserialize)]
+pub struct Reap;
+
+impl Action for Reap {
+	type Output = ();
+
+	const NAME: &'static str = "reap";
+}
+
+/// Live actor contexts on this instance, keyed by actor id. Lets the process
+/// shutdown path report actors as crashed when the platform reclaims the
+/// container out from under them.
+static ACTOR_CTXS: LazyLock<scc::HashMap<String, Ctx<GameServer>>> =
+	LazyLock::new(scc::HashMap::new);
+
 pub struct GameServer {
 	child: TokioMutex<Option<Arc<ChildProcess>>>,
 }
@@ -35,6 +62,7 @@
 		// deliberate, then stop. `stop` is idempotent if the process shutdown
 		// sweep already stopped this child.
 		children().remove_async(actor_id).await;
+		ACTOR_CTXS.remove_async(actor_id).await;
 		let child = self.child.lock().await.take();
 		if let Some(child) = child {
 			child.stop(effective_stop_grace()).await;
@@ -52,19 +80,22 @@
 
 #[async_trait]
 impl Actor for GameServer {
-	// The launch spec is the persisted state: a woken actor restores the same
-	// spec without the engine re-sending input.
-	type State = ActorInput;
+	// State wraps the launch spec plus the one-shot `did_start` flag. A woken
+	// actor restores the same spec without the engine re-sending input.
+	type State = ActorState;
 	type Input = ActorInput;
-	type Actions = ();
+	type Actions = (Reap,);
 	type Events = ();
 	type Queue = ();
 	type ConnParams = ();
 	type ConnState = ();
 	type Action = action::Raw;
 
 	async fn create_state(_ctx: &Ctx<Self>, input: Self::Input) -> Result<Self::State> {
-		Ok(input)
+		Ok(ActorState {
+			input,
+			did_start: false,
+		})
 	}
 
 	async fn create(_ctx: &Ctx<Self>) -> Result<Self> {
@@ -74,27 +105,49 @@
 	}
 
 	async fn on_start(self: Arc<Self>, ctx: Ctx<Self>) -> Result<()> {
 		let cfg = runner_config();
 		let actor_id = ctx.actor_id().to_string();
 		let key = actor_key_string(&ctx);
 
+
 		// An engine retry for an actor that is already running here must be an
 		// idempotent no-op: rejecting it would make the engine tear down a
-		// healthy actor.
+		// healthy actor. This is the same generation still hosting its child, not
+		// a restart, so it must be handled before the one-shot guard below.
 		if let Some(existing) = children().read_async(&actor_id, |_, c| c.clone()).await {
 			if !existing.has_exited() {
 				println!(
 					"{} runner: actor already running, ignoring duplicate start",
 					log_prefix(&actor_id, existing.key.as_deref())
 				);
+				register_ctx(&actor_id, &ctx).await;
 				*self.child.lock().await = Some(existing);
 				return Ok(());
 			}
 		}
 
+		// A container hosts exactly one child lifetime. `did_start` is persisted the
+		// first time the container starts, so seeing it already set means this is a
+		// restart after the container stopped (slept or crashed). Destroy instead of
+		// respawning the child. This runs after the framework marks the lifecycle
+		// started, so `destroy` is valid here; `run` exits cleanly once the destroy
+		// is in flight. Writing the flag on start (not in `on_sleep`) is what makes
+		// it durable: state written during startup is captured by the normal and
+		// final save cycles, whereas an `on_sleep` write races the shutdown
+		// serialization.
+		if ctx.state().did_start {
+			tracing::info!(
+				actor_id = %actor_id,
+				"container restarted after stopping, destroying instead of respawning child"
+			);
+			return ctx.destroy();
+		}
+		ctx.state_mut().did_start = true;
+		ctx.request_save();
+
 		// Copy the launch spec out of the state guard before any await.
 		let (input_port, mut parts, env) = {
-			let input = ctx.state();
+			let input = &ctx.state().input;
 			// input.command overrides the CLI template; input.args are appended.
 			let mut parts = input
 				.command
@@ -152,6 +205,10 @@
 			release_child_port(child_port).await;
 			anyhow::bail!("a child for actor {actor_id} is already registered");
 		}
+		// Register only now that startup has succeeded. Registering earlier would
+		// leak an entry for any generation whose start failed, since a failed
+		// start never runs on_destroy/on_sleep to remove it.
+		register_ctx(&actor_id, &ctx).await;
 		*self.child.lock().await = Some(child);
 		Ok(())
 	}
@@ -163,6 +220,13 @@
 	/// the framework reports an errored stop and the engine records the crash.
 	async fn run(self: Arc<Self>, ctx: Ctx<Self>) -> Result<()> {
 		let Some(child) = self.child.lock().await.clone() else {
+			// A restart guarded by `did_start` requests destroy in `on_start`
+			// without spawning a child. Exit cleanly rather than reporting a
+			// spurious crash; only a missing child with no destroy in flight is a
+			// real bug.
+			if ctx.inner().is_destroy_requested() {
+				return Ok(());
+			}
 			anyhow::bail!("run: child process was never spawned");
 		};
 
@@ -226,6 +290,21 @@
 	/// ahead of instance retirement); leaving the child running would orphan
 	/// it on an instance the engine considers vacated.
 	async fn on_sleep(self: Arc<Self>, ctx: Ctx<Self>) -> Result<()> {
+		// Arm a wake so the stopped container is reaped proactively instead of
+		// lingering asleep until its next request. On that wake `on_start` sees
+		// `did_start` and destroys. This runs on every stop, including a
+		// crash-induced sleep (an errored `run` makes the engine sleep the actor,
+		// which runs this hook). A scheduled event is used rather than the raw
+		// `set_alarm`, because core's shutdown alarm sync recomputes the engine
+		// alarm from the schedule store and would wipe a raw alarm.
+		let jitter = Duration::from_millis((rand::random::<f64>() * REAP_MAX_JITTER_MS) as u64);
+		if let Err(err) = ctx.schedule().after(jitter, Reap::NAME, &[]).await {
+			tracing::warn!(
+				actor_id = %ctx.actor_id(),
+				error = ?err,
+				"failed to schedule reap wake for stopped container"
+			);
+		}
 		self.stop_child(ctx.actor_id(), "actor sleeping").await;
 		Ok(())
 	}
@@ -236,6 +315,54 @@
 	}
 }
 
+impl Handles<Reap> for GameServer {
+	type Future = Pin<Box<dyn Future<Output = Result<()>> + Send>>;
+
+	fn handle(self: Arc<Self>, ctx: Ctx<Self>, _action: Reap) -> Self::Future {
+		Box::pin(async move {
+			// Reached only if the reap wake fires before `on_start` destroyed the
+			// actor (which it should have). Request destroy defensively so a stopped
+			// container never keeps running.
+			if let Err(err) = ctx.destroy() {
+				tracing::debug!(actor_id = %ctx.actor_id(), error = ?err, "reap destroy failed");
+			}
+			Ok(())
+		})
+	}
+}
+
+/// Register an actor context for crash-on-shutdown reporting. Overwrites any
+/// stale entry left by a prior generation with the same id.
+async fn register_ctx(actor_id: &str, ctx: &Ctx<GameServer>) {
+	ACTOR_CTXS.remove_async(actor_id).await;
+	let _ = ACTOR_CTXS
+		.insert_async(actor_id.to_string(), ctx.clone())
+		.await;
+}
+
+/// Report every live actor on this instance as crashed. Called when the
+/// platform reclaims the container (an unexpected SIGTERM) so the reclaim
+/// surfaces as a crash on the engine instead of a silent reallocation. Runs
+/// while the envoy is still connected so the crash reaches the engine.
+pub async fn crash_all_actors(message: &str) {
+	let mut ctxs = Vec::new();
+	ACTOR_CTXS
+		.retain_async(|_, ctx| {
+			ctxs.push(ctx.clone());
+			false
+		})
+		.await;
+	for ctx in ctxs {
+		if let Err(err) = ctx.stop_with_error(message) {
+			tracing::debug!(
+				actor_id = %ctx.actor_id(),
+				error = ?err,
+				"crash-on-shutdown stop_with_error failed"
+			);
+		}
+	}
+}
+
 fn actor_key_string(ctx: &Ctx<GameServer>) -> Option<String> {
 	let key = ctx.key();
 	if key.is_empty() {

container-runner/src/input.rs

@@ -1,20 +1,20 @@
-//! The actor input payload describing how to launch the child game server.
+//! The actor input payload describing how to launch the child game server, and
+//! the persisted actor state that wraps it.
 //!
 //! Everything the game server needs to launch (command, args, env, port) is
 //! carried in the actor's create-time `input` payload, CBOR-encoded per the
 //! RivetKit convention. All fields are optional; anything omitted falls back
 //! to the CLI-provided template (`rivet-container-runner -- <command...>`).
-//! The decoded input is also the actor's persisted state so a woken actor
-//! restores the same launch spec without re-decoding input.
+//! The input is preserved inside `ActorState` so a woken actor restores the
+//! same launch spec without re-decoding input.
 
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
 
 /// Shape of the actor `input` payload. Unknown fields are ignored rather than
-/// rejected: this type is also the persisted actor state, and a strict decode
-/// would break waking actors after a rollback to a binary that predates a
-/// newly added field.
-#[derive(Debug, Default, Serialize, Deserialize)]
+/// rejected: a strict decode would break waking actors after a rollback to a
+/// binary that predates a newly added field.
+#[derive(Debug, Default, Clone, Serialize, Deserialize)]
 pub struct ActorInput {
 	/// Overrides the CLI command template entirely (program + fixed args).
 	#[serde(default)]
@@ -31,6 +31,25 @@ pub struct ActorInput {
 	pub port: Option<u16>,
 }
 
+/// Persisted actor state. Wraps the launch spec and tracks whether the container
+/// has already started once. Unknown fields are ignored for the same
+/// rollback-safety reason as `ActorInput`.
+#[derive(Debug, Default, Serialize, Deserialize)]
+pub struct ActorState {
+	/// The create-time launch spec, preserved so a woken actor can respawn the
+	/// same child without the engine re-sending input.
+	#[serde(default)]
+	pub input: ActorInput,
+	/// `false` on create, set to `true` on the container's first start and
+	/// persisted. A container hosts exactly one child lifetime: seeing this set on
+	/// a later start means a restart after the container stopped, so the woken
+	/// actor destroys itself instead of respawning. Tracked on start rather than
+	/// on stop because startup writes persist reliably while an `on_sleep` write
+	/// races the shutdown state serialization.
+	#[serde(default)]
+	pub did_start: bool,
+}
+
 #[cfg(test)]
 #[path = "../tests/inline/input.rs"]
 mod tests;

container-runner/src/main.rs

@@ -75,6 +75,11 @@ static EXIT: LazyLock<CancellationToken> = LazyLock::new(CancellationToken::new)
 /// grace period instead).
 static SIGNAL_SHUTDOWN: AtomicBool = AtomicBool::new(false);
 
+/// Set when shutdown was triggered by a platform SIGTERM (an instance reclaim),
+/// as opposed to a local SIGINT (developer Ctrl-C). Distinguishes a reclaim
+/// from a manual stop for downstream shutdown handling.
+static PLATFORM_RECLAIM: AtomicBool = AtomicBool::new(false);
+
 /// How long the platform gives this container between SIGTERM and SIGKILL.
 /// Cloud Run defaults to 10 seconds (configurable up to 60 on the service);
 /// keep this in sync with the platform setting via RIVET_SIGTERM_BUDGET_SECS.
@@ -363,6 +368,17 @@ async fn async_main() -> Result<()> {
 	// and the runtime drains unbounded so the /start SSE flushes cleanly.
 	EXIT.cancelled().await;
 	if SIGNAL_SHUTDOWN.load(Ordering::Acquire) {
+		// A platform SIGTERM reclaims this instance. Report every actor as crashed
+		// before draining so an unexpected SIGTERM (OOM or the ~60 minute request
+		// cap) surfaces as a crash on the engine instead of a silent reallocation.
+		// This runs while the envoy is still connected so the crash reaches the
+		// engine. A local SIGINT (Ctrl-C) drains gracefully without a crash.
+		if PLATFORM_RECLAIM.load(Ordering::Acquire) {
+			crate::actor::crash_all_actors(
+				"runner received unexpected platform SIGTERM, likely OOM or running longer than 60 minutes",
+			)
+			.await;
+		}
 		if tokio::time::timeout(signal_drain_timeout(), runtime.shutdown())
 			.await
 			.is_err()
@@ -417,7 +433,12 @@ fn spawn_signal_handler() {
 		let mut sigterm = signal(SignalKind::terminate()).expect("install SIGTERM handler");
 		let mut sigint = signal(SignalKind::interrupt()).expect("install SIGINT handler");
 		tokio::select! {
-			_ = sigterm.recv() => tracing::info!("received SIGTERM"),
+			_ = sigterm.recv() => {
+				PLATFORM_RECLAIM.store(true, Ordering::Release);
+				tracing::warn!(
+					"unexpected platform SIGTERM received, likely hitting OOM or running longer than 60 minutes"
+				);
+			}
 			_ = sigint.recv() => tracing::info!("received SIGINT"),
 		}
 		SIGNAL_SHUTDOWN.store(true, Ordering::Release);