From 5d9598ada61c9b27c0bf9b178b920d893c890c68 Mon Sep 17 00:00:00 2001 From: ABCxFF <79597906+abcxff@users.noreply.github.com> Date: Wed, 12 Aug 2026 14:57:58 -0400 Subject: [PATCH 1/9] fix(envoy-client): ack terminating stop commands so pegboard-envoy stops replaying them --- engine/sdks/rust/envoy-client/src/commands.rs | 95 +++++++++-- .../rust/envoy-client/tests/command_dedup.rs | 153 +++++++++++++++++- 2 files changed, 230 insertions(+), 18 deletions(-) diff --git a/engine/sdks/rust/envoy-client/src/commands.rs b/engine/sdks/rust/envoy-client/src/commands.rs index 342fa4d022..ffe86f993b 100644 --- a/engine/sdks/rust/envoy-client/src/commands.rs +++ b/engine/sdks/rust/envoy-client/src/commands.rs @@ -1,3 +1,5 @@ +use std::collections::HashMap; + use rivet_envoy_protocol as protocol; use crate::actor::create_actor; @@ -16,6 +18,14 @@ pub async fn handle_commands(ctx: &mut EnvoyContext, commands: Vec = commands + .iter() + .filter(|c| matches!(c.inner, protocol::Command::CommandStopActor(_))) + .map(|c| (c.checkpoint.actor_id.clone(), c.checkpoint.generation)) + .collect(); + for command_wrapper in commands { let checkpoint = command_wrapper.checkpoint; let dedup_key = (checkpoint.actor_id.clone(), checkpoint.generation); @@ -80,35 +90,59 @@ pub async fn handle_commands(ctx: &mut EnvoyContext, commands: Vec = Vec::new(); +/// Ack only the given actors' latest processed command index. Used for the +/// immediate stop 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)]) { + let mut highest: HashMap<(String, u32), i64> = HashMap::new(); + for key in actors { + if let Some(&index) = ctx.processed_command_idx.get(key) { + highest.insert(key.clone(), index); + } + } + if highest.is_empty() { + return; + } + + send_ack_checkpoints(ctx, checkpoints_from(highest)).await; +} + +pub async fn send_command_ack(ctx: &mut EnvoyContext) { + // Merge live actors and the dedup map, highest index per actor-generation. + // Live actors are re-acked every tick (recovers an ack accepted locally but + // never committed by the server); the dedup map covers stops whose actor was + // already removed and is cleared once acked. + let mut highest: HashMap<(String, u32), i64> = HashMap::new(); for (actor_id, generations) in &ctx.actors { for (generation, entry) in generations { - if entry.last_command_idx < 0 { - continue; + if entry.last_command_idx >= 0 { + highest.insert((actor_id.clone(), *generation), entry.last_command_idx); } - last_command_checkpoints.push(protocol::ActorCheckpoint { - actor_id: actor_id.clone(), - generation: *generation, - index: entry.last_command_idx, - }); } } + for ((actor_id, generation), &index) in &ctx.processed_command_idx { + highest + .entry((actor_id.clone(), *generation)) + .and_modify(|existing| *existing = (*existing).max(index)) + .or_insert(index); + } - if last_command_checkpoints.is_empty() { + if highest.is_empty() { return; } - let send_failed = ws_send( - &ctx.shared, - protocol::ToRivet::ToRivetAckCommands(protocol::ToRivetAckCommands { - last_command_checkpoints: last_command_checkpoints.clone(), - }), - ) - .await; + let last_command_checkpoints = checkpoints_from(highest); + let send_failed = send_ack_checkpoints(ctx, last_command_checkpoints.clone()).await; // Skip the dedup clear if the ack never left this process. Otherwise // `pegboard-envoy` would replay the commands on reconnect with no dedup @@ -127,8 +161,35 @@ pub async fn send_command_ack(ctx: &mut EnvoyContext) { // window is narrow (the gap between OS-accepted bytes and the FDB // commit), but a strictly correct fix needs an ack-of-ack from // `pegboard-envoy` so we only clear after positive confirmation. + // This now also applies to removed actors whose stops are acked here: a + // short-lived actor can be resurrected in the same window. Same fix. for cp in &last_command_checkpoints { ctx.processed_command_idx .remove(&(cp.actor_id.clone(), cp.generation)); } } + +fn checkpoints_from(highest: HashMap<(String, u32), i64>) -> Vec { + highest + .into_iter() + .map(|((actor_id, generation), index)| protocol::ActorCheckpoint { + actor_id, + generation, + index, + }) + .collect() +} + +/// Send an ack for the given checkpoints. Returns whether the send failed. +async fn send_ack_checkpoints( + ctx: &EnvoyContext, + last_command_checkpoints: Vec, +) -> bool { + ws_send( + &ctx.shared, + protocol::ToRivet::ToRivetAckCommands(protocol::ToRivetAckCommands { + last_command_checkpoints, + }), + ) + .await +} diff --git a/engine/sdks/rust/envoy-client/tests/command_dedup.rs b/engine/sdks/rust/envoy-client/tests/command_dedup.rs index 0f48ef5dfe..79b445cba8 100644 --- a/engine/sdks/rust/envoy-client/tests/command_dedup.rs +++ b/engine/sdks/rust/envoy-client/tests/command_dedup.rs @@ -3,7 +3,7 @@ use std::sync::Arc; use rivet_envoy_client::actor::ToActor; use rivet_envoy_client::async_counter::AsyncCounter; -use rivet_envoy_client::commands::handle_commands; +use rivet_envoy_client::commands::{handle_commands, send_command_ack}; use rivet_envoy_client::config::{ BoxFuture, EnvoyCallbacks, EnvoyConfig, HttpRequest, HttpResponse, WebSocketHandler, WebSocketSender, @@ -18,6 +18,7 @@ use rivet_envoy_client::sqlite::{ use rivet_envoy_client::utils::{BufferMap, RemoteSqliteIndeterminateResultError}; use rivet_envoy_protocol as protocol; use tokio::sync::mpsc; +use vbare::OwnedVersionedData; struct IdleCallbacks; @@ -271,3 +272,153 @@ async fn replayed_command_is_dropped_after_remote_sql_lost_response() { handle_commands(&mut ctx, vec![stop_command("actor-replay", 1, 5)]).await; assert!(actor_rx.try_recv().is_err()); } + +fn decode_ack_checkpoints(msg: WsTxMessage) -> Vec { + let WsTxMessage::Send(bytes) = msg else { + panic!("expected a websocket send, got a close"); + }; + let message = protocol::versioned::ToRivet::deserialize(&bytes, protocol::PROTOCOL_VERSION) + .expect("failed to decode ToRivet message"); + match message { + protocol::ToRivet::ToRivetAckCommands(val) => val.last_command_checkpoints, + _ => panic!("expected ToRivetAckCommands"), + } +} + +#[tokio::test] +async fn stop_command_is_acked_immediately() { + let mut ctx = new_envoy_context(); + let (actor_tx, mut actor_rx) = mpsc::unbounded_channel::(); + ctx.insert_actor( + "actor-a".to_string(), + 1, + actor_tx, + Arc::new(AsyncCounter::new()), + "actor-a".to_string(), + -1, + ); + + let (ws_tx, mut ws_rx) = mpsc::unbounded_channel(); + *ctx.shared.ws_tx.lock().await = Some(ws_tx); + + handle_commands(&mut ctx, vec![stop_command("actor-a", 1, 5)]).await; + assert!(matches!( + actor_rx.try_recv(), + Ok(ToActor::Stop { command_idx: 5, .. }) + )); + + // The stop must be acked right away rather than waiting for the periodic + // tick, otherwise the actor entry is gone before the next ack. + let checkpoints = + decode_ack_checkpoints(ws_rx.try_recv().expect("stop 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, 5); + + // A successful immediate ack must still retain the dedup entry. Only the + // periodic tick clears it, so a replay can re-ack if the server never + // committed this ack. + assert_eq!( + ctx.processed_command_idx.get(&("actor-a".to_string(), 1)), + Some(&5) + ); +} + +#[tokio::test] +async fn stop_ack_retried_via_replay_after_failed_send() { + let mut ctx = new_envoy_context(); + let (actor_tx, mut actor_rx) = mpsc::unbounded_channel::(); + ctx.insert_actor( + "actor-a".to_string(), + 1, + actor_tx, + Arc::new(AsyncCounter::new()), + "actor-a".to_string(), + -1, + ); + + // No websocket is connected, so the immediate ack send fails. The processed + // index must be retained so a later replay can re-ack it. + handle_commands(&mut ctx, vec![stop_command("actor-a", 1, 5)]).await; + assert!(matches!( + actor_rx.try_recv(), + Ok(ToActor::Stop { command_idx: 5, .. }) + )); + assert_eq!( + ctx.processed_command_idx.get(&("actor-a".to_string(), 1)), + Some(&5) + ); + + // Remove the actor, as happens once it emits its Stopped event. The re-ack + // must still work with no live actor, sourced from the dedup map. + ctx.remove_actor("actor-a", 1); + + // Reconnect and replay the same stop. Dedup skips reprocessing, but the batch + // still carries a stop, so the retained checkpoint is re-acked. + let (ws_tx, mut ws_rx) = mpsc::unbounded_channel(); + *ctx.shared.ws_tx.lock().await = Some(ws_tx); + handle_commands(&mut ctx, vec![stop_command("actor-a", 1, 5)]).await; + assert!( + actor_rx.try_recv().is_err(), + "replayed stop must not be reprocessed" + ); + + let checkpoints = + decode_ack_checkpoints(ws_rx.try_recv().expect("replayed stop should re-ack")); + assert_eq!(checkpoints.len(), 1); + assert_eq!(checkpoints[0].index, 5); +} + +#[tokio::test] +async fn unknown_actor_stop_is_acked() { + let mut ctx = new_envoy_context(); + let (ws_tx, mut ws_rx) = mpsc::unbounded_channel(); + *ctx.shared.ws_tx.lock().await = Some(ws_tx); + + // No actor inserted: models a stop replayed to a process that never started + // it (e.g. after restart). It must still be acked to stop the replay. + handle_commands(&mut ctx, vec![stop_command("actor-gone", 3, 9)]).await; + + let checkpoints = decode_ack_checkpoints( + ws_rx + .try_recv() + .expect("unknown-actor stop should still ack"), + ); + assert_eq!(checkpoints.len(), 1); + assert_eq!(checkpoints[0].actor_id, "actor-gone"); + assert_eq!(checkpoints[0].generation, 3); + assert_eq!(checkpoints[0].index, 9); +} + +#[tokio::test] +async fn live_actor_is_reacked_on_each_tick() { + let mut ctx = new_envoy_context(); + let (actor_tx, _actor_rx) = mpsc::unbounded_channel::(); + // A live actor whose latest command index is 3. + ctx.insert_actor( + "actor-a".to_string(), + 1, + actor_tx, + Arc::new(AsyncCounter::new()), + "actor-a".to_string(), + 3, + ); + + let (ws_tx, mut ws_rx) = mpsc::unbounded_channel(); + *ctx.shared.ws_tx.lock().await = Some(ws_tx); + + // The first tick acks index 3 and clears the dedup map. The second tick must + // still re-ack from the live-actor scan, recovering an ack the server may + // never have committed. + send_command_ack(&mut ctx).await; + let first = decode_ack_checkpoints(ws_rx.try_recv().expect("first tick should ack")); + assert_eq!(first.len(), 1); + assert_eq!(first[0].index, 3); + + send_command_ack(&mut ctx).await; + let second = decode_ack_checkpoints(ws_rx.try_recv().expect("second tick should re-ack")); + assert_eq!(second.len(), 1); + assert_eq!(second[0].actor_id, "actor-a"); + assert_eq!(second[0].index, 3); +} From 2d0f090bc134943a2633a994d69d818ef3100bdb Mon Sep 17 00:00:00 2001 From: ABCxFF <79597906+abcxff@users.noreply.github.com> Date: Thu, 20 Aug 2026 14:32:04 -0400 Subject: [PATCH 2/9] feat(container-runner): drain children and engine concurrently on SIGTERM --- container-runner/src/main.rs | 61 +++++++++++++++++------------------- 1 file changed, 29 insertions(+), 32 deletions(-) diff --git a/container-runner/src/main.rs b/container-runner/src/main.rs index 4594598431..9059e10bec 100644 --- a/container-runner/src/main.rs +++ b/container-runner/src/main.rs @@ -29,6 +29,7 @@ use std::time::Duration; use anyhow::Result; use clap::Parser; +use futures_util::future::join_all; use rivetkit::serverless_http::{self, ListenerConfig}; use rivetkit::{ActorConfig, EngineSpawnMode, Registry, ServeConfig}; use tokio_util::sync::CancellationToken; @@ -103,30 +104,19 @@ static SIGNAL_SHUTDOWN: AtomicBool = AtomicBool::new(false); static PLATFORM_RECLAIM: AtomicBool = AtomicBool::new(false); /// How long the platform gives this container between SIGTERM and SIGKILL. -/// Defaults to 10 seconds; keep this in sync with the platform's actual budget -/// via RIVET_SIGTERM_BUDGET_SECS. -/// The signal-path teardown splits the budget: ~60% for the engine drain -/// (whose per-actor stops SIGTERM children with a grace capped at ~40%), 1s -/// for the straggler sweep, and the rest as margin. +/// Defaults to 9s, one second under the common ~10s platform budget so the whole +/// teardown lands before SIGKILL; keep it in sync with the platform's actual +/// budget via RIVET_SIGTERM_BUDGET_SECS. On the signal path the engine drain and +/// the child kills run concurrently, each bounded by this full budget. static SIGTERM_BUDGET: LazyLock = LazyLock::new(|| { let secs = std::env::var("RIVET_SIGTERM_BUDGET_SECS") .ok() .and_then(|value| value.parse::().ok()) - .unwrap_or(10) + .unwrap_or(9) .max(3); Duration::from_secs(secs) }); -fn signal_drain_timeout() -> Duration { - SIGTERM_BUDGET.mul_f64(0.6) -} - -fn signal_child_stop_grace() -> Duration { - SIGTERM_BUDGET.mul_f64(0.4) -} - -const SIGNAL_SWEEP_GRACE: Duration = Duration::from_secs(1); - pub fn runner_config() -> Arc { RUNNER_CONFIG .get() @@ -203,7 +193,7 @@ pub async fn release_child_port(port: u16) { pub fn effective_stop_grace() -> Duration { let grace = runner_config().stop_grace; if SIGNAL_SHUTDOWN.load(Ordering::Acquire) { - grace.min(signal_child_stop_grace()) + grace.min(*SIGTERM_BUDGET) } else { grace } @@ -395,11 +385,13 @@ async fn async_main() -> Result<()> { // currently unreachable and kept only as a fallback for a future // actor-driven exit. // - // Signal (platform is reclaiming the instance): tell the engine FIRST so - // it can start re-placing actors immediately. Its per-actor stops run our - // on_destroy hooks, which SIGTERM children with the capped signal grace. - // The drain is bounded so an unreachable engine cannot eat the whole - // platform budget; the sweep then catches any child whose hooks never ran. + // Signal (platform is reclaiming the instance): kill our children AND + // notify the engine at the SAME time, each bounded by the full SIGTERM + // budget. The direct sweep is what guarantees children die within budget + // rather than waiting on an engine round-trip to run our on_destroy hooks; + // notifying the engine in parallel just lets it start re-placing actors + // immediately. Bounding the drain means an unreachable engine cannot eat + // the budget the children need. // // Fallback actor-driven exit (unreachable today): no platform deadline. // Children are already reaped by the hooks (the sweep is a no-op backstop), @@ -417,13 +409,15 @@ async fn async_main() -> Result<()> { ) .await; } - if tokio::time::timeout(signal_drain_timeout(), runtime.shutdown()) - .await - .is_err() - { - tracing::warn!("engine drain exceeded the signal budget, sweeping children directly"); - } - stop_all_children(SIGNAL_SWEEP_GRACE).await; + let drain = async { + if tokio::time::timeout(*SIGTERM_BUDGET, runtime.shutdown()) + .await + .is_err() + { + tracing::warn!("engine drain exceeded the signal budget"); + } + }; + tokio::join!(drain, stop_all_children(*SIGTERM_BUDGET)); } else { stop_all_children(stop_grace).await; runtime.shutdown().await; @@ -442,7 +436,9 @@ async fn async_main() -> Result<()> { /// Stop every child still in the registry. Actor `on_destroy` normally reaps /// its own child first; this is the belt-and-suspenders sweep for the signal -/// path so children are never orphaned. +/// path so children are never orphaned. Children are stopped concurrently so +/// each gets the full `grace` within the SIGTERM budget instead of queueing +/// behind the others. async fn stop_all_children(grace: Duration) { let mut children: Vec> = Vec::new(); CHILDREN @@ -455,10 +451,11 @@ async fn stop_all_children(grace: Duration) { return; } println!("runner: shutdown, stopping {} child(ren)", children.len()); - for child in children { + join_all(children.into_iter().map(|child| async move { child.stop(grace).await; release_child_port(child.child_port).await; - } + })) + .await; } fn env_u16(key: &str) -> Option { From b391ae42f486919d352a73719d466338eb65d83b Mon Sep 17 00:00:00 2001 From: ABCxFF <79597906+abcxff@users.noreply.github.com> Date: Fri, 21 Aug 2026 12:35:14 -0400 Subject: [PATCH 3/9] feat(container-runner): drain child before SIGTERM on engine pause --- container-runner/src/actor.rs | 32 ++++++++++++++++++-- container-runner/src/main.rs | 57 ++++++++++++++++++++++------------- 2 files changed, 65 insertions(+), 24 deletions(-) diff --git a/container-runner/src/actor.rs b/container-runner/src/actor.rs index 79dc1dd7b7..800618ca49 100644 --- a/container-runner/src/actor.rs +++ b/container-runner/src/actor.rs @@ -17,7 +17,8 @@ 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, + children, drain_grace, effective_stop_grace, exit_token, release_child_port, + reserve_child_port, runner_config, }; /// Live actor contexts on this instance, keyed by actor id. Lets the process @@ -54,6 +55,31 @@ impl GameServer { // self-exit could otherwise lose. tracing::info!(actor_id = %actor_id, reason, "actor stopped, keeping instance warm"); } + + /// Engine pause path (sleep, lost, going-away). Give the child up to + /// `DRAIN_GRACE` to finish its in-flight work and exit on its own before we + /// force a stop; the child is not signalled during the window. A natural + /// child exit ends the wait immediately, and a platform SIGTERM (which + /// cancels the exit token) cuts it short so the reclaim's SIGTERM→SIGKILL + /// budget is honored. + async fn drain_then_stop_child(&self, actor_id: &str, reason: &str) { + let child = self.child.lock().await.clone(); + if let Some(child) = child { + if !child.has_exited() { + let prefix = log_prefix(actor_id, child.key.as_deref()); + println!( + "{prefix} runner: draining child for up to {:?} before stopping", + drain_grace() + ); + tokio::select! { + _ = child.wait_exit() => {} + _ = tokio::time::sleep(drain_grace()) => {} + _ = exit_token().cancelled() => {} + } + } + } + self.stop_child(actor_id, reason).await; + } } #[async_trait] @@ -183,7 +209,7 @@ impl Actor for GameServer { .is_err() { // Unreachable given the duplicate-start check above; defensive. - child.stop(cfg.stop_grace).await; + child.stop(effective_stop_grace()).await; release_child_port(child_port).await; anyhow::bail!("a child for actor {actor_id} is already registered"); } @@ -265,7 +291,7 @@ 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<()> { - self.stop_child(ctx.actor_id(), "actor sleeping").await; + self.drain_then_stop_child(ctx.actor_id(), "actor sleeping").await; Ok(()) } diff --git a/container-runner/src/main.rs b/container-runner/src/main.rs index 9059e10bec..04f5f492ba 100644 --- a/container-runner/src/main.rs +++ b/container-runner/src/main.rs @@ -65,8 +65,6 @@ pub struct RunnerConfig { pub command_template: Vec, /// Default local port for the child when `input.port` is absent. pub default_child_port: u16, - /// SIGTERM→SIGKILL grace period on stop. - pub stop_grace: Duration, /// How long to wait for the child's port to open before failing the start. pub readiness_timeout: Duration, } @@ -117,6 +115,30 @@ static SIGTERM_BUDGET: LazyLock = LazyLock::new(|| { Duration::from_secs(secs) }); +/// How long an engine-initiated pause (sleep, lost, going-away) lets the child +/// keep serving and finish its own work before we force a stop. The child is not +/// signalled during this window; it either exits on its own or is SIGTERM'd at +/// the end. Defaults to 15 minutes. It must fit inside the engine's per-runner +/// `drain_grace_period` or a platform reclaim cuts it short. +static DRAIN_GRACE: LazyLock = LazyLock::new(|| { + let secs = std::env::var("RIVET_DRAIN_GRACE_SECS") + .ok() + .and_then(|value| value.parse::().ok()) + .unwrap_or(900); + Duration::from_secs(secs) +}); + +/// The drain window for an engine pause. See [`DRAIN_GRACE`]. +pub fn drain_grace() -> Duration { + *DRAIN_GRACE +} + +/// The cancellation token that fires on a platform shutdown signal. Used to cut +/// a drain wait short so the platform's SIGTERM→SIGKILL budget is honored. +pub fn exit_token() -> &'static CancellationToken { + &EXIT +} + pub fn runner_config() -> Arc { RUNNER_CONFIG .get() @@ -187,16 +209,12 @@ pub async fn release_child_port(port: u16) { RESERVED_PORTS.remove_async(&port).await; } -/// The SIGTERM→SIGKILL grace to give a child right now: the configured -/// `--stop-grace-secs` normally, capped to the platform budget while the -/// process is shutting down due to an OS signal. +/// The SIGTERM→SIGKILL window for the child: always `SIGTERM_BUDGET`, whatever +/// triggered the stop, so the child never gets a different kill deadline on one +/// path than another. The wait *before* SIGTERM (letting the child exit on its +/// own during an engine pause) is separate; see [`drain_grace`]. pub fn effective_stop_grace() -> Duration { - let grace = runner_config().stop_grace; - if SIGNAL_SHUTDOWN.load(Ordering::Acquire) { - grace.min(*SIGTERM_BUDGET) - } else { - grace - } + *SIGTERM_BUDGET } /// End the process. Only the platform shutdown signal drives this now: actors @@ -276,10 +294,6 @@ struct Args { #[arg(long, env = "RIVET_SERVERLESS_BASE_PATH", default_value = "/api/rivet")] base_path: String, - /// SIGTERM→SIGKILL grace period (seconds) when stopping the child. - #[arg(long, env = "RIVET_STOP_GRACE_SECS", default_value_t = 10)] - stop_grace_secs: u64, - /// How long (seconds) to wait for the child's port to open before failing start. #[arg(long, env = "RIVET_READINESS_TIMEOUT_SECS", default_value_t = 30)] readiness_timeout_secs: u64, @@ -325,12 +339,10 @@ async fn async_main() -> Result<()> { .or_else(|| env_u16("PORT")) .unwrap_or(8080); - let stop_grace = Duration::from_secs(args.stop_grace_secs); RUNNER_CONFIG .set(Arc::new(RunnerConfig { command_template: args.command.clone(), default_child_port: args.child_port, - stop_grace, readiness_timeout: Duration::from_secs(args.readiness_timeout_secs), })) .map_err(|_| anyhow::anyhow!("runner config already set"))?; @@ -341,9 +353,12 @@ async fn async_main() -> Result<()> { ActorConfig { // Game servers hold live in-memory state; never idle-sleep the actor. no_sleep: true, - // The destroy grace deadline must outlast the child's SIGTERM→SIGKILL - // window or core aborts `on_destroy` mid-stop and leaks the child. - sleep_grace_period: stop_grace + Duration::from_secs(10), + // Core force-aborts the stop hook after this deadline. `on_sleep` + // legitimately runs for the whole drain window plus the SIGTERM + // budget, so the deadline must outlast both or core would abort the + // drain mid-flight and leak the child. The extra 5s keeps the + // deadline from racing a hook that finishes right on time. + sleep_grace_period: *DRAIN_GRACE + *SIGTERM_BUDGET + Duration::from_secs(5), sleep_grace_period_overridden: true, ..Default::default() }, @@ -419,7 +434,7 @@ async fn async_main() -> Result<()> { }; tokio::join!(drain, stop_all_children(*SIGTERM_BUDGET)); } else { - stop_all_children(stop_grace).await; + stop_all_children(*SIGTERM_BUDGET).await; runtime.shutdown().await; } serve_shutdown.cancel(); From 94d4dccd09ee75233077857a0e4879a1f0ec206d Mon Sep 17 00:00:00 2001 From: ABCxFF <79597906+abcxff@users.noreply.github.com> Date: Fri, 21 Aug 2026 14:37:51 -0400 Subject: [PATCH 4/9] feat(container-runner): exit process when the last child stops --- container-runner/src/actor.rs | 22 +++++++++++++++------- container-runner/src/main.rs | 21 +++++++++------------ 2 files changed, 24 insertions(+), 19 deletions(-) diff --git a/container-runner/src/actor.rs b/container-runner/src/actor.rs index 800618ca49..7a8d3cea2d 100644 --- a/container-runner/src/actor.rs +++ b/container-runner/src/actor.rs @@ -18,7 +18,7 @@ use crate::child::{ChildProcess, SpawnSpec, log_prefix}; use crate::input::ActorInput; use crate::{ children, drain_grace, effective_stop_grace, exit_token, release_child_port, - reserve_child_port, runner_config, + request_exit, reserve_child_port, runner_config, }; /// Live actor contexts on this instance, keyed by actor id. Lets the process @@ -48,12 +48,20 @@ impl GameServer { release_child_port(child.child_port).await; } - // The instance stays alive and warm after its last actor stops, ready to - // host the next placement. It is reaped by the platform's own shutdown - // signal, not by self-exit. This keeps the serverless container long - // lived enough for the log agent to drain its stderr, which a fast - // self-exit could otherwise lose. - tracing::info!(actor_id = %actor_id, reason, "actor stopped, keeping instance warm"); + // Exit the whole process once the last child on this instance stops. The + // runner is PID 1, so `request_exit` cancels `EXIT`, which wakes `main` + // to run the graceful envoy close and then return, stopping the container + // so the platform reaps it. Guarded on an empty registry so a multi-actor + // instance does not tear down siblings still hosting a child. + if children().is_empty() { + request_exit(actor_id, reason); + } else { + tracing::info!( + actor_id = %actor_id, + reason, + "actor stopped, other actors still running on this instance" + ); + } } /// Engine pause path (sleep, lost, going-away). Give the child up to diff --git a/container-runner/src/main.rs b/container-runner/src/main.rs index 04f5f492ba..5c2bf9bd5f 100644 --- a/container-runner/src/main.rs +++ b/container-runner/src/main.rs @@ -217,10 +217,10 @@ pub fn effective_stop_grace() -> Duration { *SIGTERM_BUDGET } -/// End the process. Only the platform shutdown signal drives this now: actors -/// stopping or failing to start no longer exit the instance, so it stays warm -/// and reusable and its logs have time to drain. The runner is PID 1 in the -/// image, so exiting stops the container and the platform reaps the instance. +/// End the process. Driven by a platform shutdown signal or by the last child on +/// this instance stopping (see `stop_child`). The runner is PID 1 in the image, +/// so cancelling `EXIT` wakes `main` to run the graceful envoy close and return, +/// which stops the container and lets the platform reap the instance. pub fn request_exit(actor_id: &str, reason: &str) { tracing::info!(actor_id = %actor_id, reason, "shutting down container"); EXIT.cancel(); @@ -394,11 +394,7 @@ async fn async_main() -> Result<()> { )); tracing::info!(port, "container-runner serverless front door listening"); - // Wait for an exit request, then tear down. Only the signal path is live - // today: nothing calls `request_exit` except `spawn_signal_handler`, which - // sets `SIGNAL_SHUTDOWN` before cancelling `EXIT`, so the `else` branch is - // currently unreachable and kept only as a fallback for a future - // actor-driven exit. + // Wait for an exit request, then tear down. Two shapes depending on why: // // Signal (platform is reclaiming the instance): kill our children AND // notify the engine at the SAME time, each bounded by the full SIGTERM @@ -408,9 +404,10 @@ async fn async_main() -> Result<()> { // immediately. Bounding the drain means an unreachable engine cannot eat // the budget the children need. // - // Fallback actor-driven exit (unreachable today): no platform deadline. - // Children are already reaped by the hooks (the sweep is a no-op backstop), - // and the runtime drains unbounded so the /start SSE flushes cleanly. + // Actor-driven exit (the last child stopped, so `stop_child` cancelled + // `EXIT`): no platform deadline. The child is already reaped by the hook + // that triggered the exit (the sweep is a no-op backstop), 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 From 0a1e43143b16d7c027571427b47e0a46b11ccc01 Mon Sep 17 00:00:00 2001 From: ABCxFF <79597906+abcxff@users.noreply.github.com> Date: Fri, 21 Aug 2026 14:43:25 -0400 Subject: [PATCH 5/9] chore(container-runner): tighten comments --- container-runner/README.md | 6 +- .../e2e-test/configure-serverless.mjs | 2 +- .../examples/e2e-test/docker-compose.yml | 2 +- .../e2e-test/host/run-host-loadtest.sh | 2 +- .../examples/test-server/Dockerfile | 4 +- .../examples/test-server/server.mjs | 2 +- container-runner/src/actor.rs | 95 ++++------- container-runner/src/child.rs | 23 +-- container-runner/src/input.rs | 17 +- container-runner/src/main.rs | 161 +++++++----------- container-runner/src/monitor.rs | 35 +--- container-runner/src/proxy.rs | 12 +- 12 files changed, 134 insertions(+), 227 deletions(-) diff --git a/container-runner/README.md b/container-runner/README.md index 0a3fda9a8c..377c46ef2a 100644 --- a/container-runner/README.md +++ b/container-runner/README.md @@ -26,8 +26,8 @@ There are two working local paths: 2. Run the built Unity server behind `container-runner`, create a Rivet actor locally, and connect a FishNet client through the local Rivet guard URL. -A production image for **Rivet Compute** (the Cloud Run serverless model) is provided — -see [Rivet Compute](#rivet-compute) below. +A production image for **Rivet Compute** (the serverless model) is provided. See +[Rivet Compute](#rivet-compute) below. ## Project Layout @@ -207,7 +207,7 @@ Knobs: `LOAD_COUNT` (default 25), `LOAD_CONCURRENCY` (default 64). **Running the full 1000:** 1000 local instances is ~2000 processes (a Rust runner + Node child each) and needs a beefy host plus a raised `ulimit -n`. For a true 1000-container run, -point `load-test.mjs` at **Rivet Cloud** instead — Cloud Run scales the containers, no local +point `load-test.mjs` at **Rivet Cloud** instead, which scales the containers with no local limit. Set the engine env and let a single pool auto-scale: ```bash diff --git a/container-runner/examples/e2e-test/configure-serverless.mjs b/container-runner/examples/e2e-test/configure-serverless.mjs index cbec506474..a03ee0a11c 100644 --- a/container-runner/examples/e2e-test/configure-serverless.mjs +++ b/container-runner/examples/e2e-test/configure-serverless.mjs @@ -11,7 +11,7 @@ const body = { // Seconds the engine holds the /start request before draining to a fresh one. request_lifespan: 900, drain_grace_period: 30, - // 1:1 actor<->container mapping (Cloud Run concurrency=1 model). + // 1:1 actor<->container mapping (serverless concurrency=1 model). slots_per_runner: 1, max_runners: 1, max_concurrent_actors: 1, diff --git a/container-runner/examples/e2e-test/docker-compose.yml b/container-runner/examples/e2e-test/docker-compose.yml index c2aaae0f32..74c198107f 100644 --- a/container-runner/examples/e2e-test/docker-compose.yml +++ b/container-runner/examples/e2e-test/docker-compose.yml @@ -1,5 +1,5 @@ # Local end-to-end: self-hosted Rivet engine + the game container (container-runner -# wrapping the Node test server). All x86_64/amd64 to mirror Cloud Run. +# wrapping the Node test server). All x86_64/amd64 to mirror the serverless platform. # # Flow: # create actor (POST /actors) -> engine POSTs /api/rivet/start to game:8080 diff --git a/container-runner/examples/e2e-test/host/run-host-loadtest.sh b/container-runner/examples/e2e-test/host/run-host-loadtest.sh index 8133ca1276..2cde1503cf 100755 --- a/container-runner/examples/e2e-test/host/run-host-loadtest.sh +++ b/container-runner/examples/e2e-test/host/run-host-loadtest.sh @@ -4,7 +4,7 @@ # server), then drive a WebSocket ping-pong through the guard to every one. # # Each container-runner instance is one "container": its own front-door port, its own child -# port, and its own serverless runner pool (load-). This mirrors the Cloud Run 1:1 +# port, and its own serverless runner pool (load-). This mirrors the serverless 1:1 # actor<->container model locally, so `LOAD_COUNT` instances == that many containers. # # engine (:7420 guard, :7421 api) diff --git a/container-runner/examples/test-server/Dockerfile b/container-runner/examples/test-server/Dockerfile index 7274aa4739..24c54e509e 100644 --- a/container-runner/examples/test-server/Dockerfile +++ b/container-runner/examples/test-server/Dockerfile @@ -6,7 +6,7 @@ # Build from the RIVET REPO ROOT so the workspace and examples are in context. # Arch-agnostic: builds for the host/requested platform. # Local (native, e.g. arm64 Mac): docker build -f container-runner/examples/test-server/Dockerfile -t game:latest . -# Cloud Run (x86_64): docker build --platform linux/amd64 -f container-runner/examples/test-server/Dockerfile -t game:amd64 . +# Serverless (x86_64): docker build --platform linux/amd64 -f container-runner/examples/test-server/Dockerfile -t game:amd64 . # # Production pattern (per spec) — instead of building from source, curl a released # binary into an existing image: @@ -33,7 +33,7 @@ COPY --from=builder /usr/local/bin/rivet-container-runner /usr/local/bin/rivet-c COPY container-runner/examples/test-server/ /app/test-server/ RUN cd /app/test-server && npm install --omit=dev -# Cloud Run sets $PORT (the serverless front door). The child listens on CHILD_PORT. +# The serverless platform sets $PORT (the front door). The child listens on CHILD_PORT. ENV PORT=8080 \ CHILD_PORT=7770 \ RIVET_ACTOR_NAME=game diff --git a/container-runner/examples/test-server/server.mjs b/container-runner/examples/test-server/server.mjs index 2822669542..5c81a47d35 100644 --- a/container-runner/examples/test-server/server.mjs +++ b/container-runner/examples/test-server/server.mjs @@ -2,7 +2,7 @@ // // This is the child process that container-runner (rivet-container-runner) spawns inside // the container. It stands in for a real Unity FishNet dedicated server while we validate -// the Rivet -> Cloud Run -> container pipeline. +// the Rivet -> serverless -> container pipeline. // // Normal behavior: // - Binds HTTP+WebSocket on $PORT (default 7770) on 0.0.0.0. diff --git a/container-runner/src/actor.rs b/container-runner/src/actor.rs index 7a8d3cea2d..86b2befbc3 100644 --- a/container-runner/src/actor.rs +++ b/container-runner/src/actor.rs @@ -1,11 +1,8 @@ //! The `GameServer` actor: wraps one child game-server process per actor. //! -//! Lifecycle: `on_start` reserves a port and spawns the child, waiting for -//! readiness (so the actor is never reported ready before the child listens), -//! `run` is a watchdog that reports unexpected child exits, -//! `on_fetch`/`on_websocket` proxy tunneled traffic to the child's port, and -//! `on_destroy` stops the child while the instance stays warm for the next -//! placement. +//! `on_start` reserves a port and spawns the child (waiting for readiness), `run` +//! watchdogs unexpected child exits, `on_fetch`/`on_websocket` proxy tunneled +//! traffic to the child, and `on_sleep`/`on_destroy` stop it. use std::sync::{Arc, LazyLock}; @@ -21,9 +18,8 @@ use crate::{ request_exit, 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. +/// Live actor contexts keyed by actor id, so the shutdown path can report actors +/// as crashed when the platform reclaims the container. static ACTOR_CTXS: LazyLock>> = LazyLock::new(scc::HashMap::new); @@ -32,14 +28,12 @@ pub struct GameServer { } impl GameServer { - /// Shared teardown for sleep and destroy: for a game server the two are - /// materially the same event, because the in-memory match state lives in - /// the child and cannot outlive the container. The launch spec is the - /// persisted actor state, so a later wake respawns an equivalent child. + /// Shared teardown for sleep and destroy. For a game server they are the same + /// event: match state lives in the child and cannot outlive the container, and + /// a later wake respawns an equivalent child from the persisted launch spec. async fn stop_child(&self, actor_id: &str, reason: &str) { - // Remove from the registry FIRST so the watchdog treats the exit as - // deliberate, then stop. `stop` is idempotent if the process shutdown - // sweep already stopped this child. + // Remove from the registry first so the watchdog treats the exit as + // deliberate. `stop` is idempotent if the shutdown sweep already ran. children().remove_async(actor_id).await; ACTOR_CTXS.remove_async(actor_id).await; let child = self.child.lock().await.take(); @@ -48,11 +42,9 @@ impl GameServer { release_child_port(child.child_port).await; } - // Exit the whole process once the last child on this instance stops. The - // runner is PID 1, so `request_exit` cancels `EXIT`, which wakes `main` - // to run the graceful envoy close and then return, stopping the container - // so the platform reaps it. Guarded on an empty registry so a multi-actor - // instance does not tear down siblings still hosting a child. + // Exit the process once the last child stops: `request_exit` cancels + // `EXIT`, waking `main` to close the envoy and return (the runner is + // PID 1). Guarded on an empty registry so siblings survive. if children().is_empty() { request_exit(actor_id, reason); } else { @@ -64,12 +56,9 @@ impl GameServer { } } - /// Engine pause path (sleep, lost, going-away). Give the child up to - /// `DRAIN_GRACE` to finish its in-flight work and exit on its own before we - /// force a stop; the child is not signalled during the window. A natural - /// child exit ends the wait immediately, and a platform SIGTERM (which - /// cancels the exit token) cuts it short so the reclaim's SIGTERM→SIGKILL - /// budget is honored. + /// Engine pause (sleep, lost, going-away): let the child finish and exit on its + /// own for up to `DRAIN_GRACE` before forcing a stop. A child exit or a platform + /// SIGTERM (which cancels the exit token) ends the wait early. async fn drain_then_stop_child(&self, actor_id: &str, reason: &str) { let child = self.child.lock().await.clone(); if let Some(child) = child { @@ -118,10 +107,8 @@ impl Actor for GameServer { let actor_id = ctx.actor_id().to_string(); let key = actor_key_string(&ctx); - // Surface the resource monitor's status here, tagged with the actor id, so - // it is visible in actor-scoped log views. The monitor's own enable/disable - // logs are process-level and have no actor id, so they are filtered out of - // those views. + // Tagged with the actor id so it shows in actor-scoped log views; the + // monitor's own process-level enable/disable logs are filtered out there. tracing::info!( actor_id = %actor_id, resource_monitor_enabled = crate::monitor::enabled(), @@ -129,9 +116,8 @@ impl Actor for GameServer { "resource monitor status" ); - // 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. + // An engine retry for an already-running actor must be an idempotent no-op; + // rejecting it would make the engine tear down a healthy actor. if let Some(existing) = children().read_async(&actor_id, |_, c| c.clone()).await { if !existing.has_exited() { println!( @@ -172,9 +158,8 @@ impl Actor for GameServer { key: key.clone(), }; - // Version line, tagged with the actor id so it is visible in actor-scoped - // logs. `git_sha` is omitted entirely when unknown rather than logged as - // "unknown". + // Tagged with the actor id for actor-scoped logs; `git_sha` is omitted when + // unknown rather than logged as "unknown". match crate::git_sha() { Some(git_sha) => tracing::info!( actor_id = %actor_id, @@ -200,17 +185,13 @@ impl Actor for GameServer { Ok(child) => Arc::new(child), Err(err) => { release_child_port(child_port).await; - // A failed start is this actor's alone and does not take the - // instance down. The container stays warm and ready for the next - // placement, and stays alive long enough for the log agent to - // drain the failure logs before the platform reaps it. + // A failed start is this actor's alone; it does not take down others. return Err(err); } }; - // The global registry lets the process shutdown path stop children - // even when actor hooks never run, and arbitrates the deliberate-stop - // vs unexpected-exit race for the watchdog in `run`. + // The global registry lets the shutdown path stop children when hooks never + // run, and arbitrates the deliberate-stop vs unexpected-exit race in `run`. if children() .insert_async(actor_id.clone(), child.clone()) .await @@ -221,19 +202,16 @@ 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 only after startup succeeds; a failed start never runs a stop + // hook to remove the entry, so registering earlier would leak it. register_ctx(&actor_id, &ctx).await; *self.child.lock().await = Some(child); Ok(()) } - /// Watchdog: waits for the child to exit. Deliberate stops remove the - /// child from the global registry first, so winning the `remove` race - /// means the exit was unexpected and the actor must be torn down. A clean - /// exit (code 0) destroys the actor; any other exit returns an error so - /// the framework reports an errored stop and the engine records the crash. + /// Watchdog for the child exiting. Deliberate stops remove it from the registry + /// first, so winning the `remove` race means the exit was unexpected: a clean + /// exit destroys the actor, any other reports an errored stop (a crash). async fn run(self: Arc, ctx: Ctx) -> Result<()> { let Some(child) = self.child.lock().await.clone() else { anyhow::bail!("run: child process was never spawned"); @@ -294,10 +272,8 @@ impl Actor for GameServer { crate::proxy::ws_proxy(child_port, path, ws).await } - /// Engine-initiated sleep. `no_sleep` suppresses idle sleep, but the - /// engine can still sleep an actor (dashboard, crash policy, eviction - /// ahead of instance retirement); leaving the child running would orphan - /// it on an instance the engine considers vacated. + /// Engine-initiated sleep. `no_sleep` blocks only idle sleep; the engine can + /// still sleep an actor (dashboard, crash policy, eviction), so we stop the child. async fn on_sleep(self: Arc, ctx: Ctx) -> Result<()> { self.drain_then_stop_child(ctx.actor_id(), "actor sleeping").await; Ok(()) @@ -318,10 +294,9 @@ async fn register_ctx(actor_id: &str, ctx: &Ctx) { .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. +/// Report every live actor as crashed. Called when the platform reclaims the +/// container (unexpected SIGTERM), while the envoy is still connected, so the +/// reclaim surfaces as a crash on the engine instead of a silent reallocation. pub async fn crash_all_actors(message: &str) { let mut ctxs = Vec::new(); ACTOR_CTXS diff --git a/container-runner/src/child.rs b/container-runner/src/child.rs index c9d5843ae6..907fc2ffbb 100644 --- a/container-runner/src/child.rs +++ b/container-runner/src/child.rs @@ -1,9 +1,8 @@ //! Child game-server process management: spawn, log piping, readiness, SIGTERM stop. //! -//! Ownership model: a dedicated "reaper" task exclusively owns the `tokio::process::Child` -//! and awaits its exit, publishing the result on a `watch` channel. `stop()` and readiness -//! checks signal/observe via the pid and the watch channel, so they never contend for the -//! child handle (which would deadlock against the reaper's long-lived `wait()`). +//! A dedicated reaper task owns the `tokio::process::Child` and publishes its exit on a +//! `watch` channel; `stop()` and readiness checks signal/observe via the pid and channel, +//! so they never contend for the child handle (which would deadlock the reaper's `wait()`). use std::collections::HashMap; use std::net::Ipv4Addr; @@ -72,12 +71,9 @@ impl ChildProcess { let prefix = log_prefix(&actor_id, key.as_deref()); - // Guarantee the child port is free BEFORE spawning. Otherwise a stale child from a - // prior start (in a reused container instance) still holding the port would make - // `wait_until_ready` below false-positive: it connects to the OLD listener and - // reports the NEW child "ready" even though the new child failed to bind - // (`Address already in use`) and is dead. Refuse the start with a clear diagnostic - // instead — this container hosts exactly one game server on a fixed port. + // Refuse to spawn if the port is already taken. A stale child from a prior start + // still holding it would make `wait_until_ready` false-positive on the OLD listener + // while the new child dies with `Address already in use`. if TcpStream::connect((Ipv4Addr::LOCALHOST, child_port)) .await .is_ok() @@ -85,7 +81,7 @@ impl ChildProcess { anyhow::bail!( "child port {child_port} is already in use before spawning `{program}`: a \ previous game server is still running in this container. container-runner \ - hosts one actor per container — configure the serverless runner with \ + hosts one actor per container; configure the serverless runner with \ max_concurrent_actors=1 and platform request concurrency=1." ); } @@ -143,9 +139,8 @@ impl ChildProcess { exited_rx, }; - // If the child crashes before opening its port, or never opens it, make sure we - // don't leave it running: kill it before surfacing the start failure. (The reaper - // task owns the tokio Child, so dropping `this` alone would NOT kill a hung child.) + // Kill the child before surfacing a readiness failure; dropping `this` alone would + // not (the reaper owns the tokio Child), leaving a hung child running. if let Err(err) = this.wait_until_ready(readiness_timeout).await { this.stop(Duration::from_secs(2)).await; return Err(err); diff --git a/container-runner/src/input.rs b/container-runner/src/input.rs index 1b39d3c089..c39fb6d0bd 100644 --- a/container-runner/src/input.rs +++ b/container-runner/src/input.rs @@ -1,19 +1,16 @@ //! The actor input payload describing how to launch the child game server. //! -//! 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 command, args, env, and port are carried in the actor's create-time `input` +//! (CBOR per RivetKit); anything omitted falls back to the CLI template +//! (`rivet-container-runner -- `). This is also the actor's persisted +//! state, so a woken actor restores the same launch spec. 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. +/// Shape of the actor `input` payload. Unknown fields are ignored, not rejected: +/// this is also the persisted state, and a strict decode would break waking actors +/// after a rollback to a binary predating a new field. #[derive(Debug, Default, Serialize, Deserialize)] pub struct ActorInput { /// Overrides the CLI command template entirely (program + fixed args). diff --git a/container-runner/src/main.rs b/container-runner/src/main.rs index 5c2bf9bd5f..d3ddcdff93 100644 --- a/container-runner/src/main.rs +++ b/container-runner/src/main.rs @@ -1,20 +1,14 @@ -//! `rivet-container-runner` — a RivetKit serverless app that hosts a single -//! actor by spawning a child game-server process and proxying Rivet's -//! tunneled traffic to it. +//! `rivet-container-runner`: a RivetKit serverless app that hosts actors by +//! spawning one child game-server process per actor and proxying Rivet's tunneled +//! traffic to it. //! -//! Serverless model: the engine's `POST /api/rivet/start` starts this -//! container (served by rivetkit-core's serverless runtime). On actor start -//! the `GameServer` actor spawns the child (`-- `), pipes its -//! logs to stdout prefixed with the actor id + key, and proxies inbound -//! HTTP/WebSocket (arriving over Rivet's tunnel) to the child's local port. -//! On actor stop it SIGTERMs the child. -//! -//! The runner hosts as many concurrent actors as the engine places on it, -//! each with its own child process on its own port; the pool's request -//! concurrency decides how many that is (1 in the recommended game-server -//! setup). The instance stays warm after its last actor stops and never -//! self-exits; the engine reaps it by draining the `/start` connection once -//! the request lifespan elapses, or the platform sends a SIGTERM. +//! The engine's `POST /api/rivet/start` boots this container. On actor start the +//! `GameServer` actor spawns the child (`-- `), pipes its logs to +//! stdout prefixed with the actor id + key, and proxies inbound HTTP/WebSocket to +//! the child's local port. On actor stop it SIGTERMs the child; once the last +//! child stops the process exits so the platform reaps the instance. The engine +//! decides how many actors land here, each on its own port (1 in the recommended +//! game-server setup). mod actor; mod child; @@ -46,9 +40,9 @@ pub(crate) const VERSION: &str = env!("CARGO_PKG_VERSION"); /// image build context excludes `.git`). pub(crate) const GIT_SHA: &str = env!("CONTAINER_RUNNER_GIT_SHA"); -/// The effective git SHA, or `None` when unknown. Prefers the build-time SHA and -/// falls back to a runtime `OVERRIDE_GIT_SHA` env var, so a deploy that cannot -/// inject a build arg can still surface it via an environment variable. +/// The effective git SHA, or `None` when unknown. Prefers the build-time SHA, +/// falling back to the `OVERRIDE_GIT_SHA` env var for deploys that cannot inject a +/// build arg. pub(crate) fn git_sha() -> Option { if GIT_SHA != "unknown" { return Some(GIT_SHA.to_string()); @@ -73,39 +67,31 @@ pub struct RunnerConfig { // configuration is ambient process state set once in `main`. static RUNNER_CONFIG: OnceLock> = OnceLock::new(); -/// Running children keyed by actor id. Owned globally (not only by actors) so -/// the process shutdown path can stop children even if actor hooks never run, -/// and so the watchdog and stop paths can arbitrate who reports an exit. +/// Running children keyed by actor id. Global (not per-actor) so the shutdown path +/// can stop children even when hooks never run, and the watchdog can arbitrate exits. static CHILDREN: LazyLock>> = LazyLock::new(scc::HashMap::new); -/// Child ports currently reserved by a spawning or running child. Multiple -/// actors may run concurrently (engine placement decides how many land here), -/// so each child needs its own port and concurrent starts must not race to -/// the same one. +/// Ports reserved by spawning or running children. Multiple actors can run here at +/// once, so each needs its own port and concurrent starts must not race for one. static RESERVED_PORTS: LazyLock> = LazyLock::new(scc::HashSet::new); -/// Cancelled to bring the whole process down (actor stopped, failed start, or -/// signal). `main` owns the exit sequencing. +/// Cancelled to bring the whole process down (last child stopped or a signal). +/// `main` owns the exit sequencing. static EXIT: LazyLock = LazyLock::new(CancellationToken::new); -/// Set when the process is shutting down because the PLATFORM sent a signal. -/// The hosting platform gives a container only a bounded window (often ~10 -/// seconds) between SIGTERM and SIGKILL, so every grace period on this path must -/// fit that budget; engine-initiated stops keep the full configured grace -/// (their budget is the pool's drain grace period instead). +/// Set when a platform signal is driving shutdown. The platform gives only a +/// bounded window (~10s) between SIGTERM and SIGKILL, so grace periods on this +/// path must fit that budget. 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. +/// Set when shutdown was a platform SIGTERM (instance reclaim) rather than a local +/// SIGINT (Ctrl-C), so the reclaim can be reported as an actor crash. static PLATFORM_RECLAIM: AtomicBool = AtomicBool::new(false); -/// How long the platform gives this container between SIGTERM and SIGKILL. -/// Defaults to 9s, one second under the common ~10s platform budget so the whole -/// teardown lands before SIGKILL; keep it in sync with the platform's actual -/// budget via RIVET_SIGTERM_BUDGET_SECS. On the signal path the engine drain and -/// the child kills run concurrently, each bounded by this full budget. +/// SIGTERM→SIGKILL window the platform gives this container. Defaults to 9s, just +/// under the common ~10s budget; override with RIVET_SIGTERM_BUDGET_SECS. On the +/// signal path the engine drain and the child kills share this budget concurrently. static SIGTERM_BUDGET: LazyLock = LazyLock::new(|| { let secs = std::env::var("RIVET_SIGTERM_BUDGET_SECS") .ok() @@ -115,11 +101,9 @@ static SIGTERM_BUDGET: LazyLock = LazyLock::new(|| { Duration::from_secs(secs) }); -/// How long an engine-initiated pause (sleep, lost, going-away) lets the child -/// keep serving and finish its own work before we force a stop. The child is not -/// signalled during this window; it either exits on its own or is SIGTERM'd at -/// the end. Defaults to 15 minutes. It must fit inside the engine's per-runner -/// `drain_grace_period` or a platform reclaim cuts it short. +/// How long an engine pause (sleep, lost, going-away) lets the child keep serving +/// and exit on its own before we SIGTERM it. Defaults to 15 min (RIVET_DRAIN_GRACE_SECS); +/// must fit inside the engine's per-runner `drain_grace_period` or a reclaim cuts it short. static DRAIN_GRACE: LazyLock = LazyLock::new(|| { let secs = std::env::var("RIVET_DRAIN_GRACE_SECS") .ok() @@ -133,8 +117,8 @@ pub fn drain_grace() -> Duration { *DRAIN_GRACE } -/// The cancellation token that fires on a platform shutdown signal. Used to cut -/// a drain wait short so the platform's SIGTERM→SIGKILL budget is honored. +/// Token that fires on a platform shutdown signal. Cuts a drain wait short so the +/// platform's SIGTERM→SIGKILL budget is honored. pub fn exit_token() -> &'static CancellationToken { &EXIT } @@ -164,11 +148,9 @@ pub async fn active_actor_ids() -> Vec { ids } -/// Reserve a local port for a new child. An explicit `input.port` is honored -/// or refused if another child holds it; otherwise the first free port at or -/// above the CLI default is picked. The reservation guards the window between -/// port selection and the child actually binding; release it via -/// [`release_child_port`] once the child is gone. +/// Reserve a local port for a new child. An explicit `input.port` is honored (or +/// refused if held); otherwise the first free port at or above the default is used. +/// Guards selection-to-bind; release via [`release_child_port`] once the child is gone. pub async fn reserve_child_port(preferred: Option, default: u16) -> Result { if let Some(port) = preferred { if RESERVED_PORTS.insert_async(port).await.is_err() { @@ -210,31 +192,28 @@ pub async fn release_child_port(port: u16) { } /// The SIGTERM→SIGKILL window for the child: always `SIGTERM_BUDGET`, whatever -/// triggered the stop, so the child never gets a different kill deadline on one -/// path than another. The wait *before* SIGTERM (letting the child exit on its -/// own during an engine pause) is separate; see [`drain_grace`]. +/// triggered the stop. The wait before SIGTERM (an engine pause) is separate; see +/// [`drain_grace`]. pub fn effective_stop_grace() -> Duration { *SIGTERM_BUDGET } -/// End the process. Driven by a platform shutdown signal or by the last child on -/// this instance stopping (see `stop_child`). The runner is PID 1 in the image, -/// so cancelling `EXIT` wakes `main` to run the graceful envoy close and return, -/// which stops the container and lets the platform reap the instance. +/// End the process. Driven by a platform signal or by the last child stopping (see +/// `stop_child`). The runner is PID 1, so cancelling `EXIT` wakes `main` to run the +/// graceful envoy close and return, which stops the container. pub fn request_exit(actor_id: &str, reason: &str) { tracing::info!(actor_id = %actor_id, reason, "shutting down container"); EXIT.cancel(); } -/// Stable per-process identifier, generated once. The runner is PID 1 in the -/// image, so one boot id == one container instance. Logged at startup and on -/// every actor start so an actor can be attributed to an instance (the log -/// stream carries no instance id). +/// Stable per-process id, generated once. The runner is PID 1, so one boot id == +/// one container instance. Logged per actor start so an actor can be attributed to +/// an instance (the log stream carries no instance id). pub fn boot_id() -> &'static str { static BOOT_ID: OnceLock = OnceLock::new(); BOOT_ID.get_or_init(|| { - // 9 random bytes -> 12 chars. Falls back to a fixed marker if the - // CSPRNG is unavailable, which would itself be worth seeing in logs. + // 9 random bytes -> 12 chars. Falls back to a marker if the CSPRNG is + // unavailable, which is itself worth seeing in logs. let mut buf = [0u8; 9]; match std::fs::File::open("/dev/urandom").and_then(|mut f| f.read_exact(&mut buf)) { Ok(()) => base64url_nopad(&buf), @@ -272,8 +251,7 @@ fn base64url_nopad(input: &[u8]) -> String { long_about = None, )] struct Args { - /// Serverless HTTP front-door port. Rivet Compute injects RIVET_PORT (other - /// serverless platforms use the conventional PORT); resolved in `main` as + /// Serverless HTTP front-door port. Resolved in `main` as /// --port > RIVET_PORT > PORT > 8080. #[arg(long)] port: Option, @@ -332,7 +310,7 @@ async fn async_main() -> Result<()> { let boot_id = boot_id(); tracing::info!(?args, %boot_id, "starting container-runner"); - // Front-door port: Rivet Compute injects RIVET_PORT; other serverless platforms use the conventional PORT. + // Front-door port: Rivet Compute injects RIVET_PORT; other platforms use PORT. let port = args .port .or_else(|| env_u16("RIVET_PORT")) @@ -353,11 +331,9 @@ async fn async_main() -> Result<()> { ActorConfig { // Game servers hold live in-memory state; never idle-sleep the actor. no_sleep: true, - // Core force-aborts the stop hook after this deadline. `on_sleep` - // legitimately runs for the whole drain window plus the SIGTERM - // budget, so the deadline must outlast both or core would abort the - // drain mid-flight and leak the child. The extra 5s keeps the - // deadline from racing a hook that finishes right on time. + // Core force-aborts the stop hook at this deadline, so it must outlast the + // full drain window plus the SIGTERM budget (with a small margin) or the + // drain is cut short and the child leaks. sleep_grace_period: *DRAIN_GRACE + *SIGTERM_BUDGET + Duration::from_secs(5), sleep_grace_period_overridden: true, ..Default::default() @@ -394,27 +370,14 @@ async fn async_main() -> Result<()> { )); tracing::info!(port, "container-runner serverless front door listening"); - // Wait for an exit request, then tear down. Two shapes depending on why: - // - // Signal (platform is reclaiming the instance): kill our children AND - // notify the engine at the SAME time, each bounded by the full SIGTERM - // budget. The direct sweep is what guarantees children die within budget - // rather than waiting on an engine round-trip to run our on_destroy hooks; - // notifying the engine in parallel just lets it start re-placing actors - // immediately. Bounding the drain means an unreachable engine cannot eat - // the budget the children need. - // - // Actor-driven exit (the last child stopped, so `stop_child` cancelled - // `EXIT`): no platform deadline. The child is already reaped by the hook - // that triggered the exit (the sweep is a no-op backstop), and the runtime - // drains unbounded so the /start SSE flushes cleanly. + // Wait for an exit request, then tear down. A platform reclaim (signal) kills children + // and notifies the engine at once, each bounded by the SIGTERM budget. An actor-driven + // exit (last child stopped) has no deadline: the child is already reaped, drain unbounded. 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. + // A platform SIGTERM reclaims this instance. Report actors as crashed while + // the envoy is still connected so the reclaim (OOM or the ~60 min request + // cap) surfaces as a crash, not a silent reallocation. SIGINT drains cleanly. if PLATFORM_RECLAIM.load(Ordering::Acquire) { crate::actor::crash_all_actors( "runner received unexpected platform SIGTERM, likely OOM or running longer than 60 minutes", @@ -446,11 +409,9 @@ async fn async_main() -> Result<()> { Ok(()) } -/// Stop every child still in the registry. Actor `on_destroy` normally reaps -/// its own child first; this is the belt-and-suspenders sweep for the signal -/// path so children are never orphaned. Children are stopped concurrently so -/// each gets the full `grace` within the SIGTERM budget instead of queueing -/// behind the others. +/// Stop every child still in the registry, concurrently so each gets the full +/// `grace`. Actor hooks normally reap their own child first; this is the sweep for +/// the signal path so children are never orphaned. async fn stop_all_children(grace: Duration) { let mut children: Vec> = Vec::new(); CHILDREN @@ -482,8 +443,8 @@ fn spawn_signal_handler() { tokio::select! { _ = sigterm.recv() => { PLATFORM_RECLAIM.store(true, Ordering::Release); - // Attribute the reclaim to each running actor so it is visible in - // actor-scoped logs, not only the process-level log stream. + // Attribute the reclaim to each running actor so it shows in + // actor-scoped logs, not only the process-level stream. let mut actor_ids = Vec::new(); CHILDREN .retain_async(|actor_id, _| { diff --git a/container-runner/src/monitor.rs b/container-runner/src/monitor.rs index 3f9afea9d9..a5cbc88598 100644 --- a/container-runner/src/monitor.rs +++ b/container-runner/src/monitor.rs @@ -1,33 +1,14 @@ //! Periodic instance resource monitor. //! -//! Opt-in via the [`ENABLE_ENV`] environment variable. When enabled it samples -//! memory and CPU usage every [`sample_interval`] and logs them, so memory -//! growth toward the limit (and the OOM that follows) is visible in the logs at -//! fine granularity. Disabled by default so nothing is logged unless explicitly -//! turned on. +//! Opt-in via [`ENABLE_ENV`]. When on, it samples memory and CPU every +//! [`sample_interval`] and logs them so growth toward the limit (and the OOM that +//! follows) is visible at fine granularity. //! -//! Memory and CPU are detected independently, because the gVisor sandbox -//! exposes cgroup v1 memory but no cgroup v2 or cgroup v1 CPU accounting, so the -//! two counters legitimately come from different sources. -//! -//! Only memory *usage* is reported, not a limit or percentage: under gVisor both -//! the cgroup `memory.limit_in_bytes` and `/proc/meminfo` report the sandbox size -//! rather than the container's configured limit, so any percentage would be -//! misleading. -//! -//! Memory sources, in preference order: -//! - **cgroup v2** `memory.current` (real Linux). Exact. -//! - **cgroup v1** `memory/memory.usage_in_bytes` (gVisor sandbox). -//! Container-wide usage (all processes plus page cache). -//! - **`/proc/meminfo`** last resort. Under gVisor this reflects the whole -//! sandbox, not the container, so it is only an approximation. -//! -//! CPU sources, in preference order: -//! - **cgroup v2** `cpu.stat`. -//! - **`/proc/stat`**, the gVisor sandbox fallback (sandbox-wide, approximate). -//! -//! If neither a memory nor a CPU source is readable the monitor logs once and -//! disables itself. +//! Memory and CPU are detected independently (see [`MemSource`]/[`CpuSource`]): the +//! gVisor sandbox exposes cgroup v1 memory but no cgroup v2 or v1 CPU, so the two +//! counters can come from different sources. Only memory *usage* is reported, not a +//! limit or percentage, since under gVisor the reported limit is the sandbox size, +//! not the container's. If neither counter is readable the monitor disables itself. use std::time::{Duration, Instant}; diff --git a/container-runner/src/proxy.rs b/container-runner/src/proxy.rs index 983b2aaedc..62d5c0461e 100644 --- a/container-runner/src/proxy.rs +++ b/container-runner/src/proxy.rs @@ -1,9 +1,8 @@ //! Bridges Rivet's decoded tunnel traffic to the child game server's local port. //! //! The rivetkit runtime reassembles tunnel frames into decoded `Request`s and -//! `WebSocket` streams (see `Actor::on_fetch` / `::on_websocket`). This module -//! forwards those to `127.0.0.1:` and back — the glue the runtime -//! deliberately leaves to the actor. +//! `WebSocket` streams (`Actor::on_fetch`/`::on_websocket`); this module forwards +//! them to `127.0.0.1:` and back, the glue the runtime leaves to the actor. use std::collections::HashMap; use std::sync::Arc; @@ -74,10 +73,9 @@ enum ClientEvent { Close(u16, String), } -/// Cap on client frames buffered toward a slow child. The old async pump -/// applied tunnel backpressure by awaiting the child sink; the sync message -/// callback cannot await, so a bounded queue plus a loud connection failure -/// on overflow replaces silent unbounded buffering. +/// Cap on client frames buffered toward a slow child. The sync message callback +/// cannot await for backpressure, so overflow fails the connection loudly instead +/// of buffering unbounded. const CLIENT_TO_CHILD_QUEUE: usize = 1024; /// Dial the child's WebSocket endpoint and pump frames in both directions: From 7f1bfe8743bd102f74685d46416f5b930df7a10e Mon Sep 17 00:00:00 2001 From: ABCxFF <79597906+abcxff@users.noreply.github.com> Date: Fri, 21 Aug 2026 15:29:16 -0400 Subject: [PATCH 6/9] feat(container-runner): sleep on startup idle timeout --- container-runner/src/actor.rs | 48 +++++++++++++++++++++++++++++++++-- container-runner/src/main.rs | 16 ++++++++++++ 2 files changed, 62 insertions(+), 2 deletions(-) diff --git a/container-runner/src/actor.rs b/container-runner/src/actor.rs index 86b2befbc3..4d1e01d73e 100644 --- a/container-runner/src/actor.rs +++ b/container-runner/src/actor.rs @@ -4,6 +4,7 @@ //! watchdogs unexpected child exits, `on_fetch`/`on_websocket` proxy tunneled //! traffic to the child, and `on_sleep`/`on_destroy` stop it. +use std::sync::atomic::{AtomicU8, Ordering}; use std::sync::{Arc, LazyLock}; use anyhow::{Context, Result}; @@ -14,7 +15,7 @@ use tokio::sync::Mutex as TokioMutex; use crate::child::{ChildProcess, SpawnSpec, log_prefix}; use crate::input::ActorInput; use crate::{ - children, drain_grace, effective_stop_grace, exit_token, release_child_port, + children, drain_grace, effective_stop_grace, exit_token, idle_timeout, release_child_port, request_exit, reserve_child_port, runner_config, }; @@ -23,8 +24,16 @@ use crate::{ static ACTOR_CTXS: LazyLock>> = LazyLock::new(scc::HashMap::new); +/// One-shot idle lifecycle for a generation, in a single atomic. The startup timer +/// sleeps the actor only while it is `ARMED`; a request moves it to `REQUESTED` so the +/// timer will not sleep. +const IDLE_ARMED: u8 = 0; +const IDLE_REQUESTED: u8 = 1; + pub struct GameServer { child: TokioMutex>>, + /// One-shot idle state: `IDLE_ARMED` / `IDLE_REQUESTED`. + idle_state: AtomicU8, } impl GameServer { @@ -77,6 +86,31 @@ impl GameServer { } self.stop_child(actor_id, reason).await; } + + /// Arm the one-shot startup idle timer when [`idle_timeout`] is set. After the + /// window, if no request has arrived, ask the actor to sleep (`stop_child` then + /// exits the container). Cancelled early if the actor starts shutting down. + fn arm_idle_timeout(self: &Arc, ctx: &Ctx, actor_id: String) { + let Some(timeout) = idle_timeout() else { + return; + }; + let this = self.clone(); + let ctx = ctx.clone(); + tokio::spawn(async move { + let abort = ctx.abort_signal(); + tokio::select! { + _ = tokio::time::sleep(timeout) => {} + _ = abort.cancelled() => return, + } + if this.idle_state.load(Ordering::Relaxed) == IDLE_REQUESTED { + return; + } + tracing::info!(actor_id = %actor_id, ?timeout, "no request within idle timeout, sleeping"); + if let Err(err) = ctx.sleep() { + tracing::debug!(error = ?err, actor_id = %actor_id, "idle sleep request failed"); + } + }); + } } #[async_trait] @@ -99,6 +133,7 @@ impl Actor for GameServer { async fn create(_ctx: &Ctx) -> Result { Ok(Self { child: TokioMutex::new(None), + idle_state: AtomicU8::new(IDLE_ARMED), }) } @@ -206,6 +241,7 @@ impl Actor for GameServer { // hook to remove the entry, so registering earlier would leak it. register_ctx(&actor_id, &ctx).await; *self.child.lock().await = Some(child); + self.arm_idle_timeout(&ctx, actor_id); Ok(()) } @@ -241,6 +277,7 @@ impl Actor for GameServer { } async fn on_fetch(self: Arc, ctx: Ctx, req: Request) -> Result { + self.idle_state.store(IDLE_REQUESTED, Ordering::Relaxed); let child_port = self .child .lock() @@ -257,6 +294,7 @@ impl Actor for GameServer { ws: WebSocket, req: Request, ) -> Result<()> { + self.idle_state.store(IDLE_REQUESTED, Ordering::Relaxed); let child_port = self .child .lock() @@ -274,8 +312,14 @@ impl Actor for GameServer { /// Engine-initiated sleep. `no_sleep` blocks only idle sleep; the engine can /// still sleep an actor (dashboard, crash policy, eviction), so we stop the child. + /// In idle-timeout mode the sleep is (treated as) an idle sleep, so it skips the + /// drain and stops promptly; otherwise it drains for in-flight work first. async fn on_sleep(self: Arc, ctx: Ctx) -> Result<()> { - self.drain_then_stop_child(ctx.actor_id(), "actor sleeping").await; + if idle_timeout().is_some() { + self.stop_child(ctx.actor_id(), "actor sleeping (idle)").await; + } else { + self.drain_then_stop_child(ctx.actor_id(), "actor sleeping").await; + } Ok(()) } diff --git a/container-runner/src/main.rs b/container-runner/src/main.rs index d3ddcdff93..10caa1dde8 100644 --- a/container-runner/src/main.rs +++ b/container-runner/src/main.rs @@ -117,6 +117,22 @@ pub fn drain_grace() -> Duration { *DRAIN_GRACE } +/// One-shot startup idle timeout. If the actor receives no request within this +/// window of starting, it sleeps (and the container exits). `None` when +/// RIVET_IDLE_TIMEOUT_SECS is unset or 0 (disabled); the first request disarms it. +static IDLE_TIMEOUT: LazyLock> = LazyLock::new(|| { + let secs = std::env::var("RIVET_IDLE_TIMEOUT_SECS") + .ok() + .and_then(|value| value.parse::().ok()) + .unwrap_or(0); + (secs > 0).then(|| Duration::from_secs(secs)) +}); + +/// The one-shot startup idle-sleep window, or `None` when disabled. See [`IDLE_TIMEOUT`]. +pub fn idle_timeout() -> Option { + *IDLE_TIMEOUT +} + /// Token that fires on a platform shutdown signal. Cuts a drain wait short so the /// platform's SIGTERM→SIGKILL budget is honored. pub fn exit_token() -> &'static CancellationToken { From a3a00976553097529e71a0723d6ef2b4cafa7545 Mon Sep 17 00:00:00 2001 From: ABCxFF <79597906+abcxff@users.noreply.github.com> Date: Fri, 21 Aug 2026 15:50:17 -0400 Subject: [PATCH 7/9] feat(container-runner): self-destroy on repeated actor start --- Cargo.lock | 1 + container-runner/Cargo.toml | 3 + .../examples/e2e-test/reject-clean.mjs | 21 ++++ container-runner/src/actor.rs | 99 ++++++++++++++++--- container-runner/src/input.rs | 27 +++-- container-runner/src/main.rs | 30 ++++++ container-runner/tests/inline/input.rs | 41 ++++++++ 7 files changed, 204 insertions(+), 18 deletions(-) create mode 100644 container-runner/examples/e2e-test/reject-clean.mjs diff --git a/Cargo.lock b/Cargo.lock index 7c96631189..ce6b296caa 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5317,6 +5317,7 @@ version = "2.3.11" dependencies = [ "anyhow", "async-trait", + "ciborium", "clap", "futures-util", "nix 0.30.1", diff --git a/container-runner/Cargo.toml b/container-runner/Cargo.toml index e01f51bc66..5ecd8d93d1 100644 --- a/container-runner/Cargo.toml +++ b/container-runner/Cargo.toml @@ -28,3 +28,6 @@ tokio-tungstenite.workspace = true tokio-util = { workspace = true, features = ["rt"] } tracing.workspace = true tracing-subscriber.workspace = true + +[dev-dependencies] +ciborium.workspace = true diff --git a/container-runner/examples/e2e-test/reject-clean.mjs b/container-runner/examples/e2e-test/reject-clean.mjs new file mode 100644 index 0000000000..6f197d1888 --- /dev/null +++ b/container-runner/examples/e2e-test/reject-clean.mjs @@ -0,0 +1,21 @@ +// Clean reject-second-start test. Checks destroy_ts (the real delete signal), +// NOT sleep_ts (which an explicit sleep would set). Single wake, long poll. +import { encode } from "cbor-x"; +const NS=process.env.NS, TOK=process.env.TOK, origin="https://api.rivet.dev"; +const H={authorization:`Bearer ${TOK}`,"content-type":"application/json"}; +const sleep=ms=>new Promise(r=>setTimeout(r,ms)); +const now=()=>new Date().toISOString().slice(11,19); +const KEY="rj-"+Date.now(); +const input=Buffer.from(encode({port:7770})).toString("base64"); +let r=await fetch(`${origin}/actors?namespace=${NS}`,{method:"POST",headers:H,body:JSON.stringify({name:"game",key:KEY,input,runner_name_selector:"default",crash_policy:"destroy"})}); +const id=(await r.json()).actor?.actor_id; console.log(now(),"created",id); +async function gw(){try{const x=await fetch(`${origin}/gateway/${id}@${TOK}/`);return x.status;}catch(e){return 0;}} +async function st(){const s=await fetch(`${origin}/actors?actor_ids=${id}&namespace=${NS}`,{headers:H});const a=(await s.json()).actors?.[0];return a?{sleep_ts:a.sleep_ts,destroy_ts:a.destroy_ts}:{missing:true};} +for(let i=0;i<15;i++){if(await gw()===200)break;await sleep(700);} +console.log(now(),"started; wait 4s so started_once persists (grace=3s)"); await sleep(4000); +console.log(now(),"explicit sleep to force the second-start scenario"); await fetch(`${origin}/actors/${id}/sleep?namespace=${NS}`,{method:"POST",headers:H,body:"{}"}); +await sleep(8000); +console.log(now(),"single wake -> second start -> reject -> ctx.destroy(); polling destroy_ts 60s..."); +await gw(); +for(let i=0;i<30;i++){await sleep(2000);const s=await st();console.log(now(),JSON.stringify(s));if(s.destroy_ts){console.log(now(),"=> DELETED (reject ctx.destroy works)");break;}if(s.missing){console.log(now(),"=> GONE");break;}} +console.log(now(),"DONE id="+id); diff --git a/container-runner/src/actor.rs b/container-runner/src/actor.rs index 4d1e01d73e..9d5a9c4aa4 100644 --- a/container-runner/src/actor.rs +++ b/container-runner/src/actor.rs @@ -4,7 +4,7 @@ //! watchdogs unexpected child exits, `on_fetch`/`on_websocket` proxy tunneled //! traffic to the child, and `on_sleep`/`on_destroy` stop it. -use std::sync::atomic::{AtomicU8, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU8, Ordering}; use std::sync::{Arc, LazyLock}; use anyhow::{Context, Result}; @@ -13,10 +13,10 @@ 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::input::{ActorInput, ActorState}; use crate::{ - children, drain_grace, effective_stop_grace, exit_token, idle_timeout, release_child_port, - request_exit, reserve_child_port, runner_config, + children, drain_grace, effective_stop_grace, exit_token, idle_timeout, reject_second_start, + release_child_port, request_exit, reserve_child_port, runner_config, second_start_grace, }; /// Live actor contexts keyed by actor id, so the shutdown path can report actors @@ -34,6 +34,9 @@ pub struct GameServer { child: TokioMutex>>, /// One-shot idle state: `IDLE_ARMED` / `IDLE_REQUESTED`. idle_state: AtomicU8, + /// Set when `on_start` detected a repeat start and skipped spawning a child, so + /// `run` destroys the actor instead of running. See [`reject_second_start`]. + reject_start: AtomicBool, } impl GameServer { @@ -111,13 +114,31 @@ impl GameServer { } }); } + + /// Record that the actor received a request: disarms the one-shot idle timer, + /// and in idle mode marks the real start on the first request so an idle-slept + /// actor that never served one can wake without tripping the second-start guard. + fn note_request(&self, ctx: &Ctx) { + // Mark the actor active so the idle timer will not sleep this generation. + let _ = self.idle_state.compare_exchange( + IDLE_ARMED, + IDLE_REQUESTED, + Ordering::SeqCst, + Ordering::SeqCst, + ); + // In idle mode the real start is recorded on the first request. `mark_started_once` + // is idempotent, so calling it on every request is fine. + if reject_second_start() && idle_timeout().is_some() { + mark_started_once(ctx); + } + } } #[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; + // The persisted state (launch spec plus `started_once`) is restored on wake, so a + // woken actor keeps its spec without the engine re-sending input. + type State = ActorState; type Input = ActorInput; type Actions = (); type Events = (); @@ -127,13 +148,17 @@ impl Actor for GameServer { type Action = action::Raw; async fn create_state(_ctx: &Ctx, input: Self::Input) -> Result { - Ok(input) + Ok(ActorState { + input, + started_once: false, + }) } async fn create(_ctx: &Ctx) -> Result { Ok(Self { child: TokioMutex::new(None), idle_state: AtomicU8::new(IDLE_ARMED), + reject_start: AtomicBool::new(false), }) } @@ -165,9 +190,19 @@ impl Actor for GameServer { } } + // Second-start guard: if this actor already did its real start (a persisted + // flag that survives sleep), do not run again. Skip spawning a child; `run` + // destroys the actor. + if reject_second_start() && ctx.state().started_once { + tracing::warn!(actor_id = %actor_id, "actor tried a second-start"); + self.reject_start.store(true, Ordering::Relaxed); + return Ok(()); + } + // Copy the launch spec out of the state guard before any await. let (input_port, mut parts, env) = { - let input = ctx.state(); + let state = ctx.state(); + let input = &state.input; // input.command overrides the CLI template; input.args are appended. let mut parts = input .command @@ -241,6 +276,10 @@ impl Actor for GameServer { // hook to remove the entry, so registering earlier would leak it. register_ctx(&actor_id, &ctx).await; *self.child.lock().await = Some(child); + // Non-idle mode commits the real start only after the actor survives the + // second-start grace, so a fast crash-restart is not rejected. Idle mode defers + // this to the first request (see `note_request`) so an idle-slept actor can wake. + arm_second_start_mark(&ctx); self.arm_idle_timeout(&ctx, actor_id); Ok(()) } @@ -249,6 +288,15 @@ impl Actor for GameServer { /// first, so winning the `remove` race means the exit was unexpected: a clean /// exit destroys the actor, any other reports an errored stop (a crash). async fn run(self: Arc, ctx: Ctx) -> Result<()> { + // A rejected repeat start spawned no child; destroy the actor so it tombstones + // instead of parking. The child-exit destroy below has no pending request. + if self.reject_start.load(Ordering::Relaxed) { + if let Err(err) = ctx.destroy() { + tracing::debug!(error = ?err, actor_id = %ctx.actor_id(), "reject-start destroy failed"); + } + return Ok(()); + } + let Some(child) = self.child.lock().await.clone() else { anyhow::bail!("run: child process was never spawned"); }; @@ -277,7 +325,7 @@ impl Actor for GameServer { } async fn on_fetch(self: Arc, ctx: Ctx, req: Request) -> Result { - self.idle_state.store(IDLE_REQUESTED, Ordering::Relaxed); + self.note_request(&ctx); let child_port = self .child .lock() @@ -294,7 +342,7 @@ impl Actor for GameServer { ws: WebSocket, req: Request, ) -> Result<()> { - self.idle_state.store(IDLE_REQUESTED, Ordering::Relaxed); + self.note_request(&ctx); let child_port = self .child .lock() @@ -329,6 +377,35 @@ impl Actor for GameServer { } } +/// Non-idle second-start guard: commit `started_once` only after the actor survives +/// [`second_start_grace`], so a child that crashes within the window is not committed +/// and its retry may run. No-op when the guard is off or idle mode is on; cancelled on shutdown. +fn arm_second_start_mark(ctx: &Ctx) { + if !reject_second_start() || idle_timeout().is_some() { + return; + } + let ctx = ctx.clone(); + let grace = second_start_grace(); + tokio::spawn(async move { + let abort = ctx.abort_signal(); + tokio::select! { + _ = tokio::time::sleep(grace) => {} + _ = abort.cancelled() => return, + } + mark_started_once(&ctx); + }); +} + +/// Persist `started_once` so a later start is treated as a repeat. Idempotent: a +/// no-op when already set. The read guard is released before the write. +fn mark_started_once(ctx: &Ctx) { + if ctx.state().started_once { + return; + } + ctx.state_mut().started_once = true; + ctx.request_save(); +} + /// 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) { diff --git a/container-runner/src/input.rs b/container-runner/src/input.rs index c39fb6d0bd..ca70f1c62d 100644 --- a/container-runner/src/input.rs +++ b/container-runner/src/input.rs @@ -1,16 +1,29 @@ -//! The actor input payload describing how to launch the child game server. +//! The actor input payload and persisted state for the child game server. //! -//! The command, args, env, and port are carried in the actor's create-time `input` -//! (CBOR per RivetKit); anything omitted falls back to the CLI template -//! (`rivet-container-runner -- `). This is also the actor's persisted -//! state, so a woken actor restores the same launch spec. +//! [`ActorInput`] carries the command, args, env, and port from the engine's +//! create-time `input` (CBOR per RivetKit); anything omitted falls back to the CLI +//! template (`rivet-container-runner -- `). [`ActorState`] is what +//! persists across sleep: the launch spec plus lifecycle bookkeeping. use serde::{Deserialize, Serialize}; use std::collections::HashMap; +/// Persisted actor state, restored on wake. The launch spec is flattened in so an +/// actor persisted before `started_once` existed (state was a bare [`ActorInput`]) +/// still decodes. +#[derive(Debug, Default, Serialize, Deserialize)] +pub struct ActorState { + #[serde(flatten)] + pub input: ActorInput, + /// Set once the actor has performed its real start. The reject-second-start guard + /// self-sleeps a repeat start when this is already set. See `RIVET_REJECT_SECOND_START`. + #[serde(default)] + pub started_once: bool, +} + /// Shape of the actor `input` payload. Unknown fields are ignored, not rejected: -/// this is also the persisted state, and a strict decode would break waking actors -/// after a rollback to a binary predating a new field. +/// it nests in the persisted [`ActorState`], and a strict decode would break waking +/// actors after a rollback to a binary predating a new field. #[derive(Debug, Default, Serialize, Deserialize)] pub struct ActorInput { /// Overrides the CLI command template entirely (program + fixed args). diff --git a/container-runner/src/main.rs b/container-runner/src/main.rs index 10caa1dde8..5f2937d558 100644 --- a/container-runner/src/main.rs +++ b/container-runner/src/main.rs @@ -133,6 +133,36 @@ pub fn idle_timeout() -> Option { *IDLE_TIMEOUT } +/// When set, an actor that starts a second time self-sleeps instead of running +/// again; its persisted `started_once` records the first real start. Configured via +/// RIVET_REJECT_SECOND_START (truthy `1`/`true`/`yes`/`on`). Off by default. +static REJECT_SECOND_START: LazyLock = LazyLock::new(|| { + std::env::var("RIVET_REJECT_SECOND_START") + .map(|value| matches!(value.trim().to_ascii_lowercase().as_str(), "1" | "true" | "yes" | "on")) + .unwrap_or(false) +}); + +/// Whether the reject-second-start guard is enabled. See [`REJECT_SECOND_START`]. +pub fn reject_second_start() -> bool { + *REJECT_SECOND_START +} + +/// Non-idle mode commits `started_once` only after the actor survives this window, +/// so a child that crashes within it is not treated as a real start and the retry +/// may run. Configured via RIVET_SECOND_START_GRACE_SECS, default 10s. +static SECOND_START_GRACE: LazyLock = LazyLock::new(|| { + let secs = std::env::var("RIVET_SECOND_START_GRACE_SECS") + .ok() + .and_then(|value| value.parse::().ok()) + .unwrap_or(10); + Duration::from_secs(secs) +}); + +/// Grace an actor must survive before its start is committed. See [`SECOND_START_GRACE`]. +pub fn second_start_grace() -> Duration { + *SECOND_START_GRACE +} + /// Token that fires on a platform shutdown signal. Cuts a drain wait short so the /// platform's SIGTERM→SIGKILL budget is honored. pub fn exit_token() -> &'static CancellationToken { diff --git a/container-runner/tests/inline/input.rs b/container-runner/tests/inline/input.rs index 2dc78209fd..30e3dc3038 100644 --- a/container-runner/tests/inline/input.rs +++ b/container-runner/tests/inline/input.rs @@ -46,3 +46,44 @@ fn default_matches_empty() { assert!(default.env.is_empty()); assert!(default.port.is_none()); } + +fn cbor_round_trip(value: &T) -> anyhow::Result +where + T: serde::Serialize, + U: serde::de::DeserializeOwned, +{ + let mut buf = Vec::new(); + ciborium::into_writer(value, &mut buf)?; + Ok(ciborium::from_reader(&buf[..])?) +} + +#[test] +fn actor_state_cbor_round_trips() { + // State persists as CBOR (ciborium), so the flattened input must survive a CBOR + // round trip, not just the JSON the other tests use. + let state = ActorState { + input: ActorInput { + port: Some(7777), + args: vec!["-x".to_string()], + ..Default::default() + }, + started_once: true, + }; + let decoded: ActorState = cbor_round_trip(&state).unwrap(); + assert_eq!(decoded.input.port, Some(7777)); + assert_eq!(decoded.input.args, vec!["-x".to_string()]); + assert!(decoded.started_once); +} + +#[test] +fn legacy_bare_input_state_decodes_into_actor_state() { + // State written before `started_once` existed was a bare ActorInput. The + // flattened input must still decode, defaulting `started_once` to false. + let legacy = ActorInput { + port: Some(7777), + ..Default::default() + }; + let decoded: ActorState = cbor_round_trip(&legacy).unwrap(); + assert_eq!(decoded.input.port, Some(7777)); + assert!(!decoded.started_once); +} From 38e0ad13eb0646720508c35281bfe421cf630ded Mon Sep 17 00:00:00 2001 From: ABCxFF <79597906+abcxff@users.noreply.github.com> Date: Fri, 21 Aug 2026 19:38:41 -0400 Subject: [PATCH 8/9] feat(container-runner): jitter idle timeout to avoid teardown waves --- container-runner/src/actor.rs | 13 ++++++++----- container-runner/src/main.rs | 22 ++++++++++++++++++++++ 2 files changed, 30 insertions(+), 5 deletions(-) diff --git a/container-runner/src/actor.rs b/container-runner/src/actor.rs index 9d5a9c4aa4..cede720440 100644 --- a/container-runner/src/actor.rs +++ b/container-runner/src/actor.rs @@ -15,8 +15,9 @@ use tokio::sync::Mutex as TokioMutex; use crate::child::{ChildProcess, SpawnSpec, log_prefix}; use crate::input::{ActorInput, ActorState}; use crate::{ - children, drain_grace, effective_stop_grace, exit_token, idle_timeout, reject_second_start, - release_child_port, request_exit, reserve_child_port, runner_config, second_start_grace, + children, drain_grace, effective_stop_grace, exit_token, idle_timeout, idle_timeout_with_jitter, + reject_second_start, release_child_port, request_exit, reserve_child_port, runner_config, + second_start_grace, }; /// Live actor contexts keyed by actor id, so the shutdown path can report actors @@ -94,21 +95,23 @@ impl GameServer { /// window, if no request has arrived, ask the actor to sleep (`stop_child` then /// exits the container). Cancelled early if the actor starts shutting down. fn arm_idle_timeout(self: &Arc, ctx: &Ctx, actor_id: String) { - let Some(timeout) = idle_timeout() else { + let Some(base) = idle_timeout() else { return; }; + // Jitter the window so instances started together do not sleep in lockstep. + let delay = idle_timeout_with_jitter(base); let this = self.clone(); let ctx = ctx.clone(); tokio::spawn(async move { let abort = ctx.abort_signal(); tokio::select! { - _ = tokio::time::sleep(timeout) => {} + _ = tokio::time::sleep(delay) => {} _ = abort.cancelled() => return, } if this.idle_state.load(Ordering::Relaxed) == IDLE_REQUESTED { return; } - tracing::info!(actor_id = %actor_id, ?timeout, "no request within idle timeout, sleeping"); + tracing::info!(actor_id = %actor_id, ?delay, "no request within idle timeout, sleeping"); if let Err(err) = ctx.sleep() { tracing::debug!(error = ?err, actor_id = %actor_id, "idle sleep request failed"); } diff --git a/container-runner/src/main.rs b/container-runner/src/main.rs index 5f2937d558..0c4ee53318 100644 --- a/container-runner/src/main.rs +++ b/container-runner/src/main.rs @@ -133,6 +133,28 @@ pub fn idle_timeout() -> Option { *IDLE_TIMEOUT } +/// The idle window plus up to 20% jitter (capped at 60s), so instances armed at the +/// same time do not all sleep in the same instant and tear down in a wave. Jitter is +/// only ever added, never subtracted, so an actor never sleeps before its window. +pub fn idle_timeout_with_jitter(base: Duration) -> Duration { + let max_jitter = base.mul_f64(0.2).min(Duration::from_secs(60)); + base + random_duration_up_to(max_jitter) +} + +/// A `Duration` uniformly in `[0, max]`, drawn from the OS CSPRNG. Falls back to no +/// jitter when the CSPRNG is unavailable. +fn random_duration_up_to(max: Duration) -> Duration { + let max_ms = max.as_millis() as u64; + if max_ms == 0 { + return Duration::ZERO; + } + let mut buf = [0u8; 8]; + match std::fs::File::open("/dev/urandom").and_then(|mut f| f.read_exact(&mut buf)) { + Ok(()) => Duration::from_millis(u64::from_le_bytes(buf) % (max_ms + 1)), + Err(_) => Duration::ZERO, + } +} + /// When set, an actor that starts a second time self-sleeps instead of running /// again; its persisted `started_once` records the first real start. Configured via /// RIVET_REJECT_SECOND_START (truthy `1`/`true`/`yes`/`on`). Off by default. From 336ae682863b6530d90fd47f6574aee2459e9791 Mon Sep 17 00:00:00 2001 From: ABCxFF <79597906+abcxff@users.noreply.github.com> Date: Fri, 21 Aug 2026 20:17:30 -0400 Subject: [PATCH 9/9] feat(container-runner): drain engine sleeps unless the idle timer fired --- .../examples/e2e-test/reject-destroy-test.mjs | 52 +++++++++++++++++++ container-runner/src/actor.rs | 27 ++++++---- 2 files changed, 69 insertions(+), 10 deletions(-) create mode 100644 container-runner/examples/e2e-test/reject-destroy-test.mjs diff --git a/container-runner/examples/e2e-test/reject-destroy-test.mjs b/container-runner/examples/e2e-test/reject-destroy-test.mjs new file mode 100644 index 0000000000..9bfb279612 --- /dev/null +++ b/container-runner/examples/e2e-test/reject-destroy-test.mjs @@ -0,0 +1,52 @@ +// Test the "wait until running, then destroy" reject path. +// RIVET_URL=https://:@api.rivet.dev node reject-destroy-test.mjs +import { encode as cborEncode } from "cbor-x"; + +const url = new URL(process.env.RIVET_URL); +const namespace = decodeURIComponent(url.username); +const token = decodeURIComponent(url.password); +const origin = `${url.protocol}//${url.host}`; +const KEY = process.env.ACTOR_KEY || `rd-${Date.now()}`; +const sleep = (ms) => new Promise((r) => setTimeout(r, ms)); +const now = () => new Date().toISOString().slice(11, 23); + +async function api(path, opts = {}) { + const sep = path.includes("?") ? "&" : "?"; + const res = await fetch(`${origin}${path}${sep}namespace=${encodeURIComponent(namespace)}`, { + ...opts, + headers: { "content-type": "application/json", authorization: `Bearer ${token}`, ...(opts.headers || {}) }, + }); + const t = await res.text(); let b; try { b = JSON.parse(t); } catch { b = t; } + return { status: res.status, body: b }; +} +async function ping(id) { + try { const r = await fetch(`${origin}/gateway/${encodeURIComponent(id)}@${encodeURIComponent(token)}/`, { method: "GET" }); + return { status: r.status, text: (await r.text()).slice(0, 30) }; } catch (e) { return { status: 0, text: String(e).slice(0, 50) }; } +} +async function status(id) { + const { body } = await api(`/actors?actor_ids=${id}`, { method: "GET" }); + const a = body?.actors?.[0]; + return a ? { sleep_ts: a.sleep_ts, destroy_ts: a.destroy_ts } : { missing: true }; +} + +const input = Buffer.from(cborEncode({ port: 7770 })).toString("base64"); +let { body } = await api(`/actors`, { method: "POST", body: JSON.stringify({ name: "game", key: KEY, input, runner_name_selector: "default", crash_policy: "destroy" }) }); +const id = body?.actor?.actor_id || body?.metadata?.existing_actor_id; +console.log(`${now()} created ${id} key=${KEY}`); + +for (let i = 0; i < 20; i++) { const p = await ping(id); if (p.status === 200) { console.log(`${now()} up: ${p.status}`); break; } await sleep(750); } +console.log(`${now()} waiting 7s past grace so started_once persists`); +await sleep(7000); +await api(`/actors/${id}/sleep`, { method: "POST", body: "{}" }); +console.log(`${now()} slept; waiting 8s for container exit`); +await sleep(8000); +console.log(`${now()} single wake ping (expect reject -> wait -> destroy, NO loop)`); +await ping(id); +// poll status for ~20s to see if it destroys (destroy_ts set) or keeps looping/sleeping +for (let i = 0; i < 10; i++) { + await sleep(2000); + const s = await status(id); + console.log(`${now()} status: ${JSON.stringify(s)}`); + if (s.destroy_ts) { console.log(`${now()} DESTROYED (destroy_ts set) -- clean`); break; } +} +console.log(`${now()} DONE key=${KEY} id=${id}`); diff --git a/container-runner/src/actor.rs b/container-runner/src/actor.rs index cede720440..7890c95b42 100644 --- a/container-runner/src/actor.rs +++ b/container-runner/src/actor.rs @@ -25,15 +25,16 @@ use crate::{ static ACTOR_CTXS: LazyLock>> = LazyLock::new(scc::HashMap::new); -/// One-shot idle lifecycle for a generation, in a single atomic. The startup timer -/// sleeps the actor only while it is `ARMED`; a request moves it to `REQUESTED` so the -/// timer will not sleep. +/// One-shot idle lifecycle for a generation, in a single atomic. The startup timer sleeps +/// only while `ARMED`; a request moves it to `REQUESTED` (no sleep), and the timer firing with +/// no request moves it to `IDLE_SLEEPING`, which `on_sleep` reads to skip the drain. const IDLE_ARMED: u8 = 0; const IDLE_REQUESTED: u8 = 1; +const IDLE_SLEEPING: u8 = 2; pub struct GameServer { child: TokioMutex>>, - /// One-shot idle state: `IDLE_ARMED` / `IDLE_REQUESTED`. + /// One-shot idle state: `IDLE_ARMED` / `IDLE_REQUESTED` / `IDLE_SLEEPING`. idle_state: AtomicU8, /// Set when `on_start` detected a repeat start and skipped spawning a child, so /// `run` destroys the actor instead of running. See [`reject_second_start`]. @@ -108,7 +109,14 @@ impl GameServer { _ = tokio::time::sleep(delay) => {} _ = abort.cancelled() => return, } - if this.idle_state.load(Ordering::Relaxed) == IDLE_REQUESTED { + // Sleep only if still armed. If a request raced in, the CAS fails and we do + // nothing; on success the state records this as an idle-timer sleep so + // `on_sleep` skips the drain. + if this + .idle_state + .compare_exchange(IDLE_ARMED, IDLE_SLEEPING, Ordering::SeqCst, Ordering::SeqCst) + .is_err() + { return; } tracing::info!(actor_id = %actor_id, ?delay, "no request within idle timeout, sleeping"); @@ -361,12 +369,11 @@ impl Actor for GameServer { crate::proxy::ws_proxy(child_port, path, ws).await } - /// Engine-initiated sleep. `no_sleep` blocks only idle sleep; the engine can - /// still sleep an actor (dashboard, crash policy, eviction), so we stop the child. - /// In idle-timeout mode the sleep is (treated as) an idle sleep, so it skips the - /// drain and stops promptly; otherwise it drains for in-flight work first. + /// Engine-initiated sleep. An idle-timer sleep with no request has nothing to drain, so it + /// stops promptly; every other sleep (an active actor, or idle timeout disabled) drains + /// in-flight work first. `no_sleep` blocks only idle sleep, not engine-driven sleeps. async fn on_sleep(self: Arc, ctx: Ctx) -> Result<()> { - if idle_timeout().is_some() { + if self.idle_state.load(Ordering::SeqCst) == IDLE_SLEEPING { self.stop_child(ctx.actor_id(), "actor sleeping (idle)").await; } else { self.drain_then_stop_child(ctx.actor_id(), "actor sleeping").await;