From e9dadf6ae386d5e5ba3bb3499f1a63b588a69674 Mon Sep 17 00:00:00 2001 From: ABCxFF <79597906+abcxff@users.noreply.github.com> Date: Wed, 29 Jul 2026 16:19:34 +0000 Subject: [PATCH 1/7] fix(rivetkit): treat empty actor input bytes as absent --- rivetkit-rust/packages/rivetkit/src/start.rs | 55 +++++++++++++++++++- 1 file changed, 53 insertions(+), 2 deletions(-) diff --git a/rivetkit-rust/packages/rivetkit/src/start.rs b/rivetkit-rust/packages/rivetkit/src/start.rs index faa5368c67..a6ee050cf6 100644 --- a/rivetkit-rust/packages/rivetkit/src/start.rs +++ b/rivetkit-rust/packages/rivetkit/src/start.rs @@ -46,7 +46,11 @@ impl Input { } pub fn decode(&self) -> Result { - match self.bytes.as_deref() { + // Treat empty input bytes as absent so a zero-length payload defaults + // like a missing one, mirroring snapshot decoding. The engine encodes + // an omitted `input` as no bytes, but an empty base64 string decodes to + // an empty (non-null) buffer. + match self.present_bytes() { Some(bytes) => decode_cbor(bytes, "actor input"), None if TypeId::of::() == TypeId::of::<()>() => { let unit: Box = Box::new(()); @@ -62,12 +66,18 @@ impl Input { where F: FnOnce() -> A::Input, { - match self.bytes.as_deref() { + match self.present_bytes() { Some(bytes) => decode_cbor(bytes, "actor input"), None => Ok(f()), } } + /// Input bytes with empty buffers normalized to `None`, so a zero-length + /// payload is treated the same as an omitted one. + fn present_bytes(&self) -> Option<&[u8]> { + self.bytes.as_deref().filter(|bytes| !bytes.is_empty()) + } + pub fn decode_or_default(&self) -> Result where A::Input: Default, @@ -746,6 +756,31 @@ mod tests { assert_eq!(input.decode().expect("missing unit input"), ()); } + #[test] + fn input_decode_or_default_treats_empty_bytes_as_missing() { + // An empty base64 input decodes to an empty (non-null) buffer, which + // must default rather than fail CBOR decoding. + let input = Input:: { + bytes: Some(Vec::new()), + _p: PhantomData, + }; + + assert_eq!( + input.decode_or_default().expect("default input"), + DefaultInput { count: 7 } + ); + } + + #[test] + fn input_decode_treats_empty_unit_as_unit() { + let input = Input:: { + bytes: Some(Vec::new()), + _p: PhantomData, + }; + + assert_eq!(input.decode().expect("empty unit input"), ()); + } + #[test] fn connection_params_decode_null_as_default() { assert_eq!( @@ -946,6 +981,22 @@ mod tests { actor.await.expect("join run_actor").expect("run actor"); } + #[tokio::test] + async fn run_actor_invalid_input_fails_to_start() { + // Non-empty bytes that are not valid CBOR for the input type must fail + // the actor start rather than silently defaulting. + let (_tx, rx) = unbounded_channel(); + let start = lifecycle_start(Some(vec![0xff, 0xff, 0xff]), None, rx.into()); + + let error = run_actor::(start) + .await + .expect_err("invalid input should fail actor start"); + assert!( + format!("{error:#}").contains("decode actor input from cbor"), + "unexpected error: {error:#}" + ); + } + #[tokio::test] async fn run_actor_default_websocket_rejects() { let (tx, rx) = unbounded_channel(); From efaf873325f9673d4e9cc2f63c5de2cc50a34f00 Mon Sep 17 00:00:00 2001 From: ABCxFF <79597906+abcxff@users.noreply.github.com> Date: Wed, 29 Jul 2026 17:38:39 +0000 Subject: [PATCH 2/7] fix(rivetkit): log and surface actor startup failures --- rivetkit-rust/packages/rivetkit/src/start.rs | 112 +++++++++++++++---- 1 file changed, 92 insertions(+), 20 deletions(-) diff --git a/rivetkit-rust/packages/rivetkit/src/start.rs b/rivetkit-rust/packages/rivetkit/src/start.rs index a6ee050cf6..58d7c833ce 100644 --- a/rivetkit-rust/packages/rivetkit/src/start.rs +++ b/rivetkit-rust/packages/rivetkit/src/start.rs @@ -7,6 +7,7 @@ use std::time::Duration; use anyhow::{Context, Result}; use futures::FutureExt; +use rivet_error::RivetError; use rivetkit_core::actor::ShutdownKind; use rivetkit_core::error::{ActorLifecycle, ActorRuntime, action_not_found}; use rivetkit_core::{ActorEvent, ActorEvents, ActorStart, QueueSendResult, QueueSendStatus, Reply}; @@ -201,23 +202,45 @@ pub async fn run_actor(start: Start) -> Result<()> { startup_ready, } = start; - let state = match snapshot.decode()? { - Some(state) => state, - // Absent input falls back to the input type's default, matching - // rivetkit-typescript where createState receives undefined input. - None => A::create_state(&ctx, input.decode_or_default()?).await?, - }; - ctx.set_state(state); - ctx.clear_state_dirty(); + // Run the whole startup phase (input decode, state creation, create, + // on_create, on_start) as one fallible unit so a failure logs the real + // cause and is forwarded to the runtime handshake. Without this, an input + // decode error would drop `startup_ready` (surfacing only a generic + // closed-channel error) and never be logged until teardown. + let startup = async { + let state = match snapshot.decode()? { + Some(state) => state, + // Absent input falls back to the input type's default, matching + // rivetkit-typescript where createState receives undefined input. + None => A::create_state(&ctx, input.decode_or_default()?).await?, + }; + ctx.set_state(state); + ctx.clear_state_dirty(); - let actor = Arc::new(A::create(&ctx).await?); - if is_new { - actor.clone().on_create(ctx.clone()).await?; - } - actor.clone().on_start(ctx.clone()).await?; - if let Some(reply) = startup_ready { - let _ = reply.send(Ok(())); + let actor = Arc::new(A::create(&ctx).await?); + if is_new { + actor.clone().on_create(ctx.clone()).await?; + } + actor.clone().on_start(ctx.clone()).await?; + Ok::<_, anyhow::Error>(actor) } + .await; + + let actor = match startup { + Ok(actor) => { + if let Some(reply) = startup_ready { + let _ = reply.send(Ok(())); + } + actor + } + Err(error) => { + tracing::error!(actor_id = %ctx.actor_id(), ?error, "actor failed to start"); + if let Some(reply) = startup_ready { + let _ = reply.send(Err(anyhow::Error::new(RivetError::extract(&error)))); + } + return Err(error); + } + }; let run_cancel = CancellationToken::new(); let run_task = spawn_run_task(actor.clone(), ctx.clone(), run_cancel.clone()); @@ -984,17 +1007,66 @@ mod tests { #[tokio::test] async fn run_actor_invalid_input_fails_to_start() { // Non-empty bytes that are not valid CBOR for the input type must fail - // the actor start rather than silently defaulting. + // the actor start rather than silently defaulting. The failure is logged + // at startup and forwarded to the startup handshake instead of dropped. + struct BufMakeWriter(std::sync::Arc>>); + struct BufWriter(std::sync::Arc>>); + impl std::io::Write for BufWriter { + fn write(&mut self, data: &[u8]) -> std::io::Result { + self.0 + .lock() + .expect("log buffer poisoned") + .extend_from_slice(data); + Ok(data.len()) + } + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } + } + impl<'a> tracing_subscriber::fmt::MakeWriter<'a> for BufMakeWriter { + type Writer = BufWriter; + fn make_writer(&'a self) -> Self::Writer { + BufWriter(self.0.clone()) + } + } + let (_tx, rx) = unbounded_channel(); - let start = lifecycle_start(Some(vec![0xff, 0xff, 0xff]), None, rx.into()); + let (mut start, _ctx) = + lifecycle_start_with_ctx(Some(vec![0xff, 0xff, 0xff]), None, rx.into()); + let (ready_tx, ready_rx) = oneshot::channel(); + start.startup_ready = Some(ready_tx); + + let buffer = std::sync::Arc::new(std::sync::Mutex::new(Vec::::new())); + let subscriber = tracing_subscriber::fmt() + .with_writer(BufMakeWriter(buffer.clone())) + .with_ansi(false) + .finish(); + let error = { + let _guard = tracing::subscriber::set_default(subscriber); + run_actor::(start) + .await + .expect_err("invalid input should fail actor start") + }; - let error = run_actor::(start) - .await - .expect_err("invalid input should fail actor start"); assert!( format!("{error:#}").contains("decode actor input from cbor"), "unexpected error: {error:#}" ); + + // The failure is logged at startup, not deferred to teardown. + let logs = String::from_utf8(buffer.lock().expect("log buffer poisoned").clone()) + .expect("logs should be utf8"); + assert!( + logs.contains("actor failed to start") + && logs.contains("decode actor input from cbor"), + "startup failure not logged, got:\n{logs}" + ); + + // The startup handshake is signaled with an error rather than dropped. + ready_rx + .await + .expect("startup_ready should be signaled") + .expect_err("startup should report failure"); } #[tokio::test] From 4a79659c3052139a8233e43928e2da048b7cc2c4 Mon Sep 17 00:00:00 2001 From: ABCxFF <79597906+abcxff@users.noreply.github.com> Date: Wed, 29 Jul 2026 17:23:13 +0000 Subject: [PATCH 3/7] chore(container-runner): default stop grace to 10s --- container-runner/src/main.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/container-runner/src/main.rs b/container-runner/src/main.rs index 07fad243fe..b08e3e4597 100644 --- a/container-runner/src/main.rs +++ b/container-runner/src/main.rs @@ -244,7 +244,7 @@ struct Args { base_path: String, /// SIGTERM→SIGKILL grace period (seconds) when stopping the child. - #[arg(long, env = "RIVET_STOP_GRACE_SECS", default_value_t = 25)] + #[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. From b3a2e19183b7902f47205231842888cdd39261df Mon Sep 17 00:00:00 2001 From: ABCxFF <79597906+abcxff@users.noreply.github.com> Date: Wed, 29 Jul 2026 18:56:28 +0000 Subject: [PATCH 4/7] feat(container-runner): keep instance warm instead of self-exiting --- container-runner/src/actor.rs | 23 +++++++++++------------ container-runner/src/main.rs | 8 ++++---- 2 files changed, 15 insertions(+), 16 deletions(-) diff --git a/container-runner/src/actor.rs b/container-runner/src/actor.rs index ff9c53f621..8ebfd647d2 100644 --- a/container-runner/src/actor.rs +++ b/container-runner/src/actor.rs @@ -16,7 +16,7 @@ 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, request_exit, reserve_child_port, + children, effective_stop_grace, release_child_port, reserve_child_port, runner_config, }; @@ -40,11 +40,12 @@ impl GameServer { release_child_port(child.child_port).await; } - // Once the last actor is gone the instance drains rather than - // lingering for the next placement. - if children().is_empty() { - request_exit(actor_id, reason); - } + // 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"); } } @@ -129,12 +130,10 @@ impl Actor for GameServer { Ok(child) => Arc::new(child), Err(err) => { release_child_port(child_port).await; - // A failed start on an otherwise idle instance poisons it; - // don't let it serve the next placement. With other actors - // running, the failure is this actor's alone. - if children().is_empty() { - request_exit(&actor_id, "child failed to start"); - } + // 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. return Err(err); } }; diff --git a/container-runner/src/main.rs b/container-runner/src/main.rs index b08e3e4597..c0d3686a69 100644 --- a/container-runner/src/main.rs +++ b/container-runner/src/main.rs @@ -167,12 +167,12 @@ pub fn effective_stop_grace() -> Duration { } } -/// End the process. Called when the LAST actor on this instance is gone (or a -/// failed start poisoned an otherwise idle instance): the instance drains -/// instead of lingering for the next placement. The runner is PID 1 in the +/// 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. pub fn request_exit(actor_id: &str, reason: &str) { - tracing::info!(actor_id = %actor_id, reason, "actor finished, exiting container"); + tracing::info!(actor_id = %actor_id, reason, "shutting down container"); EXIT.cancel(); } From 858d26f01310828cc6500f4e81b532037fab61ba Mon Sep 17 00:00:00 2001 From: ABCxFF <79597906+abcxff@users.noreply.github.com> Date: Thu, 30 Jul 2026 15:11:19 +0000 Subject: [PATCH 5/7] fix(rivetkit-core): cancel driver alarm before sqlite teardown on destroy --- .../packages/rivetkit-core/src/actor/task.rs | 29 +++++++++++++------ 1 file changed, 20 insertions(+), 9 deletions(-) diff --git a/rivetkit-rust/packages/rivetkit-core/src/actor/task.rs b/rivetkit-rust/packages/rivetkit-core/src/actor/task.rs index 0c0a53cbdf..c71358969d 100644 --- a/rivetkit-rust/packages/rivetkit-core/src/actor/task.rs +++ b/rivetkit-rust/packages/rivetkit-core/src/actor/task.rs @@ -1801,6 +1801,25 @@ impl ActorTask { step = "sync_alarm", "actor shutdown cleanup step completed" ); + // Destroy cancels the engine-side alarm BEFORE the SQLite teardown. + // `cancel_driver_alarm_logged` issues a `set_alarm(None)` that spawns a + // last-pushed-alarm SQLite persist; running it here (rather than after + // `cleanup_sqlite`) lets `wait_for_pending_alarm_writes` below await that + // persist so it cannot race the teardown and fail with + // `transaction_closed`. Sleep keeps the persisted engine alarm armed for + // the next instance, so it only aborts the local timer, after cleanup. + match reason { + ShutdownKind::Destroy => { + ctx.cancel_driver_alarm_logged(); + tracing::debug!( + actor_id = %actor_id, + reason = reason_label, + step = "cancel_driver_alarm", + "actor shutdown cleanup step completed" + ); + } + ShutdownKind::Sleep => {} + } ctx.wait_for_pending_alarm_writes().await; tracing::debug!( actor_id = %actor_id, @@ -1834,15 +1853,7 @@ impl ActorTask { "actor shutdown cleanup step completed" ); } - ShutdownKind::Destroy => { - ctx.cancel_driver_alarm_logged(); - tracing::debug!( - actor_id = %actor_id, - reason = reason_label, - step = "cancel_driver_alarm", - "actor shutdown cleanup step completed" - ); - } + ShutdownKind::Destroy => {} } Ok(()) } From 30aa498b2a9d2792aff559f4c8369c68f40ba836 Mon Sep 17 00:00:00 2001 From: ABCxFF <79597906+abcxff@users.noreply.github.com> Date: Fri, 31 Jul 2026 04:23:39 +0000 Subject: [PATCH 6/7] feat(container-runner): log unexpected platform SIGTERM as an error --- container-runner/src/main.rs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/container-runner/src/main.rs b/container-runner/src/main.rs index c0d3686a69..5563dcb92c 100644 --- a/container-runner/src/main.rs +++ b/container-runner/src/main.rs @@ -413,7 +413,9 @@ fn spawn_signal_handler() { let mut sigterm = signal(SignalKind::terminate()).expect("install SIGTERM handler"); let mut sigint = signal(SignalKind::interrupt()).expect("install SIGINT handler"); tokio::select! { - _ = sigterm.recv() => tracing::info!("received SIGTERM"), + _ = sigterm.recv() => tracing::error!( + "unexpected loadbalancer-sourced SIGTERM received, likely hitting OOM or running longer than 60 minutes" + ), _ = sigint.recv() => tracing::info!("received SIGINT"), } SIGNAL_SHUTDOWN.store(true, Ordering::Release); From aee70efef6699c2644307ff1b12799b09195537a Mon Sep 17 00:00:00 2001 From: ABCxFF <79597906+abcxff@users.noreply.github.com> Date: Fri, 31 Jul 2026 04:45:38 +0000 Subject: [PATCH 7/7] feat(container-runner): destroy actor on sleep instead of parking it --- container-runner/src/actor.rs | 22 +++++++++++++++++----- 1 file changed, 17 insertions(+), 5 deletions(-) diff --git a/container-runner/src/actor.rs b/container-runner/src/actor.rs index 8ebfd647d2..f7bde5acb9 100644 --- a/container-runner/src/actor.rs +++ b/container-runner/src/actor.rs @@ -220,12 +220,24 @@ 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, which also covers GoingAway (eviction or + /// reallocation) since core maps that stop reason to a sleep. `no_sleep` + /// suppresses idle sleep, but the engine can still sleep an actor + /// (dashboard, crash policy, eviction ahead of instance retirement). + /// + /// A game server's live match state lives in the child process and cannot + /// survive it, so a woken actor would come back as an empty shell. Stop the + /// child and escalate the sleep into a full destroy so the engine tears the + /// actor down completely instead of parking it as sleeping. async fn on_sleep(self: Arc, ctx: Ctx) -> Result<()> { - self.stop_child(ctx.actor_id(), "actor sleeping").await; + let actor_id = ctx.actor_id().to_string(); + self.stop_child(&actor_id, "actor sleeping").await; + tracing::info!(actor_id = %actor_id, "received sleep, destroying actor"); + if let Err(err) = ctx.destroy() { + // The engine may already be destroying this generation; escalating a + // sleep that has already become a destroy is a harmless race. + tracing::debug!(error = ?err, actor_id = %actor_id, "destroy on sleep failed"); + } Ok(()) }