From 686f2656a35ff8b98e454c35bfe87a769208eb76 Mon Sep 17 00:00:00 2001 From: ABCxFF <79597906+abcxff@users.noreply.github.com> Date: Mon, 10 Aug 2026 21:53:14 +0000 Subject: [PATCH 1/7] fix(sdks): drop rivet-util metrics dep from protocol crates to unlink rocksdb --- Cargo.lock | 5 - engine/sdks/rust/data/Cargo.toml | 1 - .../src/versioned/namespace_runner_config.rs | 24 +-- engine/sdks/rust/depot-protocol/Cargo.toml | 1 - .../sdks/rust/depot-protocol/src/versioned.rs | 4 +- engine/sdks/rust/envoy-protocol/Cargo.toml | 1 - .../rust/envoy-protocol/src/versioned/mod.rs | 148 +++++++++--------- engine/sdks/rust/epoxy-protocol/Cargo.toml | 1 - .../sdks/rust/epoxy-protocol/src/versioned.rs | 32 ++-- engine/sdks/rust/ups-protocol/Cargo.toml | 1 - .../sdks/rust/ups-protocol/src/versioned.rs | 12 +- 11 files changed, 110 insertions(+), 120 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 60e1e08b32..a2ade3bdd1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1731,7 +1731,6 @@ name = "epoxy-protocol" version = "2.3.7" dependencies = [ "anyhow", - "rivet-util", "rivet-vbare-compiler", "serde", "serde_bare", @@ -5340,7 +5339,6 @@ dependencies = [ "anyhow", "gasoline", "rivet-runner-protocol", - "rivet-util", "rivet-vbare-compiler", "serde", "serde_bare", @@ -5385,7 +5383,6 @@ name = "rivet-depot-protocol" version = "2.3.7" dependencies = [ "anyhow", - "rivet-util", "rivet-vbare-compiler", "serde", "serde_bare", @@ -5520,7 +5517,6 @@ dependencies = [ "anyhow", "hex", "rand 0.8.5", - "rivet-util", "rivet-vbare-compiler", "serde", "serde_bare", @@ -5983,7 +5979,6 @@ version = "2.3.7" dependencies = [ "anyhow", "base64 0.22.1", - "rivet-util", "rivet-vbare-compiler", "serde", "serde_bare", diff --git a/engine/sdks/rust/data/Cargo.toml b/engine/sdks/rust/data/Cargo.toml index f7c8c28c8c..0779ad14e5 100644 --- a/engine/sdks/rust/data/Cargo.toml +++ b/engine/sdks/rust/data/Cargo.toml @@ -10,7 +10,6 @@ edition.workspace = true anyhow.workspace = true gas.workspace = true rivet-runner-protocol.workspace = true -rivet-util.workspace = true serde_bare.workspace = true serde.workspace = true vbare.workspace = true diff --git a/engine/sdks/rust/data/src/versioned/namespace_runner_config.rs b/engine/sdks/rust/data/src/versioned/namespace_runner_config.rs index f887175e90..26097a4bd2 100644 --- a/engine/sdks/rust/data/src/versioned/namespace_runner_config.rs +++ b/engine/sdks/rust/data/src/versioned/namespace_runner_config.rs @@ -31,22 +31,22 @@ impl OwnedVersionedData for NamespaceRunnerConfig { fn deserialize_version(payload: &[u8], version: u16) -> Result { match version { 1 => Ok(NamespaceRunnerConfig::V1( - rivet_util::serde::bare_from_slice!(payload)?, + serde_bare::from_slice(payload)?, )), 2 => Ok(NamespaceRunnerConfig::V2( - rivet_util::serde::bare_from_slice!(payload)?, + serde_bare::from_slice(payload)?, )), 3 => Ok(NamespaceRunnerConfig::V3( - rivet_util::serde::bare_from_slice!(payload)?, + serde_bare::from_slice(payload)?, )), 4 => Ok(NamespaceRunnerConfig::V4( - rivet_util::serde::bare_from_slice!(payload)?, + serde_bare::from_slice(payload)?, )), 5 => Ok(NamespaceRunnerConfig::V5( - rivet_util::serde::bare_from_slice!(payload)?, + serde_bare::from_slice(payload)?, )), 6 => Ok(NamespaceRunnerConfig::V6( - rivet_util::serde::bare_from_slice!(payload)?, + serde_bare::from_slice(payload)?, )), _ => bail!("invalid version: {version}"), } @@ -55,22 +55,22 @@ impl OwnedVersionedData for NamespaceRunnerConfig { fn serialize_version(self, _version: u16) -> Result> { match self { NamespaceRunnerConfig::V1(data) => { - rivet_util::serde::bare_to_vec!(&data).map_err(Into::into) + serde_bare::to_vec(&data).map_err(Into::into) } NamespaceRunnerConfig::V2(data) => { - rivet_util::serde::bare_to_vec!(&data).map_err(Into::into) + serde_bare::to_vec(&data).map_err(Into::into) } NamespaceRunnerConfig::V3(data) => { - rivet_util::serde::bare_to_vec!(&data).map_err(Into::into) + serde_bare::to_vec(&data).map_err(Into::into) } NamespaceRunnerConfig::V4(data) => { - rivet_util::serde::bare_to_vec!(&data).map_err(Into::into) + serde_bare::to_vec(&data).map_err(Into::into) } NamespaceRunnerConfig::V5(data) => { - rivet_util::serde::bare_to_vec!(&data).map_err(Into::into) + serde_bare::to_vec(&data).map_err(Into::into) } NamespaceRunnerConfig::V6(data) => { - rivet_util::serde::bare_to_vec!(&data).map_err(Into::into) + serde_bare::to_vec(&data).map_err(Into::into) } } } diff --git a/engine/sdks/rust/depot-protocol/Cargo.toml b/engine/sdks/rust/depot-protocol/Cargo.toml index 99d8117349..274836220a 100644 --- a/engine/sdks/rust/depot-protocol/Cargo.toml +++ b/engine/sdks/rust/depot-protocol/Cargo.toml @@ -8,7 +8,6 @@ edition.workspace = true [dependencies] anyhow.workspace = true -rivet-util.workspace = true serde_bare.workspace = true serde.workspace = true vbare.workspace = true diff --git a/engine/sdks/rust/depot-protocol/src/versioned.rs b/engine/sdks/rust/depot-protocol/src/versioned.rs index d90a617eec..c8432da5ae 100644 --- a/engine/sdks/rust/depot-protocol/src/versioned.rs +++ b/engine/sdks/rust/depot-protocol/src/versioned.rs @@ -22,14 +22,14 @@ impl OwnedVersionedData for DBHead { fn deserialize_version(payload: &[u8], version: u16) -> Result { match version { - 1 => Ok(Self::V1(rivet_util::serde::bare_from_slice!(payload)?)), + 1 => Ok(Self::V1(serde_bare::from_slice(payload)?)), _ => bail!("invalid depot db head version: {version}"), } } fn serialize_version(self, _version: u16) -> Result> { match self { - Self::V1(data) => rivet_util::serde::bare_to_vec!(&data).map_err(Into::into), + Self::V1(data) => serde_bare::to_vec(&data).map_err(Into::into), } } } diff --git a/engine/sdks/rust/envoy-protocol/Cargo.toml b/engine/sdks/rust/envoy-protocol/Cargo.toml index 907353804d..d0da835e5e 100644 --- a/engine/sdks/rust/envoy-protocol/Cargo.toml +++ b/engine/sdks/rust/envoy-protocol/Cargo.toml @@ -12,7 +12,6 @@ description = "Versioned Envoy protocol types for Rivet actor hosts" anyhow.workspace = true hex.workspace = true rand.workspace = true -rivet-util.workspace = true serde_bare.workspace = true serde.workspace = true utoipa.workspace = true diff --git a/engine/sdks/rust/envoy-protocol/src/versioned/mod.rs b/engine/sdks/rust/envoy-protocol/src/versioned/mod.rs index 5357809472..fb5b57ed98 100644 --- a/engine/sdks/rust/envoy-protocol/src/versioned/mod.rs +++ b/engine/sdks/rust/envoy-protocol/src/versioned/mod.rs @@ -129,24 +129,24 @@ impl OwnedVersionedData for ToEnvoy { fn deserialize_version(payload: &[u8], version: u16) -> Result { match version { - 1 => Ok(Self::V1(rivet_util::serde::bare_from_slice!(payload)?)), - 2 => Ok(Self::V2(rivet_util::serde::bare_from_slice!(payload)?)), - 3 => Ok(Self::V3(rivet_util::serde::bare_from_slice!(payload)?)), - 4 => Ok(Self::V4(rivet_util::serde::bare_from_slice!(payload)?)), - 5 => Ok(Self::V5(rivet_util::serde::bare_from_slice!(payload)?)), - 6 => Ok(Self::V6(rivet_util::serde::bare_from_slice!(payload)?)), + 1 => Ok(Self::V1(serde_bare::from_slice(payload)?)), + 2 => Ok(Self::V2(serde_bare::from_slice(payload)?)), + 3 => Ok(Self::V3(serde_bare::from_slice(payload)?)), + 4 => Ok(Self::V4(serde_bare::from_slice(payload)?)), + 5 => Ok(Self::V5(serde_bare::from_slice(payload)?)), + 6 => Ok(Self::V6(serde_bare::from_slice(payload)?)), _ => bail!("invalid version: {version}"), } } fn serialize_version(self, _version: u16) -> Result> { match self { - Self::V1(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into), - Self::V2(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into), - Self::V3(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into), - Self::V4(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into), - Self::V5(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into), - Self::V6(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into), + Self::V1(x) => serde_bare::to_vec(&x).map_err(Into::into), + Self::V2(x) => serde_bare::to_vec(&x).map_err(Into::into), + Self::V3(x) => serde_bare::to_vec(&x).map_err(Into::into), + Self::V4(x) => serde_bare::to_vec(&x).map_err(Into::into), + Self::V5(x) => serde_bare::to_vec(&x).map_err(Into::into), + Self::V6(x) => serde_bare::to_vec(&x).map_err(Into::into), } } @@ -261,24 +261,24 @@ impl OwnedVersionedData for ToRivet { fn deserialize_version(payload: &[u8], version: u16) -> Result { match version { - 1 => Ok(Self::V1(rivet_util::serde::bare_from_slice!(payload)?)), - 2 => Ok(Self::V2(rivet_util::serde::bare_from_slice!(payload)?)), - 3 => Ok(Self::V3(rivet_util::serde::bare_from_slice!(payload)?)), - 4 => Ok(Self::V4(rivet_util::serde::bare_from_slice!(payload)?)), - 5 => Ok(Self::V5(rivet_util::serde::bare_from_slice!(payload)?)), - 6 => Ok(Self::V6(rivet_util::serde::bare_from_slice!(payload)?)), + 1 => Ok(Self::V1(serde_bare::from_slice(payload)?)), + 2 => Ok(Self::V2(serde_bare::from_slice(payload)?)), + 3 => Ok(Self::V3(serde_bare::from_slice(payload)?)), + 4 => Ok(Self::V4(serde_bare::from_slice(payload)?)), + 5 => Ok(Self::V5(serde_bare::from_slice(payload)?)), + 6 => Ok(Self::V6(serde_bare::from_slice(payload)?)), _ => bail!("invalid version: {version}"), } } fn serialize_version(self, _version: u16) -> Result> { match self { - Self::V1(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into), - Self::V2(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into), - Self::V3(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into), - Self::V4(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into), - Self::V5(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into), - Self::V6(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into), + Self::V1(x) => serde_bare::to_vec(&x).map_err(Into::into), + Self::V2(x) => serde_bare::to_vec(&x).map_err(Into::into), + Self::V3(x) => serde_bare::to_vec(&x).map_err(Into::into), + Self::V4(x) => serde_bare::to_vec(&x).map_err(Into::into), + Self::V5(x) => serde_bare::to_vec(&x).map_err(Into::into), + Self::V6(x) => serde_bare::to_vec(&x).map_err(Into::into), } } @@ -393,24 +393,24 @@ impl OwnedVersionedData for ToEnvoyConn { fn deserialize_version(payload: &[u8], version: u16) -> Result { match version { - 1 => Ok(Self::V1(rivet_util::serde::bare_from_slice!(payload)?)), - 2 => Ok(Self::V2(rivet_util::serde::bare_from_slice!(payload)?)), - 3 => Ok(Self::V3(rivet_util::serde::bare_from_slice!(payload)?)), - 4 => Ok(Self::V4(rivet_util::serde::bare_from_slice!(payload)?)), - 5 => Ok(Self::V5(rivet_util::serde::bare_from_slice!(payload)?)), - 6 => Ok(Self::V6(rivet_util::serde::bare_from_slice!(payload)?)), + 1 => Ok(Self::V1(serde_bare::from_slice(payload)?)), + 2 => Ok(Self::V2(serde_bare::from_slice(payload)?)), + 3 => Ok(Self::V3(serde_bare::from_slice(payload)?)), + 4 => Ok(Self::V4(serde_bare::from_slice(payload)?)), + 5 => Ok(Self::V5(serde_bare::from_slice(payload)?)), + 6 => Ok(Self::V6(serde_bare::from_slice(payload)?)), _ => bail!("invalid version: {version}"), } } fn serialize_version(self, _version: u16) -> Result> { match self { - Self::V1(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into), - Self::V2(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into), - Self::V3(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into), - Self::V4(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into), - Self::V5(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into), - Self::V6(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into), + Self::V1(x) => serde_bare::to_vec(&x).map_err(Into::into), + Self::V2(x) => serde_bare::to_vec(&x).map_err(Into::into), + Self::V3(x) => serde_bare::to_vec(&x).map_err(Into::into), + Self::V4(x) => serde_bare::to_vec(&x).map_err(Into::into), + Self::V5(x) => serde_bare::to_vec(&x).map_err(Into::into), + Self::V6(x) => serde_bare::to_vec(&x).map_err(Into::into), } } @@ -525,24 +525,24 @@ impl OwnedVersionedData for ToGateway { fn deserialize_version(payload: &[u8], version: u16) -> Result { match version { - 1 => Ok(Self::V1(rivet_util::serde::bare_from_slice!(payload)?)), - 2 => Ok(Self::V2(rivet_util::serde::bare_from_slice!(payload)?)), - 3 => Ok(Self::V3(rivet_util::serde::bare_from_slice!(payload)?)), - 4 => Ok(Self::V4(rivet_util::serde::bare_from_slice!(payload)?)), - 5 => Ok(Self::V5(rivet_util::serde::bare_from_slice!(payload)?)), - 6 => Ok(Self::V6(rivet_util::serde::bare_from_slice!(payload)?)), + 1 => Ok(Self::V1(serde_bare::from_slice(payload)?)), + 2 => Ok(Self::V2(serde_bare::from_slice(payload)?)), + 3 => Ok(Self::V3(serde_bare::from_slice(payload)?)), + 4 => Ok(Self::V4(serde_bare::from_slice(payload)?)), + 5 => Ok(Self::V5(serde_bare::from_slice(payload)?)), + 6 => Ok(Self::V6(serde_bare::from_slice(payload)?)), _ => bail!("invalid version: {version}"), } } fn serialize_version(self, _version: u16) -> Result> { match self { - Self::V1(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into), - Self::V2(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into), - Self::V3(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into), - Self::V4(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into), - Self::V5(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into), - Self::V6(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into), + Self::V1(x) => serde_bare::to_vec(&x).map_err(Into::into), + Self::V2(x) => serde_bare::to_vec(&x).map_err(Into::into), + Self::V3(x) => serde_bare::to_vec(&x).map_err(Into::into), + Self::V4(x) => serde_bare::to_vec(&x).map_err(Into::into), + Self::V5(x) => serde_bare::to_vec(&x).map_err(Into::into), + Self::V6(x) => serde_bare::to_vec(&x).map_err(Into::into), } } @@ -657,24 +657,24 @@ impl OwnedVersionedData for ToOutbound { fn deserialize_version(payload: &[u8], version: u16) -> Result { match version { - 1 => Ok(Self::V1(rivet_util::serde::bare_from_slice!(payload)?)), - 2 => Ok(Self::V2(rivet_util::serde::bare_from_slice!(payload)?)), - 3 => Ok(Self::V3(rivet_util::serde::bare_from_slice!(payload)?)), - 4 => Ok(Self::V4(rivet_util::serde::bare_from_slice!(payload)?)), - 5 => Ok(Self::V5(rivet_util::serde::bare_from_slice!(payload)?)), - 6 => Ok(Self::V6(rivet_util::serde::bare_from_slice!(payload)?)), + 1 => Ok(Self::V1(serde_bare::from_slice(payload)?)), + 2 => Ok(Self::V2(serde_bare::from_slice(payload)?)), + 3 => Ok(Self::V3(serde_bare::from_slice(payload)?)), + 4 => Ok(Self::V4(serde_bare::from_slice(payload)?)), + 5 => Ok(Self::V5(serde_bare::from_slice(payload)?)), + 6 => Ok(Self::V6(serde_bare::from_slice(payload)?)), _ => bail!("invalid version: {version}"), } } fn serialize_version(self, _version: u16) -> Result> { match self { - Self::V1(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into), - Self::V2(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into), - Self::V3(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into), - Self::V4(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into), - Self::V5(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into), - Self::V6(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into), + Self::V1(x) => serde_bare::to_vec(&x).map_err(Into::into), + Self::V2(x) => serde_bare::to_vec(&x).map_err(Into::into), + Self::V3(x) => serde_bare::to_vec(&x).map_err(Into::into), + Self::V4(x) => serde_bare::to_vec(&x).map_err(Into::into), + Self::V5(x) => serde_bare::to_vec(&x).map_err(Into::into), + Self::V6(x) => serde_bare::to_vec(&x).map_err(Into::into), } } @@ -789,24 +789,24 @@ impl OwnedVersionedData for ActorCommandKeyData { fn deserialize_version(payload: &[u8], version: u16) -> Result { match version { - 1 => Ok(Self::V1(rivet_util::serde::bare_from_slice!(payload)?)), - 2 => Ok(Self::V2(rivet_util::serde::bare_from_slice!(payload)?)), - 3 => Ok(Self::V3(rivet_util::serde::bare_from_slice!(payload)?)), - 4 => Ok(Self::V4(rivet_util::serde::bare_from_slice!(payload)?)), - 5 => Ok(Self::V5(rivet_util::serde::bare_from_slice!(payload)?)), - 6 => Ok(Self::V6(rivet_util::serde::bare_from_slice!(payload)?)), + 1 => Ok(Self::V1(serde_bare::from_slice(payload)?)), + 2 => Ok(Self::V2(serde_bare::from_slice(payload)?)), + 3 => Ok(Self::V3(serde_bare::from_slice(payload)?)), + 4 => Ok(Self::V4(serde_bare::from_slice(payload)?)), + 5 => Ok(Self::V5(serde_bare::from_slice(payload)?)), + 6 => Ok(Self::V6(serde_bare::from_slice(payload)?)), _ => bail!("invalid version: {version}"), } } fn serialize_version(self, _version: u16) -> Result> { match self { - Self::V1(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into), - Self::V2(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into), - Self::V3(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into), - Self::V4(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into), - Self::V5(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into), - Self::V6(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into), + Self::V1(x) => serde_bare::to_vec(&x).map_err(Into::into), + Self::V2(x) => serde_bare::to_vec(&x).map_err(Into::into), + Self::V3(x) => serde_bare::to_vec(&x).map_err(Into::into), + Self::V4(x) => serde_bare::to_vec(&x).map_err(Into::into), + Self::V5(x) => serde_bare::to_vec(&x).map_err(Into::into), + Self::V6(x) => serde_bare::to_vec(&x).map_err(Into::into), } } @@ -934,7 +934,7 @@ mod tests { #[test] fn v1_start_command_deserializes_into_latest_without_sqlite_startup_data() -> Result<()> { - let payload = rivet_util::serde::bare_to_vec!(&v1::ToEnvoy::ToEnvoyCommands(vec![ + let payload = serde_bare::to_vec(&v1::ToEnvoy::ToEnvoyCommands(vec![ v1::CommandWrapper { checkpoint: v1::ActorCheckpoint { actor_id: "actor".into(), @@ -970,7 +970,7 @@ mod tests { #[test] fn v2_sqlite_response_does_not_deserialize_to_stateless_protocol() -> Result<()> { - let payload = rivet_util::serde::bare_to_vec!(&v2::ToEnvoy::ToEnvoySqliteCommitResponse( + let payload = serde_bare::to_vec(&v2::ToEnvoy::ToEnvoySqliteCommitResponse( v2::ToEnvoySqliteCommitResponse { request_id: 1, data: v2::SqliteCommitResponse::SqliteErrorResponse(v2::SqliteErrorResponse { diff --git a/engine/sdks/rust/epoxy-protocol/Cargo.toml b/engine/sdks/rust/epoxy-protocol/Cargo.toml index 5903387849..0197f4926d 100644 --- a/engine/sdks/rust/epoxy-protocol/Cargo.toml +++ b/engine/sdks/rust/epoxy-protocol/Cargo.toml @@ -8,7 +8,6 @@ edition.workspace = true [dependencies] anyhow.workspace = true -rivet-util.workspace = true serde_bare.workspace = true serde.workspace = true vbare.workspace = true diff --git a/engine/sdks/rust/epoxy-protocol/src/versioned.rs b/engine/sdks/rust/epoxy-protocol/src/versioned.rs index 2dd501bbb8..63d8f736a2 100644 --- a/engine/sdks/rust/epoxy-protocol/src/versioned.rs +++ b/engine/sdks/rust/epoxy-protocol/src/versioned.rs @@ -26,10 +26,10 @@ impl OwnedVersionedData for CommittedValue { fn deserialize_version(payload: &[u8], version: u16) -> Result { match version { - 2 => Ok(CommittedValue::V2(rivet_util::serde::bare_from_slice!( + 2 => Ok(CommittedValue::V2(serde_bare::from_slice( payload )?)), - 3 => Ok(CommittedValue::V3(rivet_util::serde::bare_from_slice!( + 3 => Ok(CommittedValue::V3(serde_bare::from_slice( payload )?)), _ => bail!("invalid version: {version}"), @@ -38,8 +38,8 @@ impl OwnedVersionedData for CommittedValue { fn serialize_version(self, _version: u16) -> Result> { match self { - CommittedValue::V2(data) => rivet_util::serde::bare_to_vec!(&data).map_err(Into::into), - CommittedValue::V3(data) => rivet_util::serde::bare_to_vec!(&data).map_err(Into::into), + CommittedValue::V2(data) => serde_bare::to_vec(&data).map_err(Into::into), + CommittedValue::V3(data) => serde_bare::to_vec(&data).map_err(Into::into), } } @@ -93,10 +93,10 @@ impl OwnedVersionedData for CachedValue { fn deserialize_version(payload: &[u8], version: u16) -> Result { match version { - 2 => Ok(CachedValue::V2(rivet_util::serde::bare_from_slice!( + 2 => Ok(CachedValue::V2(serde_bare::from_slice( payload )?)), - 3 => Ok(CachedValue::V3(rivet_util::serde::bare_from_slice!( + 3 => Ok(CachedValue::V3(serde_bare::from_slice( payload )?)), _ => bail!("invalid version: {version}"), @@ -105,8 +105,8 @@ impl OwnedVersionedData for CachedValue { fn serialize_version(self, _version: u16) -> Result> { match self { - CachedValue::V2(data) => rivet_util::serde::bare_to_vec!(&data).map_err(Into::into), - CachedValue::V3(data) => rivet_util::serde::bare_to_vec!(&data).map_err(Into::into), + CachedValue::V2(data) => serde_bare::to_vec(&data).map_err(Into::into), + CachedValue::V3(data) => serde_bare::to_vec(&data).map_err(Into::into), } } @@ -161,10 +161,10 @@ impl OwnedVersionedData for AcceptedValue { fn deserialize_version(payload: &[u8], version: u16) -> Result { match version { - 2 => Ok(AcceptedValue::V2(rivet_util::serde::bare_from_slice!( + 2 => Ok(AcceptedValue::V2(serde_bare::from_slice( payload )?)), - 3 => Ok(AcceptedValue::V3(rivet_util::serde::bare_from_slice!( + 3 => Ok(AcceptedValue::V3(serde_bare::from_slice( payload )?)), _ => bail!("invalid version: {version}"), @@ -173,8 +173,8 @@ impl OwnedVersionedData for AcceptedValue { fn serialize_version(self, _version: u16) -> Result> { match self { - AcceptedValue::V2(data) => rivet_util::serde::bare_to_vec!(&data).map_err(Into::into), - AcceptedValue::V3(data) => rivet_util::serde::bare_to_vec!(&data).map_err(Into::into), + AcceptedValue::V2(data) => serde_bare::to_vec(&data).map_err(Into::into), + AcceptedValue::V3(data) => serde_bare::to_vec(&data).map_err(Into::into), } } @@ -228,16 +228,16 @@ impl OwnedVersionedData for Request { fn deserialize_version(payload: &[u8], version: u16) -> Result { match version { - 2 => Ok(Request::V2(rivet_util::serde::bare_from_slice!(payload)?)), - 3 => Ok(Request::V3(rivet_util::serde::bare_from_slice!(payload)?)), + 2 => Ok(Request::V2(serde_bare::from_slice(payload)?)), + 3 => Ok(Request::V3(serde_bare::from_slice(payload)?)), _ => bail!("invalid version: {version}"), } } fn serialize_version(self, _version: u16) -> Result> { match self { - Request::V2(data) => rivet_util::serde::bare_to_vec!(&data).map_err(Into::into), - Request::V3(data) => rivet_util::serde::bare_to_vec!(&data).map_err(Into::into), + Request::V2(data) => serde_bare::to_vec(&data).map_err(Into::into), + Request::V3(data) => serde_bare::to_vec(&data).map_err(Into::into), } } diff --git a/engine/sdks/rust/ups-protocol/Cargo.toml b/engine/sdks/rust/ups-protocol/Cargo.toml index 9105a5ea23..f6af0850c4 100644 --- a/engine/sdks/rust/ups-protocol/Cargo.toml +++ b/engine/sdks/rust/ups-protocol/Cargo.toml @@ -9,7 +9,6 @@ edition.workspace = true [dependencies] anyhow.workspace = true base64.workspace = true -rivet-util.workspace = true serde_bare.workspace = true serde.workspace = true vbare.workspace = true diff --git a/engine/sdks/rust/ups-protocol/src/versioned.rs b/engine/sdks/rust/ups-protocol/src/versioned.rs index ee756e64bc..a49b575209 100644 --- a/engine/sdks/rust/ups-protocol/src/versioned.rs +++ b/engine/sdks/rust/ups-protocol/src/versioned.rs @@ -26,13 +26,13 @@ impl OwnedVersionedData for UpsMessage { fn deserialize_version(payload: &[u8], version: u16) -> Result { match version { - 1 => Ok(UpsMessage::V1(rivet_util::serde::bare_from_slice!( + 1 => Ok(UpsMessage::V1(serde_bare::from_slice( payload )?)), - 2 => Ok(UpsMessage::V2(rivet_util::serde::bare_from_slice!( + 2 => Ok(UpsMessage::V2(serde_bare::from_slice( payload )?)), - 3 => Ok(UpsMessage::V3(rivet_util::serde::bare_from_slice!( + 3 => Ok(UpsMessage::V3(serde_bare::from_slice( payload )?)), _ => bail!("invalid version: {version}"), @@ -41,9 +41,9 @@ impl OwnedVersionedData for UpsMessage { fn serialize_version(self, _version: u16) -> Result> { match self { - UpsMessage::V1(data) => rivet_util::serde::bare_to_vec!(&data).map_err(Into::into), - UpsMessage::V2(data) => rivet_util::serde::bare_to_vec!(&data).map_err(Into::into), - UpsMessage::V3(data) => rivet_util::serde::bare_to_vec!(&data).map_err(Into::into), + UpsMessage::V1(data) => serde_bare::to_vec(&data).map_err(Into::into), + UpsMessage::V2(data) => serde_bare::to_vec(&data).map_err(Into::into), + UpsMessage::V3(data) => serde_bare::to_vec(&data).map_err(Into::into), } } From d0770c7c367d174506a4d00208fa117a04c57690 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 | 82 ++++++++++++++++---- 1 file changed, 67 insertions(+), 15 deletions(-) diff --git a/rivetkit-rust/packages/rivetkit/src/start.rs b/rivetkit-rust/packages/rivetkit/src/start.rs index faa5368c67..c1092cbf84 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}; @@ -191,23 +192,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 is forwarded to + // the runtime handshake as the real cause instead of being dropped. Without + // this, an input decode error would drop `startup_ready`, surfacing only a + // generic closed-channel error rather than the actual failure. The failure + // itself is logged by rivetkit-core when it drains the run handle. + 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) => { + 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()); @@ -946,6 +969,35 @@ 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, and the failure must + // be forwarded to the startup handshake instead of dropped. + 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 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 startup handshake is signaled with the error rather than dropped, + // so the runtime handshake surfaces the real cause. rivetkit-core owns + // logging the failure when it drains the run handle. + 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(); From e0d54274d76286ba966e14538fd0adfc6f7a9d95 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/child.rs | 2 +- container-runner/src/main.rs | 21 +++++++++++---------- 2 files changed, 12 insertions(+), 11 deletions(-) diff --git a/container-runner/src/child.rs b/container-runner/src/child.rs index 64c4785074..c9d5843ae6 100644 --- a/container-runner/src/child.rs +++ b/container-runner/src/child.rs @@ -86,7 +86,7 @@ impl ChildProcess { "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 \ - max_concurrent_actors=1 and Cloud Run request concurrency=1." + max_concurrent_actors=1 and platform request concurrency=1." ); } diff --git a/container-runner/src/main.rs b/container-runner/src/main.rs index 07fad243fe..6627dd68ea 100644 --- a/container-runner/src/main.rs +++ b/container-runner/src/main.rs @@ -68,15 +68,15 @@ static RESERVED_PORTS: LazyLock> = LazyLock::new(scc::HashSet: static EXIT: LazyLock = LazyLock::new(CancellationToken::new); /// Set when the process is shutting down because the PLATFORM sent a signal. -/// Cloud Run gives a container roughly 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). +/// 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). static SIGNAL_SHUTDOWN: AtomicBool = AtomicBool::new(false); /// How long the platform gives this container between SIGTERM and SIGKILL. -/// Cloud Run defaults to 10 seconds (configurable up to 60 on the service); -/// keep this in sync with the platform setting via RIVET_SIGTERM_BUDGET_SECS. +/// 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. @@ -222,8 +222,9 @@ fn base64url_nopad(input: &[u8]) -> String { long_about = None, )] struct Args { - /// Serverless HTTP front-door port. Rivet Compute injects RIVET_PORT (plain Cloud Run - /// uses PORT); resolved in `main` as --port > RIVET_PORT > PORT > 8080. + /// Serverless HTTP front-door port. Rivet Compute injects RIVET_PORT (other + /// serverless platforms use the conventional PORT); resolved in `main` as + /// --port > RIVET_PORT > PORT > 8080. #[arg(long)] port: Option, @@ -244,7 +245,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. @@ -285,7 +286,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; plain Cloud Run uses PORT. + // Front-door port: Rivet Compute injects RIVET_PORT; other serverless platforms use the conventional PORT. let port = args .port .or_else(|| env_u16("RIVET_PORT")) From 3384fd338bbcb78a779d999ab73f90a28e8208b1 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/README.md | 3 ++- container-runner/src/actor.rs | 26 +++++++++++++------------- container-runner/src/main.rs | 26 +++++++++++++++----------- 3 files changed, 30 insertions(+), 25 deletions(-) diff --git a/container-runner/README.md b/container-runner/README.md index 2dfc56f4e7..0a3fda9a8c 100644 --- a/container-runner/README.md +++ b/container-runner/README.md @@ -8,7 +8,8 @@ This README covers development, the example projects, and the local test harness spawning a child game-server process per actor and proxying Rivet's tunneled HTTP/WebSocket traffic to it. Each child gets its own port; the pool's request concurrency decides how many actors share a container (one, in the recommended -game-server setup), and the process exits when the last actor stops. Wrap any dedicated server (Unity, Godot, a plain Node process) in a +game-server setup). The instance stays warm after its last actor stops; the engine +reaps it by draining the `/start` connection after the request lifespan. Wrap any dedicated server (Unity, Godot, a plain Node process) in a container with this binary as the entrypoint and Rivet Compute can cold-start and route to it. diff --git a/container-runner/src/actor.rs b/container-runner/src/actor.rs index ff9c53f621..80d25cda59 100644 --- a/container-runner/src/actor.rs +++ b/container-runner/src/actor.rs @@ -4,7 +4,8 @@ //! 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, exiting the process once no actors remain. +//! `on_destroy` stops the child while the instance stays warm for the next +//! placement. use std::sync::Arc; @@ -16,7 +17,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 +41,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 +131,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 6627dd68ea..35b42dd14f 100644 --- a/container-runner/src/main.rs +++ b/container-runner/src/main.rs @@ -12,8 +12,9 @@ //! 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). When the last actor stops the process exits so the platform reaps -//! the instance. +//! 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. mod actor; mod child; @@ -167,12 +168,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(); } @@ -346,7 +347,11 @@ async fn async_main() -> Result<()> { )); tracing::info!(port, "container-runner serverless front door listening"); - // Wait for an exit request, then tear down. Two orders depending on why: + // 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. // // Signal (platform is reclaiming the instance): tell the engine FIRST so // it can start re-placing actors immediately. Its per-actor stops run our @@ -354,10 +359,9 @@ async fn async_main() -> Result<()> { // The drain is bounded so an unreachable engine cannot eat the whole // platform budget; the sweep then catches any child whose hooks never ran. // - // Actor-driven exit (last actor stopped or a failed start poisoned an - // idle instance): 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 its stopping frame cleanly. + // 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. EXIT.cancelled().await; if SIGNAL_SHUTDOWN.load(Ordering::Acquire) { if tokio::time::timeout(signal_drain_timeout(), runtime.shutdown()) From c025c644e1dfd0ff38784bbab441da3392d47bd6 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 | 24 ++++++++++++------- 1 file changed, 15 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..cbb8737b63 100644 --- a/rivetkit-rust/packages/rivetkit-core/src/actor/task.rs +++ b/rivetkit-rust/packages/rivetkit-core/src/actor/task.rs @@ -1801,6 +1801,20 @@ impl ActorTask { step = "sync_alarm", "actor shutdown cleanup step completed" ); + // Destroy cancels the engine alarm here so the persist it spawns is awaited by + // `wait_for_pending_alarm_writes` below and cannot race the SQLite teardown. + 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 +1848,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 99da275d612a361c6df0afc115245eb36a6ae481 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 | 30 +++++++++++++++++++++++++++++- 1 file changed, 29 insertions(+), 1 deletion(-) diff --git a/container-runner/src/main.rs b/container-runner/src/main.rs index 35b42dd14f..ad9a5ba07a 100644 --- a/container-runner/src/main.rs +++ b/container-runner/src/main.rs @@ -75,6 +75,11 @@ static EXIT: LazyLock = LazyLock::new(CancellationToken::new) /// (their budget is the pool's drain grace period instead). static SIGNAL_SHUTDOWN: AtomicBool = AtomicBool::new(false); +/// Set when shutdown was triggered by a platform SIGTERM (an instance reclaim), +/// as opposed to a local SIGINT (developer Ctrl-C). Distinguishes a reclaim +/// from a manual stop for downstream shutdown handling. +static PLATFORM_RECLAIM: AtomicBool = AtomicBool::new(false); + /// How long the platform gives this container between SIGTERM and SIGKILL. /// Defaults to 10 seconds; keep this in sync with the platform's actual budget /// via RIVET_SIGTERM_BUDGET_SECS. @@ -418,7 +423,30 @@ fn spawn_signal_handler() { let mut sigterm = signal(SignalKind::terminate()).expect("install SIGTERM handler"); let mut sigint = signal(SignalKind::interrupt()).expect("install SIGINT handler"); tokio::select! { - _ = sigterm.recv() => tracing::info!("received SIGTERM"), + _ = sigterm.recv() => { + PLATFORM_RECLAIM.store(true, Ordering::Release); + // Attribute the reclaim to each running actor so it is visible in + // actor-scoped logs, not only the process-level log stream. + let mut actor_ids = Vec::new(); + CHILDREN + .retain_async(|actor_id, _| { + actor_ids.push(actor_id.clone()); + true + }) + .await; + if actor_ids.is_empty() { + tracing::error!( + "unexpected platform SIGTERM received, likely hitting OOM or running longer than 60 minutes" + ); + } else { + for actor_id in actor_ids { + tracing::error!( + actor_id = %actor_id, + "unexpected platform SIGTERM received, likely hitting OOM or running longer than 60 minutes" + ); + } + } + } _ = sigint.recv() => tracing::info!("received SIGINT"), } SIGNAL_SHUTDOWN.store(true, Ordering::Release); From c78a0b786cab1b5992c80efff4b2ace0db15e480 Mon Sep 17 00:00:00 2001 From: ABCxFF <79597906+abcxff@users.noreply.github.com> Date: Fri, 31 Jul 2026 05:55:26 +0000 Subject: [PATCH 7/7] feat(container-runner): report actors as crashed on unexpected platform SIGTERM --- container-runner/src/actor.rs | 46 ++++++++++++++++++++++++++++++++++- container-runner/src/main.rs | 11 +++++++++ 2 files changed, 56 insertions(+), 1 deletion(-) diff --git a/container-runner/src/actor.rs b/container-runner/src/actor.rs index 80d25cda59..0a278c81b5 100644 --- a/container-runner/src/actor.rs +++ b/container-runner/src/actor.rs @@ -7,7 +7,7 @@ //! `on_destroy` stops the child while the instance stays warm for the next //! placement. -use std::sync::Arc; +use std::sync::{Arc, LazyLock}; use anyhow::{Context, Result}; use async_trait::async_trait; @@ -21,6 +21,12 @@ use crate::{ runner_config, }; +/// Live actor contexts on this instance, keyed by actor id. Lets the process +/// shutdown path report actors as crashed when the platform reclaims the +/// container out from under them. +static ACTOR_CTXS: LazyLock>> = + LazyLock::new(scc::HashMap::new); + pub struct GameServer { child: TokioMutex>>, } @@ -35,6 +41,7 @@ impl GameServer { // deliberate, then stop. `stop` is idempotent if the process shutdown // sweep already stopped this child. children().remove_async(actor_id).await; + ACTOR_CTXS.remove_async(actor_id).await; let child = self.child.lock().await.take(); if let Some(child) = child { child.stop(effective_stop_grace()).await; @@ -87,6 +94,7 @@ impl Actor for GameServer { "{} runner: actor already running, ignoring duplicate start", log_prefix(&actor_id, existing.key.as_deref()) ); + register_ctx(&actor_id, &ctx).await; *self.child.lock().await = Some(existing); return Ok(()); } @@ -152,6 +160,10 @@ impl Actor for GameServer { release_child_port(child_port).await; anyhow::bail!("a child for actor {actor_id} is already registered"); } + // Register only now that startup has succeeded. Registering earlier would + // leak an entry for any generation whose start failed, since a failed + // start never runs on_destroy/on_sleep to remove it. + register_ctx(&actor_id, &ctx).await; *self.child.lock().await = Some(child); Ok(()) } @@ -236,6 +248,38 @@ impl Actor for GameServer { } } +/// Register an actor context for crash-on-shutdown reporting. Overwrites any +/// stale entry left by a prior generation with the same id. +async fn register_ctx(actor_id: &str, ctx: &Ctx) { + ACTOR_CTXS.remove_async(actor_id).await; + let _ = ACTOR_CTXS + .insert_async(actor_id.to_string(), ctx.clone()) + .await; +} + +/// Report every live actor on this instance as crashed. Called when the +/// platform reclaims the container (an unexpected SIGTERM) so the reclaim +/// surfaces as a crash on the engine instead of a silent reallocation. Runs +/// while the envoy is still connected so the crash reaches the engine. +pub async fn crash_all_actors(message: &str) { + let mut ctxs = Vec::new(); + ACTOR_CTXS + .retain_async(|_, ctx| { + ctxs.push(ctx.clone()); + false + }) + .await; + for ctx in ctxs { + if let Err(err) = ctx.stop_with_error(message) { + tracing::debug!( + actor_id = %ctx.actor_id(), + error = ?err, + "crash-on-shutdown stop_with_error failed" + ); + } + } +} + fn actor_key_string(ctx: &Ctx) -> Option { let key = ctx.key(); if key.is_empty() { diff --git a/container-runner/src/main.rs b/container-runner/src/main.rs index ad9a5ba07a..afb79de192 100644 --- a/container-runner/src/main.rs +++ b/container-runner/src/main.rs @@ -369,6 +369,17 @@ async fn async_main() -> Result<()> { // and the runtime drains unbounded so the /start SSE flushes cleanly. EXIT.cancelled().await; if SIGNAL_SHUTDOWN.load(Ordering::Acquire) { + // A platform SIGTERM reclaims this instance. Report every actor as crashed + // before draining so an unexpected SIGTERM (OOM or the ~60 minute request + // cap) surfaces as a crash on the engine instead of a silent reallocation. + // This runs while the envoy is still connected so the crash reaches the + // engine. A local SIGINT (Ctrl-C) drains gracefully without a crash. + if PLATFORM_RECLAIM.load(Ordering::Acquire) { + crate::actor::crash_all_actors( + "runner received unexpected platform SIGTERM, likely OOM or running longer than 60 minutes", + ) + .await; + } if tokio::time::timeout(signal_drain_timeout(), runtime.shutdown()) .await .is_err()