feat(container-runner): destroy actor on sleep instead of parking it

container-runner/src/actor.rs

@@ -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::{
-	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 @@
 			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 @@
 			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);
 			}
 		};
@@ -221,12 +220,24 @@
 		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<Self>, ctx: Ctx<Self>) -> 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(())
 	}
 

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();
 }
 
@@ -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.
@@ -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);

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(())
 	}

rivetkit-rust/packages/rivetkit/src/start.rs

@@ -7,6 +7,7 @@
 
 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};
@@ -46,7 +47,11 @@
 	}
 
 	pub fn decode(&self) -> Result<A::Input> {
-		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::<A::Input>() == TypeId::of::<()>() => {
 				let unit: Box<dyn Any> = Box::new(());
@@ -62,12 +67,18 @@
 	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<A::Input>
 	where
 		A::Input: Default,
@@ -191,23 +202,45 @@
 		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());
@@ -746,6 +779,31 @@
 		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::<DefaultActor> {
+			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::<EmptyActor> {
+			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 +1004,71 @@
 		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. The failure is logged
+		// at startup and forwarded to the startup handshake instead of dropped.
+		struct BufMakeWriter(std::sync::Arc<std::sync::Mutex<Vec<u8>>>);
+		struct BufWriter(std::sync::Arc<std::sync::Mutex<Vec<u8>>>);
+		impl std::io::Write for BufWriter {
+			fn write(&mut self, data: &[u8]) -> std::io::Result<usize> {
+				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 (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::<u8>::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::<LifecycleActor>(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]
 	async fn run_actor_default_websocket_rejects() {
 		let (tx, rx) = unbounded_channel();