diff --git a/Cargo.lock b/Cargo.lock index 12ad23aba3..183b07031f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5318,6 +5318,7 @@ dependencies = [ "clap", "futures-util", "nix 0.30.1", + "rand 0.8.5", "reqwest 0.12.22", "rivetkit", "scc", diff --git a/container-runner/Cargo.toml b/container-runner/Cargo.toml index e01f51bc66..74d16d54a3 100644 --- a/container-runner/Cargo.toml +++ b/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 diff --git a/container-runner/src/actor.rs b/container-runner/src/actor.rs index 80d25cda59..423987b089 100644 --- a/container-runner/src/actor.rs +++ b/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>> = + LazyLock::new(scc::HashMap::new); + pub struct GameServer { child: TokioMutex>>, } @@ -35,6 +62,7 @@ impl GameServer { // 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,11 +80,11 @@ impl GameServer { #[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 = (); @@ -64,7 +92,10 @@ impl Actor for GameServer { type Action = action::Raw; async fn create_state(_ctx: &Ctx, input: Self::Input) -> Result { - Ok(input) + Ok(ActorState { + input, + did_start: false, + }) } async fn create(_ctx: &Ctx) -> Result { @@ -78,23 +109,45 @@ impl Actor for GameServer { 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 @@ impl Actor for GameServer { 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 @@ impl Actor for GameServer { /// the framework reports an errored stop and the engine records the crash. async fn run(self: Arc, ctx: Ctx) -> 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 @@ impl Actor for GameServer { /// ahead of instance retirement); leaving the child running would orphan /// it on an instance the engine considers vacated. async fn on_sleep(self: Arc, ctx: Ctx) -> 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::() * 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 Actor for GameServer { } } +impl Handles for GameServer { + type Future = Pin> + Send>>; + + fn handle(self: Arc, ctx: Ctx, _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) { + 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) -> Option { let key = ctx.key(); if key.is_empty() { diff --git a/container-runner/src/input.rs b/container-runner/src/input.rs index 1b39d3c089..a56b5de843 100644 --- a/container-runner/src/input.rs +++ b/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 -- `). -//! 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, } +/// 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; diff --git a/container-runner/src/main.rs b/container-runner/src/main.rs index 081b031615..748fc84c8e 100644 --- a/container-runner/src/main.rs +++ b/container-runner/src/main.rs @@ -75,6 +75,11 @@ static EXIT: LazyLock = 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); diff --git a/rivetkit-rust/packages/rivetkit/src/start.rs b/rivetkit-rust/packages/rivetkit/src/start.rs index c2dd8ae576..fd21b4f3d0 100644 --- a/rivetkit-rust/packages/rivetkit/src/start.rs +++ b/rivetkit-rust/packages/rivetkit/src/start.rs @@ -436,7 +436,12 @@ async fn handle_actor_event( ShutdownKind::Destroy => actor.on_destroy(ctx).await, }; reply.send(result); - return Ok(true); + // Do not end the loop here. Core sends `SerializeState` during the + // finalize phase, after this cleanup, to capture any state the hook + // wrote (the NAPI runtime keeps its loop alive the same way). + // Returning early would drop the event receiver and make that + // enqueue fail with `not_ready`. The loop ends when core closes the + // channel after `save_final_state`. } ActorEvent::DisconnectConn { conn_id, reply } => { reply.send(ctx.disconnect_conn(&conn_id).await); @@ -947,6 +952,7 @@ mod tests { ); request_sleep(&tx).await; + drop(tx); actor.await.expect("join run_actor").expect("run actor"); } @@ -968,6 +974,7 @@ mod tests { ); request_sleep(&tx).await; + drop(tx); actor.await.expect("join run_actor").expect("run actor"); } @@ -988,6 +995,7 @@ mod tests { assert_eq!(response.status().as_u16(), 404); request_sleep(&tx).await; + drop(tx); actor.await.expect("join run_actor").expect("run actor"); } @@ -1001,6 +1009,7 @@ mod tests { assert_eq!(state.created, 1); request_sleep(&tx).await; + drop(tx); actor.await.expect("join run_actor").expect("run actor"); } @@ -1055,6 +1064,7 @@ mod tests { assert!(error.to_string().contains("websockets not supported")); request_sleep(&tx).await; + drop(tx); actor.await.expect("join run_actor").expect("run actor"); } @@ -1090,6 +1100,7 @@ mod tests { tx.send(ActorEvent::ConnectionClosed { conn }) .expect("send connection closed"); request_sleep(&tx).await; + drop(tx); actor.await.expect("join run_actor").expect("run actor"); let log = &ctx.state().log; @@ -1119,6 +1130,7 @@ mod tests { assert!(error.to_string().contains("subscribe denied")); request_sleep(&tx).await; + drop(tx); actor.await.expect("join run_actor").expect("run actor"); let log = &ctx.state().log; @@ -1153,6 +1165,7 @@ mod tests { assert!(error.to_string().contains("connection rejected")); request_sleep(&tx).await; + drop(tx); actor.await.expect("join run_actor").expect("run actor"); assert!( !ctx.state() @@ -1170,6 +1183,7 @@ mod tests { let actor = tokio::spawn(run_actor::(start)); request_sleep(&tx).await; + drop(tx); actor.await.expect("join run_actor").expect("run actor"); assert!(ctx.state().log.iter().any(|entry| entry == "run_aborted")); @@ -1183,11 +1197,35 @@ mod tests { let actor = tokio::spawn(run_actor::(start)); request_destroy(&tx).await; + drop(tx); actor.await.expect("join run_actor").expect("run actor"); assert!(ctx.state().log.iter().any(|entry| entry == "on_destroy")); } + #[tokio::test] + async fn run_actor_serializes_state_after_cleanup() { + // Core sends `SerializeState` during finalize, after the cleanup hook, + // to capture state the hook wrote. The event loop must stay alive to + // handle it instead of ending on the cleanup event (which dropped the + // receiver and failed the serialize with `not_ready`). + let (tx, rx) = unbounded_channel(); + let start = lifecycle_start(Some(cbor(&LifecycleInput { count: 5 })), None, rx.into()); + let actor = tokio::spawn(run_actor::(start)); + + request_sleep(&tx).await; + let state = decode_actor_state(request_serialize(&tx).await); + assert_eq!(state.count, 5); + assert!( + state.log.iter().any(|entry| entry == "on_sleep"), + "serialize after cleanup lost on_sleep state, got: {:?}", + state.log + ); + + drop(tx); + actor.await.expect("join run_actor").expect("run actor"); + } + #[tokio::test] async fn run_actor_error_requests_errored_stop() { let (tx, rx) = unbounded_channel(); @@ -1335,6 +1373,7 @@ mod tests { ); request_sleep(&tx).await; + drop(tx); actor.await.expect("join run_actor").expect("run actor"); } @@ -1373,6 +1412,7 @@ mod tests { ); request_sleep(&tx).await; + drop(tx); actor.await.expect("join run_actor").expect("run actor"); } @@ -1400,6 +1440,7 @@ mod tests { assert_eq!(error.code(), "action_not_found"); request_sleep(&tx).await; + drop(tx); actor.await.expect("join run_actor").expect("run actor"); } @@ -1429,6 +1470,7 @@ mod tests { ); request_sleep(&tx).await; + drop(tx); actor.await.expect("join run_actor").expect("run actor"); } @@ -1465,6 +1507,7 @@ mod tests { assert_eq!(error.code(), "not_found"); request_sleep(&tx).await; + drop(tx); actor.await.expect("join run_actor").expect("run actor"); } @@ -1506,6 +1549,7 @@ mod tests { ); request_sleep(&tx).await; + drop(tx); actor.await.expect("join run_actor").expect("run actor"); }