fix: figure out actor log shipping

packages/common/util/core/Cargo.toml

@@ -13,7 +13,6 @@ macros = []
 async-trait = "0.1"
 bcrypt = "0.13.0"
 chrono = "0.4"
-fdb-util.workspace = true
 formatted-error = { workspace = true, optional = true }
 futures-util = "0.3"
 global-error.workspace = true
@@ -25,6 +24,7 @@ regex = "1.4"
 reqwest = { version = "0.12", default-features = false }
 rivet-config.workspace = true
 rivet-metrics.workspace = true
+rivet-util-id = { workspace = true }
 rivet-util-macros.workspace = true
 serde = { version = "1.0", features = ["derive"] }
 serde_json = { version = "1.0" }

packages/common/util/core/src/lib.rs

@@ -1,6 +1,7 @@
-pub use id::Id;
 use rand::Rng;
 pub use rivet_util_macros as macros;
+pub use rivet_util_id as id;
+pub use id::Id;
 use tokio::time::{Duration, Instant};
 
 pub mod billing;
@@ -13,7 +14,6 @@ pub mod format;
 pub mod future;
 pub mod geo;
 pub mod glob;
-pub mod id;
 pub mod math;
 pub mod req;
 pub mod route;

packages/common/util/id/Cargo.toml

@@ -0,0 +1,22 @@
+[package]
+name = "rivet-util-id"
+version.workspace = true
+authors.workspace = true
+license.workspace = true
+edition.workspace = true
+
+[dependencies]
+fdb-util.workspace = true
+serde = { version = "1.0", features = ["derive"] }
+thiserror = "1.0"
+uuid = { version = "1", features = ["v4", "serde"] }
+
+[dependencies.sqlx]
+workspace = true
+features = [
+	"runtime-tokio",
+	"postgres",
+	"uuid",
+	"json",
+	"ipnetwork"
+]

packages/edge/infra/client/container-runner/Cargo.toml

@@ -13,6 +13,7 @@ path = "src/main.rs"
 anyhow.workspace = true
 nix.workspace = true
 rivet-logs.workspace = true
+rivet-util.workspace = true
 serde = { version = "1.0.195", features = ["derive"] }
 serde_json = "1.0.111"
 signal-hook = "0.3.17"

packages/edge/infra/client/container-runner/Dockerfile

@@ -1,15 +1,26 @@
-FROM clux/muslrust:1.81.0-stable AS rust
+# IMPORTANT: This version is required for GLIBC 2.31 (used by edge servers on Linode)
+FROM rust:1.82.0-bullseye AS rust
 
 WORKDIR /app
 COPY . .
 
+# Installs shared libs
+#
+# The FDB version should match `cluster::workflows::server::install::install_scripts::components::fdb::FDB_VERSION`
+RUN \
+	apt-get update -y && \
+	apt-get install -y \
+	libclang-dev protobuf-compiler && \
+	curl -Lf -o /lib/libfdb_c.so "https://github.com/apple/foundationdb/releases/download/7.1.60/libfdb_c.x86_64.so"
+
 RUN \
 	--mount=type=cache,target=/root/.cargo/git \
 	--mount=type=cache,target=/root/.cargo/registry \
 	--mount=type=cache,target=/app/target \
-	RUSTFLAGS="--cfg tokio_unstable" cargo build --bin rivet-container-runner --release && \
+	# TODO: release is too slow
+	RUSTFLAGS="--cfg tokio_unstable" cargo build --target x86_64-unknown-linux-gnu --bin rivet-container-runner && \
 	mkdir -p /app/dist && \
-	mv /app/target/x86_64-unknown-linux-musl/release/rivet-container-runner /app/dist/rivet-container-runner
+	mv /app/target/x86_64-unknown-linux-gnu/debug/rivet-container-runner /app/dist/rivet-container-runner
 
 # Create an empty image and copy binaries into it to minimize the size of the image
 FROM scratch

packages/edge/infra/client/container-runner/src/log_shipper.rs

