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/9] 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/9] 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/9] 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/9] 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/9] 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/9] 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/9] 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() From 4451538ea0c7e60d2eeb770853104e4eba6d968f Mon Sep 17 00:00:00 2001 From: ABCxFF <79597906+abcxff@users.noreply.github.com> Date: Fri, 31 Jul 2026 17:05:05 +0000 Subject: [PATCH 8/9] fix(rivetkit): keep actor event loop alive through shutdown state serialization --- rivetkit-rust/packages/rivetkit/src/start.rs | 40 +++++++++++++++++++- 1 file changed, 39 insertions(+), 1 deletion(-) diff --git a/rivetkit-rust/packages/rivetkit/src/start.rs b/rivetkit-rust/packages/rivetkit/src/start.rs index c1092cbf84..32e6c56d57 100644 --- a/rivetkit-rust/packages/rivetkit/src/start.rs +++ b/rivetkit-rust/packages/rivetkit/src/start.rs @@ -426,7 +426,7 @@ async fn handle_actor_event( ShutdownKind::Destroy => actor.on_destroy(ctx).await, }; reply.send(result); - return Ok(true); + // Keep the loop alive so the finalize-phase `SerializeState` can capture state the hook wrote; core ends it by closing the channel. } ActorEvent::DisconnectConn { conn_id, reply } => { reply.send(ctx.disconnect_conn(&conn_id).await); @@ -912,6 +912,7 @@ mod tests { ); request_sleep(&tx).await; + drop(tx); actor.await.expect("join run_actor").expect("run actor"); } @@ -933,6 +934,7 @@ mod tests { ); request_sleep(&tx).await; + drop(tx); actor.await.expect("join run_actor").expect("run actor"); } @@ -953,6 +955,7 @@ mod tests { assert_eq!(response.status().as_u16(), 404); request_sleep(&tx).await; + drop(tx); actor.await.expect("join run_actor").expect("run actor"); } @@ -966,6 +969,7 @@ mod tests { assert_eq!(state.created, 1); request_sleep(&tx).await; + drop(tx); actor.await.expect("join run_actor").expect("run actor"); } @@ -1020,6 +1024,7 @@ mod tests { assert!(error.to_string().contains("websockets not supported")); request_sleep(&tx).await; + drop(tx); actor.await.expect("join run_actor").expect("run actor"); } @@ -1055,6 +1060,7 @@ mod tests { tx.send(ActorEvent::ConnectionClosed { conn }) .expect("send connection closed"); request_sleep(&tx).await; + drop(tx); actor.await.expect("join run_actor").expect("run actor"); let log = &ctx.state().log; @@ -1084,6 +1090,7 @@ mod tests { assert!(error.to_string().contains("subscribe denied")); request_sleep(&tx).await; + drop(tx); actor.await.expect("join run_actor").expect("run actor"); let log = &ctx.state().log; @@ -1118,6 +1125,7 @@ mod tests { assert!(error.to_string().contains("connection rejected")); request_sleep(&tx).await; + drop(tx); actor.await.expect("join run_actor").expect("run actor"); assert!( !ctx.state() @@ -1135,6 +1143,7 @@ mod tests { let actor = tokio::spawn(run_actor::(start)); request_sleep(&tx).await; + drop(tx); actor.await.expect("join run_actor").expect("run actor"); assert!(ctx.state().log.iter().any(|entry| entry == "run_aborted")); @@ -1148,11 +1157,34 @@ mod tests { let actor = tokio::spawn(run_actor::(start)); request_destroy(&tx).await; + drop(tx); actor.await.expect("join run_actor").expect("run actor"); assert!(ctx.state().log.iter().any(|entry| entry == "on_destroy")); } + #[tokio::test] + async fn run_actor_serializes_state_after_cleanup() { + // Core sends `SerializeState` during finalize, after the cleanup hook, to + // capture state the hook wrote. The event loop must stay alive to handle + // it instead of ending on the cleanup event. + let (tx, rx) = unbounded_channel(); + let start = lifecycle_start(Some(cbor(&LifecycleInput { count: 5 })), None, rx.into()); + let actor = tokio::spawn(run_actor::(start)); + + request_sleep(&tx).await; + let state = decode_actor_state(request_serialize(&tx).await); + assert_eq!(state.count, 5); + assert!( + state.log.iter().any(|entry| entry == "on_sleep"), + "serialize after cleanup lost on_sleep state, got: {:?}", + state.log + ); + + drop(tx); + actor.await.expect("join run_actor").expect("run actor"); + } + #[tokio::test] async fn run_actor_error_requests_errored_stop() { let (tx, rx) = unbounded_channel(); @@ -1300,6 +1332,7 @@ mod tests { ); request_sleep(&tx).await; + drop(tx); actor.await.expect("join run_actor").expect("run actor"); } @@ -1338,6 +1371,7 @@ mod tests { ); request_sleep(&tx).await; + drop(tx); actor.await.expect("join run_actor").expect("run actor"); } @@ -1365,6 +1399,7 @@ mod tests { assert_eq!(error.code(), "action_not_found"); request_sleep(&tx).await; + drop(tx); actor.await.expect("join run_actor").expect("run actor"); } @@ -1394,6 +1429,7 @@ mod tests { ); request_sleep(&tx).await; + drop(tx); actor.await.expect("join run_actor").expect("run actor"); } @@ -1430,6 +1466,7 @@ mod tests { assert_eq!(error.code(), "not_found"); request_sleep(&tx).await; + drop(tx); actor.await.expect("join run_actor").expect("run actor"); } @@ -1471,6 +1508,7 @@ mod tests { ); request_sleep(&tx).await; + drop(tx); actor.await.expect("join run_actor").expect("run actor"); } From d0608e26d59b54a2ea3eb6e20e8f48147e246593 Mon Sep 17 00:00:00 2001 From: ABCxFF <79597906+abcxff@users.noreply.github.com> Date: Tue, 4 Aug 2026 20:55:24 +0000 Subject: [PATCH 9/9] feat(container-runner): setup instance memory and cpu usage logger --- container-runner/src/actor.rs | 11 ++ container-runner/src/main.rs | 16 ++ container-runner/src/monitor.rs | 314 ++++++++++++++++++++++++++++++++ 3 files changed, 341 insertions(+) create mode 100644 container-runner/src/monitor.rs diff --git a/container-runner/src/actor.rs b/container-runner/src/actor.rs index 0a278c81b5..560aab6497 100644 --- a/container-runner/src/actor.rs +++ b/container-runner/src/actor.rs @@ -85,6 +85,17 @@ 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. + tracing::info!( + actor_id = %actor_id, + resource_monitor_enabled = crate::monitor::enabled(), + resource_monitor_source = crate::monitor::sampling_source(), + "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. diff --git a/container-runner/src/main.rs b/container-runner/src/main.rs index afb79de192..8e11080bf8 100644 --- a/container-runner/src/main.rs +++ b/container-runner/src/main.rs @@ -19,6 +19,7 @@ mod actor; mod child; mod input; +mod monitor; mod proxy; use std::io::Read; @@ -116,6 +117,20 @@ pub fn children() -> &'static scc::HashMap> { &CHILDREN } +/// Actor ids currently hosting a running child on this instance. Used to +/// attribute the instance-wide resource samples to the running actor(s). +pub async fn active_actor_ids() -> Vec { + let mut ids = Vec::new(); + // `retain_async` returning `true` keeps every entry: a read-only scan. + CHILDREN + .retain_async(|actor_id, _| { + ids.push(actor_id.clone()); + true + }) + .await; + 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 @@ -336,6 +351,7 @@ async fn async_main() -> Result<()> { let serve_shutdown = CancellationToken::new(); spawn_signal_handler(); + monitor::spawn_resource_monitor(); let serve = tokio::spawn(serverless_http::serve( runtime.clone(), diff --git a/container-runner/src/monitor.rs b/container-runner/src/monitor.rs new file mode 100644 index 0000000000..11447a8189 --- /dev/null +++ b/container-runner/src/monitor.rs @@ -0,0 +1,314 @@ +//! 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. +//! +//! Two counter sources are supported, picked at startup: +//! - **cgroup v2** under `/sys/fs/cgroup` (real Linux). Exact against the +//! container limits. +//! - **`/proc`** (`/proc/meminfo`, `/proc/stat`), the fallback for a gVisor +//! sandbox, which does not expose cgroup v2. Values reflect the sandbox and +//! are approximate. +//! If neither is readable the monitor logs once and disables itself. + +use std::time::{Duration, Instant}; + +use tokio::time::{MissedTickBehavior, interval}; + +/// Environment variable that enables the monitor. Unset or a falsey value means +/// the monitor does not run and nothing is logged. Truthy values are `1`, +/// `true`, `yes`, and `on` (case-insensitive). +const ENABLE_ENV: &str = "RIVET_LOG_RESOURCE_USAGE"; + +/// How often to sample and log resource usage. +const SAMPLE_INTERVAL: Duration = Duration::from_millis(500); + +const MEMORY_CURRENT: &str = "/sys/fs/cgroup/memory.current"; +const MEMORY_MAX: &str = "/sys/fs/cgroup/memory.max"; +const CPU_STAT: &str = "/sys/fs/cgroup/cpu.stat"; +const CPU_MAX: &str = "/sys/fs/cgroup/cpu.max"; + +const PROC_MEMINFO: &str = "/proc/meminfo"; +const PROC_STAT: &str = "/proc/stat"; + +/// Kernel clock ticks per second, the unit of `/proc/stat` jiffies. 100 on Linux +/// and gVisor. +const USER_HZ: u64 = 100; + +/// Where the monitor reads counters from. +#[derive(Clone, Copy, PartialEq, Eq)] +enum Source { + /// cgroup v2 under `/sys/fs/cgroup` (real Linux). + CgroupV2, + /// `/proc` (gVisor sandbox, which does not expose cgroup v2). + Proc, +} + +impl Source { + fn label(self) -> &'static str { + match self { + Source::CgroupV2 => "cgroup_v2", + Source::Proc => "proc", + } + } +} + +/// Spawn the background resource monitor if enabled via [`ENABLE_ENV`]. A no-op +/// when disabled (the default). Fire-and-forget for the process lifetime; it +/// runs until the process exits. +pub fn spawn_resource_monitor() { + if !monitor_enabled() { + return; + } + tracing::info!(interval_ms = SAMPLE_INTERVAL.as_millis() as u64, "resource monitor enabled"); + tokio::spawn(run_monitor()); +} + +fn monitor_enabled() -> bool { + match std::env::var(ENABLE_ENV) { + Ok(value) => matches!( + value.trim().to_ascii_lowercase().as_str(), + "1" | "true" | "yes" | "on" + ), + Err(_) => false, + } +} + +/// Whether the monitor is turned on via [`ENABLE_ENV`]. Exposed so the actor can +/// report the monitor's status tagged with its id (process-level monitor logs +/// are otherwise invisible in actor-scoped log views). +pub fn enabled() -> bool { + monitor_enabled() +} + +/// The counter source the monitor would use right now: `"cgroup_v2"`, `"proc"`, +/// or `"none"` when neither is readable. Exposed for the actor-scoped status log. +pub fn sampling_source() -> &'static str { + match detect_source() { + Some(source) => source.label(), + None => "none", + } +} + +/// Pick the counter source, preferring exact cgroup v2 over the `/proc` fallback. +fn detect_source() -> Option { + if read_u64(MEMORY_CURRENT).is_some() && read_cgroup_cpu_usage_usec().is_some() { + Some(Source::CgroupV2) + } else if read_proc_memory().is_some() && read_proc_cpu_busy_usec().is_some() { + Some(Source::Proc) + } else { + None + } +} + +async fn run_monitor() { + let Some(source) = detect_source() else { + tracing::warn!( + cgroup_dir = "/sys/fs/cgroup", + proc_meminfo = PROC_MEMINFO, + proc_stat = PROC_STAT, + "resource monitor disabled: neither cgroup v2 nor /proc counters are readable" + ); + return; + }; + tracing::info!(source = source.label(), "resource monitor sampling"); + + let mut ticker = interval(SAMPLE_INTERVAL); + // A slow sample must not make the monitor try to catch up with a burst of + // back-to-back ticks; just resume on the next boundary. + ticker.set_missed_tick_behavior(MissedTickBehavior::Skip); + // The first tick fires immediately; consume it so the first logged line + // already covers a full interval of CPU time. + ticker.tick().await; + + // Limits are fixed for the instance lifetime, so read them once. + let mem_limit_mib = memory_limit_bytes(source).map(bytes_to_mib); + let cpu_limit_cores = cpu_limit_cores(source); + + let mut prev_cpu_usec = cpu_busy_usec(source); + let mut prev_at = Instant::now(); + + loop { + ticker.tick().await; + let now = Instant::now(); + let elapsed = now.duration_since(prev_at); + + let cpu_now_usec = cpu_busy_usec(source); + // CPU time used over the interval, divided by wall time, is the number of + // vCPU cores consumed (1.0 == one core fully busy). + let cpu_cores = match (prev_cpu_usec, cpu_now_usec) { + (Some(prev), Some(cur)) if !elapsed.is_zero() => { + cur.saturating_sub(prev) as f64 / elapsed.as_micros() as f64 + } + _ => 0.0, + }; + prev_cpu_usec = cpu_now_usec; + prev_at = now; + + // Only log while an actor is running, and attribute the sample to it so it + // shows up in that actor's logs. The counters are instance-wide; with more + // than one actor on the instance the same sample is logged for each. + let actor_ids = crate::active_actor_ids().await; + if actor_ids.is_empty() { + continue; + } + + let mem_used_mib = memory_used_bytes(source).map(bytes_to_mib); + let mem_pct = match (mem_used_mib, mem_limit_mib) { + (Some(used), Some(limit)) if limit > 0.0 => Some(used / limit * 100.0), + _ => None, + }; + let cpu_pct = match cpu_limit_cores { + Some(limit) if limit > 0.0 => Some(cpu_cores / limit * 100.0), + _ => None, + }; + + for actor_id in actor_ids { + tracing::info!( + actor_id = %actor_id, + source = source.label(), + mem_used_mib = ?mem_used_mib, + mem_limit_mib = ?mem_limit_mib, + mem_pct = ?mem_pct, + cpu_cores, + cpu_limit_cores = ?cpu_limit_cores, + cpu_pct = ?cpu_pct, + "instance resource usage" + ); + } + } +} + +/// Current memory usage in bytes for the selected source. +fn memory_used_bytes(source: Source) -> Option { + match source { + Source::CgroupV2 => read_u64(MEMORY_CURRENT), + Source::Proc => read_proc_memory().map(|(used, _limit)| used), + } +} + +/// Memory limit in bytes for the selected source, or `None` when unlimited. +fn memory_limit_bytes(source: Source) -> Option { + match source { + Source::CgroupV2 => read_cgroup_memory_max(), + Source::Proc => read_proc_memory().map(|(_used, limit)| limit), + } +} + +/// Cumulative busy CPU time in microseconds for the selected source. +fn cpu_busy_usec(source: Source) -> Option { + match source { + Source::CgroupV2 => read_cgroup_cpu_usage_usec(), + Source::Proc => read_proc_cpu_busy_usec(), + } +} + +/// CPU limit in vCPU cores for the selected source, or `None` when unknown. +fn cpu_limit_cores(source: Source) -> Option { + match source { + Source::CgroupV2 => read_cgroup_cpu_limit_cores(), + Source::Proc => read_proc_cpu_count(), + } +} + +/// Read a pseudo-file holding a single unsigned integer. These are in-memory +/// kernel files, so the synchronous read does not block meaningfully. +fn read_u64(path: &str) -> Option { + std::fs::read_to_string(path).ok()?.trim().parse().ok() +} + +/// cgroup v2 memory limit in bytes, or `None` when unlimited (`memory.max` is +/// `"max"`). +fn read_cgroup_memory_max() -> Option { + let raw = std::fs::read_to_string(MEMORY_MAX).ok()?; + let raw = raw.trim(); + if raw == "max" { + None + } else { + raw.parse().ok() + } +} + +/// Cumulative CPU time consumed by the cgroup, in microseconds, from the +/// `usage_usec` line of `cpu.stat`. +fn read_cgroup_cpu_usage_usec() -> Option { + let stat = std::fs::read_to_string(CPU_STAT).ok()?; + stat.lines() + .find_map(|line| line.strip_prefix("usage_usec ")) + .and_then(|value| value.trim().parse().ok()) +} + +/// cgroup v2 CPU limit in vCPU cores from `cpu.max` (`" "`), or +/// `None` when unlimited (`quota` is `"max"`). +fn read_cgroup_cpu_limit_cores() -> Option { + let raw = std::fs::read_to_string(CPU_MAX).ok()?; + let mut parts = raw.split_whitespace(); + let quota = parts.next()?; + let period: f64 = parts.next()?.parse().ok()?; + if quota == "max" || period <= 0.0 { + return None; + } + let quota: f64 = quota.parse().ok()?; + Some(quota / period) +} + +/// `(used_bytes, total_bytes)` from `/proc/meminfo`. Used is `MemTotal - +/// MemAvailable`; total doubles as the limit under gVisor, where it reflects the +/// sandbox memory. +fn read_proc_memory() -> Option<(u64, u64)> { + let total_kb = read_meminfo_kb("MemTotal")?; + let available_kb = read_meminfo_kb("MemAvailable")?; + let used_kb = total_kb.saturating_sub(available_kb); + Some((used_kb * 1024, total_kb * 1024)) +} + +/// Value in kB of a `/proc/meminfo` key such as `"MemTotal"`. +fn read_meminfo_kb(key: &str) -> Option { + let content = std::fs::read_to_string(PROC_MEMINFO).ok()?; + content.lines().find_map(|line| { + let rest = line.strip_prefix(key)?; + rest.trim_start_matches(':') + .split_whitespace() + .next()? + .parse() + .ok() + }) +} + +/// Cumulative busy CPU time in microseconds from the aggregate `cpu` line of +/// `/proc/stat` (`total - idle - iowait`, converted from jiffies). +fn read_proc_cpu_busy_usec() -> Option { + let content = std::fs::read_to_string(PROC_STAT).ok()?; + let line = content.lines().next()?; + let mut fields = line.split_whitespace(); + if fields.next()? != "cpu" { + return None; + } + let values: Vec = fields.filter_map(|value| value.parse().ok()).collect(); + if values.len() < 4 { + return None; + } + let total: u64 = values.iter().sum(); + // Fields are user, nice, system, idle, iowait, ... Treat idle + iowait as idle. + let idle = values[3] + values.get(4).copied().unwrap_or(0); + let busy = total.saturating_sub(idle); + Some(busy * (1_000_000 / USER_HZ)) +} + +/// Number of vCPUs from the per-CPU (`cpu0`, `cpu1`, ...) lines of `/proc/stat`. +fn read_proc_cpu_count() -> Option { + let content = std::fs::read_to_string(PROC_STAT).ok()?; + // The aggregate line is `"cpu "` (trailing space); per-CPU lines are `"cpu0"`. + let count = content + .lines() + .filter(|line| line.starts_with("cpu") && !line.starts_with("cpu ")) + .count(); + (count > 0).then_some(count as f64) +} + +fn bytes_to_mib(bytes: u64) -> f64 { + bytes as f64 / (1024.0 * 1024.0) +}