feat(container-runner): report actors as crashed on unexpected platform SIGTERM

container-runner/src/actor.rs

@@ -7,20 +7,26 @@
 //! `on_destroy` stops the child while the instance stays warm for the next
 //! placement.
 
-use std::sync::Arc;
+use std::sync::{Arc, LazyLock};
 
 use anyhow::{Context, Result};
 use async_trait::async_trait;
 use rivetkit::{Actor, ActorKeySegment, Ctx, Request, Response, WebSocket, action};
 use tokio::sync::Mutex as TokioMutex;
 
 use crate::child::{ChildProcess, SpawnSpec, log_prefix};
 use crate::input::ActorInput;
 use crate::{
 	children, effective_stop_grace, release_child_port, reserve_child_port,
 	runner_config,
 };
 
+/// 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 +41,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;
@@ -87,6 +94,7 @@
 					"{} 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(());
 			}
@@ -152,6 +160,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(())
 	}
@@ -236,6 +248,38 @@
 	}
 }
 
+/// 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/main.rs

@@ -368,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()