@@ -93,9 +93,39 @@ impl LogShipper {
 		println!("Log shipper connected");
 
 		while let Result::Ok(message) = self.msg_rx.recv() {
+			// // If actor id is not provided, extract from logs
+			// let actor_id = if self.actor_id.is_some() {
+			// 	self.actor_id.as_deref()
+			// } else {
+			// 	if let Some(start_idx) = message.message.find("actor_") {
+			// 		let start_idx = start_idx + 6;
+
+			// 		// Look for next non alphanum (end of actor id)
+			// 		let end_idx = if let Some(end_idx) =
+			// 			message.message[start_idx..].find(|c: char| !c.is_ascii_alphanumeric())
+			// 		{
+			// 			start_idx + end_idx
+			// 		} else {
+			// 			message.message.len()
+			// 		};
+
+			// 		let actor_id = &message.message[start_idx..end_idx];
+
+			// 		// Check if valid id
+			// 		rivet_util::Id::parse(actor_id).is_ok().then_some(actor_id)
+			// 	} else {
+			// 		None
+			// 	}
+			// };
+
+			// // Cannot determine actor id, ignore log
+			// let Some(actor_id) = actor_id else {
+			// 	continue;
+			// };
+
 			let vector_message = VectorMessage::Actors {
-				// runner_id: self.runner_id.as_str(),
-				actor_id: self.actor_id.as_ref().map(|x| x.as_str()),
+				runner_id: self.runner_id.as_str(),
+				actor_id: self.actor_id.as_deref(),
 				env_id: self.env_id,
 				stream_type: message.stream_type as u8,
 				ts: message.ts,
@@ -118,7 +148,7 @@ impl LogShipper {
 enum VectorMessage<'a> {
 	#[serde(rename = "actors")]
 	Actors {
-		// runner_id: &'a str,
+		runner_id: &'a str,
 		actor_id: Option<&'a str>,
 		env_id: Uuid,
 		stream_type: u8,

packages/edge/infra/client/echo/Dockerfile

@@ -17,20 +17,16 @@ RUN useradd -m -d /home/nonroot -s /bin/sh nonroot
 RUN grep nonroot /etc/passwd > /passwd && \
     grep nonroot /etc/group > /group
 
-USER nonroot
-
-CMD ["/app/dist/pegboard-echo-server"]
+# Create an empty image and copy binaries into it to minimize the size of the image
+FROM scratch
 
-# # Create an empty image and copy binaries into it to minimize the size of the image
-# FROM scratch
+# Copy passwd and group files
+COPY --from=rust /passwd /etc/passwd
+COPY --from=rust /group /etc/group
 
-# # Copy passwd and group files
-# COPY --from=rust /passwd /etc/passwd
-# COPY --from=rust /group /etc/group
+COPY --from=rust /app/dist/ /
 
-# COPY --from=rust /app/dist/ /
-
-# # Switch to the non-root user
-# USER nonroot
+# Switch to the non-root user
+USER nonroot
 
-# CMD ["/pegboard-echo-server"]
+CMD ["/pegboard-echo-server"]

packages/edge/infra/client/echo/src/main.rs

@@ -143,7 +143,7 @@ fn encode_frame<T: Serialize>(payload: &T) -> Result<Bytes> {
 	let mut buf = Vec::with_capacity(4);
 	let mut cursor = Cursor::new(&mut buf);
 
-	cursor.write(&[0u8; 4]); // header (currently unused)
+	cursor.write(&[0u8; 4])?; // header (currently unused)
 
 	serde_json::to_writer(&mut cursor, payload)?;
 

packages/edge/infra/client/manager/tests/container_lifecycle.rs

@@ -8,6 +8,7 @@ use pegboard::protocol;
 use pegboard_manager::Ctx;
 use tokio::{net::TcpStream, sync::Mutex};
 use tokio_tungstenite::tungstenite::protocol::Message;
+use uuid::Uuid;
 
 mod common;
 use common::*;
@@ -184,6 +185,16 @@ async fn handle_connection(
 											);
 										}
 
+										// Stop actor
+										send_command(
+											&mut tx,
+											protocol::Command::SignalRunner {
+												runner_id: Uuid::nil(),
+												signal: Signal::SIGKILL as i32,
+											},
+										)
+										.await;
+
 										tokio::time::sleep(Duration::from_millis(1000)).await;
 
 										// Verify client state

packages/edge/infra/client/manager/tests/vector.json

@@ -34,7 +34,7 @@
 			"inputs": [
 				"actors"
 			],
-			"source": ".message, err = \"\u001b[2m\" + \"actor_id=\" + .actor_id + \"\u001b[0m \" + .message"
+			"source": ".message, err = \"\u001b[2m\" + \"runner_id=\" + .runner_id + \"\u001b[0m \" + .message"
 		}
 	},
 	"sinks": {