fix: get pegboard working e2e

packages/common/chirp/worker/src/test.rs

@@ -61,6 +61,10 @@ impl TestCtx {
 		&self.op_ctx
 	}
 
+	pub fn config(&self) -> &rivet_config::Config {
+		self.op_ctx.config()
+	}
+
 	pub async fn crdb(&self) -> Result<CrdbPool, rivet_pools::Error> {
 		self.op_ctx.crdb().await
 	}

packages/infra/pegboard/manager/src/actor/mod.rs

@@ -1,8 +1,5 @@
 use std::{
-	sync::{
-		atomic::{AtomicBool, Ordering},
-		Arc,
-	},
+	sync::Arc,
 	time::Duration,
 };
 
@@ -32,7 +29,7 @@ pub struct Actor {
 	config: protocol::ActorConfig,
 
 	runner: Mutex<Option<runner::Handle>>,
-	exited: AtomicBool,
+	exited: Mutex<bool>,
 }
 
 impl Actor {
@@ -42,7 +39,7 @@ impl Actor {
 			config,
 
 			runner: Mutex::new(None),
-			exited: AtomicBool::new(false),
+			exited: Mutex::new(false),
 		})
 	}
 
@@ -56,7 +53,7 @@ impl Actor {
 			config,
 
 			runner: Mutex::new(Some(runner)),
-			exited: AtomicBool::new(false),
+			exited: Mutex::new(false),
 		})
 	}
 
