[SLOP(claude-opus-4-8)] feat(pegboard-envoy): rate-limit envoy ws ingress and cap get_pages, trim unused metrics

Cargo.toml

@@ -672,6 +672,16 @@ lto = "fat"
 codegen-units = 1
 opt-level = 3
 
+# Release-grade optimization with DWARF line info and no symbol stripping so
+# heaptrack can resolve native allocation backtraces during leak investigation.
+# Uses thin LTO and more codegen units to keep frames un-inlined and builds fast.
+[profile.profiling]
+inherits = "release"
+debug = 1
+lto = "thin"
+codegen-units = 16
+strip = false
+
 [profile.quick]
 inherits = "dev"
 debug = false          # no debug info → faster link, smaller binary

engine/packages/api-peer/src/internal.rs

@@ -62,7 +62,7 @@ pub async fn set_tracing_config(
 	body: SetTracingConfigRequest,
 ) -> Result<SetTracingConfigResponse> {
 	// Broadcast message to all services via UPS
-	let message = serde_json::to_vec(&body)?;
+	let message = rivet_util::serde::json_to_vec!(&body)?;
 
 	ctx.ups()?
 		.publish(TracingConfigSubject, &message, PublishOpts::broadcast())

engine/packages/cache/src/req_config.rs

@@ -365,7 +365,7 @@ impl RequestConfig {
 				keys: cache_keys.clone(),
 			};
 
-			let payload = serde_json::to_vec(&message)?;
+			let payload = rivet_util::serde::json_to_vec!(&message)?;
 
 			if let Err(err) = ups
 				.publish(
@@ -495,12 +495,12 @@ impl RequestConfig {
 			keys,
 			getter,
 			|value: &Value| -> Result<Vec<u8>> {
-				serde_json::to_vec(&value)
+				rivet_util::serde::json_to_vec!(&value)
 					.map_err(Error::SerdeEncode)
 					.map_err(Into::into)
 			},
 			|value: &[u8]| -> Result<Value> {
-				serde_json::from_slice(value)
+				rivet_util::serde::json_from_slice!(value)
 					.map_err(Error::SerdeDecode)
 					.map_err(Into::into)
 			},

engine/packages/config/src/config/pegboard.rs

@@ -117,6 +117,12 @@ pub struct Pegboard {
 	pub gateway_hws_max_pending_size: Option<u64>,
 	/// Max HTTP request body size in bytes for requests to actors.
 	pub gateway_http_max_request_body_size: Option<usize>,
+	/// Max burst of inbound WebSocket messages on a single connection before throttling.
+	pub gateway_websocket_rate_limit_requests: Option<u64>,
+	/// Time to regain one inbound WebSocket message token on a single connection.
+	///
+	/// Unit is in milliseconds.
+	pub gateway_websocket_rate_limit_drip_rate_ms: Option<u64>,
 
 	// === Envoy Settings ===
 	/// How long to wait before considering an envoy lost and evicting all of its actors.
@@ -143,6 +149,14 @@ pub struct Pegboard {
 	pub envoy_expire_scheduler_max_concurrent_expires: Option<usize>,
 	/// Maximum pending envoys tracked by the read-path envoy expire scheduler.
 	pub envoy_expire_scheduler_max_pending: Option<usize>,
+	/// Max burst of inbound WebSocket messages on a single envoy connection before throttling.
+	pub envoy_websocket_rate_limit_requests: Option<u64>,
+	/// Time to regain one inbound WebSocket message token on a single envoy connection.
+	///
+	/// Unit is in microseconds. The envoy connection multiplexes every actor on a runner, so the
+	/// sustained ceiling is far higher than the per-client gateway limit and needs sub-millisecond
+	/// granularity to express.
+	pub envoy_websocket_rate_limit_drip_rate_us: Option<u64>,
 
 	// === Serverless Settings ===
 	/// **Deprecated** Configure the drain period in the runner config.
@@ -162,6 +176,14 @@ pub struct Pegboard {
 	///
 	/// Unit is in bytes. Default: 1,048,576 (1 MiB).
 	pub preload_max_total_bytes: Option<u64>,
+
+	// === Rate Limiting ===
+	/// Max burst of actor creations per namespace before throttling.
+	pub actor_create_rate_limit_requests: Option<u64>,
+	/// Time to regain one actor creation token per namespace.
+	///
+	/// Unit is in milliseconds.
+	pub actor_create_rate_limit_drip_rate_ms: Option<u64>,
 }
 
 impl Pegboard {
@@ -369,6 +391,30 @@ impl Pegboard {
 		self.serverless_drain_grace_period.unwrap_or(10_000)
 	}
 
+	pub fn gateway_websocket_rate_limit_requests(&self) -> u64 {
+		self.gateway_websocket_rate_limit_requests.unwrap_or(2_000)
+	}
+
+	pub fn gateway_websocket_rate_limit_drip_rate_ms(&self) -> u64 {
+		self.gateway_websocket_rate_limit_drip_rate_ms.unwrap_or(10)
+	}
+
+	pub fn envoy_websocket_rate_limit_requests(&self) -> u64 {
+		self.envoy_websocket_rate_limit_requests.unwrap_or(16_384)
+	}
+
+	pub fn envoy_websocket_rate_limit_drip_rate_us(&self) -> u64 {
+		self.envoy_websocket_rate_limit_drip_rate_us.unwrap_or(200)
+	}
+
+	pub fn actor_create_rate_limit_requests(&self) -> u64 {
+		self.actor_create_rate_limit_requests.unwrap_or(500)
+	}
+
+	pub fn actor_create_rate_limit_drip_rate_ms(&self) -> u64 {
+		self.actor_create_rate_limit_drip_rate_ms.unwrap_or(10)
+	}
+
 	pub fn preload_max_total_bytes(&self) -> u64 {
 		self.preload_max_total_bytes.unwrap_or(1_048_576)
 	}

engine/packages/depot/Cargo.toml

@@ -30,19 +30,19 @@ rivet-error.workspace = true
 rivet-metrics.workspace = true
 rivet-pools.workspace = true
 rivet-runtime.workspace = true
+rivet-util.workspace = true
+rusqlite.workspace = true
 scc.workspace = true
-serde.workspace = true
 serde_bare.workspace = true
 serde_json.workspace = true
+serde.workspace = true
 sha2.workspace = true
-rusqlite.workspace = true
 tempfile.workspace = true
 tokio.workspace = true
 tokio-util.workspace = true
 tracing.workspace = true
 universaldb.workspace = true
 universalpubsub.workspace = true
-util.workspace = true
 uuid.workspace = true
 vbare.workspace = true
 

engine/packages/depot/src/conveyer/types/branch.rs

@@ -77,14 +77,14 @@ impl OwnedVersionedData for VersionedDatabaseBranchRecord {
 
 	fn deserialize_version(payload: &[u8], version: u16) -> Result<Self> {
 		match version {
-			1 => Ok(Self::Current(serde_bare::from_slice(payload)?)),
+			1 => Ok(Self::Current(rivet_util::serde::bare_from_slice!(payload)?)),
 			_ => bail!("invalid depot DatabaseBranchRecord version: {version}"),
 		}
 	}
 
 	fn serialize_version(self, _version: u16) -> Result<Vec<u8>> {
 		match self {
-			Self::Current(data) => serde_bare::to_vec(&data).map_err(Into::into),
+			Self::Current(data) => rivet_util::serde::bare_to_vec!(&data).map_err(Into::into),
 		}
 	}
 }
@@ -108,14 +108,14 @@ impl OwnedVersionedData for VersionedDatabasePointer {
 
 	fn deserialize_version(payload: &[u8], version: u16) -> Result<Self> {
 		match version {
-			1 => Ok(Self::V1(serde_bare::from_slice(payload)?)),
+			1 => Ok(Self::V1(rivet_util::serde::bare_from_slice!(payload)?)),
 			_ => bail!("invalid depot DatabasePointer version: {version}"),
 		}
 	}
 
 	fn serialize_version(self, _version: u16) -> Result<Vec<u8>> {
 		match self {
-			Self::V1(data) => serde_bare::to_vec(&data).map_err(Into::into),
+			Self::V1(data) => rivet_util::serde::bare_to_vec!(&data).map_err(Into::into),
 		}
 	}
 }
@@ -139,14 +139,14 @@ impl OwnedVersionedData for VersionedBucketBranchRecord {
 
 	fn deserialize_version(payload: &[u8], version: u16) -> Result<Self> {
 		match version {
-			1 => Ok(Self::V1(serde_bare::from_slice(payload)?)),
+			1 => Ok(Self::V1(rivet_util::serde::bare_from_slice!(payload)?)),
 			_ => bail!("invalid depot BucketBranchRecord version: {version}"),
 		}
 	}
 
 	fn serialize_version(self, _version: u16) -> Result<Vec<u8>> {
 		match self {
-			Self::V1(data) => serde_bare::to_vec(&data).map_err(Into::into),
+			Self::V1(data) => rivet_util::serde::bare_to_vec!(&data).map_err(Into::into),
 		}
 	}
 }
@@ -170,14 +170,14 @@ impl OwnedVersionedData for VersionedBucketPointer {
 
 	fn deserialize_version(payload: &[u8], version: u16) -> Result<Self> {
 		match version {
-			1 => Ok(Self::V1(serde_bare::from_slice(payload)?)),
+			1 => Ok(Self::V1(rivet_util::serde::bare_from_slice!(payload)?)),
 			_ => bail!("invalid depot BucketPointer version: {version}"),
 		}
 	}
 
 	fn serialize_version(self, _version: u16) -> Result<Vec<u8>> {
 		match self {
-			Self::V1(data) => serde_bare::to_vec(&data).map_err(Into::into),
+			Self::V1(data) => rivet_util::serde::bare_to_vec!(&data).map_err(Into::into),
 		}
 	}
 }
@@ -201,14 +201,14 @@ impl OwnedVersionedData for VersionedPointerSnapshot {
 
 	fn deserialize_version(payload: &[u8], version: u16) -> Result<Self> {
 		match version {
-			1 => Ok(Self::V1(serde_bare::from_slice(payload)?)),
+			1 => Ok(Self::V1(rivet_util::serde::bare_from_slice!(payload)?)),
 			_ => bail!("invalid depot PointerSnapshot version: {version}"),
 		}
 	}
 
 	fn serialize_version(self, _version: u16) -> Result<Vec<u8>> {
 		match self {
-			Self::V1(data) => serde_bare::to_vec(&data).map_err(Into::into),
+			Self::V1(data) => rivet_util::serde::bare_to_vec!(&data).map_err(Into::into),
 		}
 	}
 }

engine/packages/depot/src/conveyer/types/compaction.rs

@@ -61,14 +61,14 @@ macro_rules! impl_compaction_versioned_data {
 
 			fn deserialize_version(payload: &[u8], version: u16) -> Result<Self> {
 				match version {
-					1 => Ok(Self::V1(serde_bare::from_slice(payload)?)),
+					1 => Ok(Self::V1(rivet_util::serde::bare_from_slice!(payload)?)),
 					_ => bail!("invalid depot {} version: {version}", $name),
 				}
 			}
 
 			fn serialize_version(self, _version: u16) -> Result<Vec<u8>> {
 				match self {
-					Self::V1(data) => serde_bare::to_vec(&data).map_err(Into::into),
+					Self::V1(data) => rivet_util::serde::bare_to_vec!(&data).map_err(Into::into),
 				}
 			}
 		}

engine/packages/depot/src/conveyer/types/history_pin.rs

@@ -43,14 +43,14 @@ impl OwnedVersionedData for VersionedDbHistoryPin {
 
 	fn deserialize_version(payload: &[u8], version: u16) -> Result<Self> {
 		match version {
-			1 => Ok(Self::V1(serde_bare::from_slice(payload)?)),
+			1 => Ok(Self::V1(rivet_util::serde::bare_from_slice!(payload)?)),
 			_ => bail!("invalid depot DbHistoryPin version: {version}"),
 		}
 	}
 
 	fn serialize_version(self, _version: u16) -> Result<Vec<u8>> {
 		match self {
-			Self::V1(data) => serde_bare::to_vec(&data).map_err(Into::into),
+			Self::V1(data) => rivet_util::serde::bare_to_vec!(&data).map_err(Into::into),
 		}
 	}
 }