@@ -392,6 +389,13 @@ impl Actor {
 
 	#[tracing::instrument(skip_all)]
 	pub async fn set_exit_code(&self, ctx: &Ctx, exit_code: Option<i32>) -> Result<()> {
+		let mut guard = self.exited.lock().await;
+
+		// Already exited
+		if *guard {
+			return Ok(());
+		}
+
 		// Update DB
 		utils::query(|| async {
 			sqlx::query(indoc!(
@@ -433,7 +437,7 @@ impl Actor {
 		})
 		.await?;
 
-		self.exited.store(true, Ordering::SeqCst);
+		*guard = true;
 
 		Ok(())
 	}
@@ -448,9 +452,8 @@ impl Actor {
 			actors.remove(&self.actor_id);
 		}
 
-		if !self.exited.load(Ordering::SeqCst) {
-			self.set_exit_code(ctx, None).await?;
-		}
+		// Set exit code if it hasn't already been set
+		self.set_exit_code(ctx, None).await?;
 
 		self.cleanup_setup(ctx).await
 	}

packages/infra/pegboard/manager/src/actor/setup.rs

@@ -434,7 +434,7 @@ impl Actor {
 			.map(|(_, port)| {
 				json!({
 					"HostPort": port.source,
-					"ActorPort": port.target,
+					"ContainerPort": port.target,
 					"Protocol": port.protocol.to_string(),
 				})
 			})
@@ -682,17 +682,41 @@ impl Actor {
 		let actor_path = ctx.actor_path(self.actor_id);
 		let netns_path = self.netns_path();
 
-		if let protocol::NetworkMode::Bridge = self.config.network_mode {
-			match fs::read_to_string(actor_path.join("cni-cap-args.json")).await {
-				Ok(cni_params_json) => {
-					match Command::new("cnitool")
+		match self.config.image.kind {
+			protocol::ImageKind::DockerImage | protocol::ImageKind::OciBundle => {
+				if let protocol::NetworkMode::Bridge = self.config.network_mode {
+					match fs::read_to_string(actor_path.join("cni-cap-args.json")).await {
+						Ok(cni_params_json) => {
+							match Command::new("cnitool")
+								.arg("del")
+								.arg(NETWORK_NAME)
+								.arg(netns_path)
+								.env("CNI_PATH", "/opt/cni/bin")
+								.env("NETCONFPATH", "/opt/cni/config")
+								.env("CNI_IFNAME", "eth0")
+								.env("CAP_ARGS", cni_params_json)
+								.output()
+								.await
+							{
+								Ok(cmd_out) => {
+									if !cmd_out.status.success() {
+										tracing::error!(
+											stdout=%std::str::from_utf8(&cmd_out.stdout)?,
+											stderr=%std::str::from_utf8(&cmd_out.stderr)?,
+											"failed `cnitool del` command",
+										);
+									}
+								}
+								Err(err) => tracing::error!(?err, "failed to run `cnitool` command"),
+							}
+						}
+						Err(err) => tracing::error!(?err, "failed to read `cni-cap-args.json`"),
+					}
+
+					match Command::new("ip")
+						.arg("netns")
 						.arg("del")
-						.arg(NETWORK_NAME)
-						.arg(netns_path)
-						.env("CNI_PATH", "/opt/cni/bin")
-						.env("NETCONFPATH", "/opt/cni/config")
-						.env("CNI_IFNAME", "eth0")
-						.env("CAP_ARGS", cni_params_json)
+						.arg(self.actor_id.to_string())
 						.output()
 						.await
 					{
@@ -701,34 +725,15 @@ impl Actor {
 								tracing::error!(
 									stdout=%std::str::from_utf8(&cmd_out.stdout)?,
 									stderr=%std::str::from_utf8(&cmd_out.stderr)?,
-									"failed `cnitool del` command",
+									"failed `ip netns` command",
 								);
 							}
 						}
-						Err(err) => tracing::error!(?err, "failed to run `cnitool` command"),
-					}
-				}
-				Err(err) => tracing::error!(?err, "failed to read `cni-cap-args.json`"),
-			}
-
-			match Command::new("ip")
-				.arg("netns")
-				.arg("del")
-				.arg(self.actor_id.to_string())
-				.output()
-				.await
-			{
-				Ok(cmd_out) => {
-					if !cmd_out.status.success() {
-						tracing::error!(
-							stdout=%std::str::from_utf8(&cmd_out.stdout)?,
-							stderr=%std::str::from_utf8(&cmd_out.stderr)?,
-							"failed `ip netns` command",
-						);
+						Err(err) => tracing::error!(?err, "failed to run `ip` command"),
 					}
 				}
-				Err(err) => tracing::error!(?err, "failed to run `ip` command"),
 			}
+			protocol::ImageKind::JavaScript => {}
 		}
 
 		Ok(())

packages/services/build/src/utils.rs

@@ -9,13 +9,17 @@ pub fn file_name(kind: BuildKind, compression: BuildCompression) -> String {
 	let file_name = match kind {
 		BuildKind::DockerImage => "image",
 		BuildKind::OciBundle => "oci-bundle",
+		BuildKind::JavaScript => "index",
+	};
+	let file_ext = match kind {
+		BuildKind::DockerImage | BuildKind::OciBundle => "tar",
 		BuildKind::JavaScript => "js",
 	};
-	let file_ext = match compression {
-		BuildCompression::None => "tar",
-		BuildCompression::Lz4 => "tar.lz4",
+	let file_ext_compression = match compression {
+		BuildCompression::None => "",
+		BuildCompression::Lz4 => ".lz4",
 	};
-	format!("{file_name}.{file_ext}")
+	format!("{file_name}.{file_ext}{file_ext_compression}")
 }
 
 pub fn build_hash(build_id: Uuid) -> u64 {

packages/services/build/util/src/lib.rs

@@ -11,13 +11,17 @@ pub fn file_name(
 	let file_name = match kind {
 		backend::build::BuildKind::DockerImage => "image",
 		backend::build::BuildKind::OciBundle => "oci-bundle",
+		backend::build::BuildKind::JavaScript => "index",
+	};
+	let file_ext = match kind {
+		backend::build::BuildKind::DockerImage | backend::build::BuildKind::OciBundle => "tar",
 		backend::build::BuildKind::JavaScript => "js",
 	};
-	let file_ext = match compression {
-		backend::build::BuildCompression::None => "tar",
-		backend::build::BuildCompression::Lz4 => "tar.lz4",
+	let file_ext_compression = match compression {
+		backend::build::BuildCompression::None => "",
+		backend::build::BuildCompression::Lz4 => ".lz4",
 	};
-	format!("{file_name}.{file_ext}")
+	format!("{file_name}.{file_ext}{file_ext_compression}")
 }
 
 pub fn build_hash(build_id: Uuid) -> u64 {

packages/services/ds/tests/lobby_connectivity.rs

@@ -10,7 +10,7 @@ use common::*;
 
 #[workflow_test]
 async fn server_connectivity_http_normal(ctx: TestCtx) {
-	if !util::feature::job_run() {
+	if ctx.config().server().unwrap().rivet.job_run.is_none() {
 		return;
 	}
 
@@ -38,7 +38,7 @@ async fn server_connectivity_http_normal(ctx: TestCtx) {
 
 #[workflow_test]
 async fn server_connectivity_http_host(ctx: TestCtx) {
-	if !util::feature::job_run() {
+	if ctx.config().server().unwrap().rivet.job_run.is_none() {
 		return;
 	}
 
@@ -87,7 +87,7 @@ async fn server_connectivity_http_host(ctx: TestCtx) {
 
 #[workflow_test]
 async fn server_connectivity_tcp(ctx: TestCtx) {
-	if !util::feature::job_run() {
+	if ctx.config().server().unwrap().rivet.job_run.is_none() {
 		return;
 	}
 
@@ -119,7 +119,7 @@ async fn server_connectivity_tcp(ctx: TestCtx) {
 
 #[workflow_test]
 async fn server_connectivity_tcp_host(ctx: TestCtx) {
-	if !util::feature::job_run() {
+	if ctx.config().server().unwrap().rivet.job_run.is_none() {
 		return;
 	}
 
@@ -177,7 +177,7 @@ async fn server_connectivity_tcp_host(ctx: TestCtx) {
 
 #[workflow_test]
 async fn server_connectivity_udp(ctx: TestCtx) {
-	if !util::feature::job_run() {
+	if ctx.config().server().unwrap().rivet.job_run.is_none() {
 		return;
 	}
 
@@ -206,7 +206,7 @@ async fn server_connectivity_udp(ctx: TestCtx) {
 
 #[workflow_test]
 async fn server_connectivity_udp_host(ctx: TestCtx) {
-	if !util::feature::job_run() {
+	if ctx.config().server().unwrap().rivet.job_run.is_none() {
 		return;
 	}
 

packages/services/ds/tests/print_test_data.rs

@@ -198,7 +198,7 @@ async fn print_test_data(ctx: TestCtx) {
 		server_id,
 		"testing2",
 		faker_region.region_id.unwrap(),
-		ctx.config().server()?.rivet.dns()?.domain_job,
+		ctx.config().server().unwrap().rivet.dns().unwrap().domain_job.as_ref().unwrap(),
 	);
 
 	// Async sleep for 5 seconds