fix(pegboard): convert ws manager socket to unix socket

Author: MasterPtato

docker/dev-full/prometheus/prometheus.yml

@@ -14,4 +14,4 @@ scrape_configs:
   - job_name: rivet-client
     static_configs:
       - targets:
-        - rivet-client:6090
+        - rivet-client:8091

docker/dev-full/vector-client/vector.yaml

@@ -8,7 +8,7 @@ sources:
   prometheus_pegboard:
     type: prometheus_scrape
     endpoints:
-      - http://rivet-client:6090
+      - http://rivet-client:8091
     scrape_interval_secs: 15
 
   pegboard_manager:

docker/monolith/build-scripts/setup_s6.ts

@@ -103,7 +103,7 @@ const services: Service[] = [
 		rootUser: true,
 		ports: {
 			runner: 6080,
-			metrics: 6090,
+			metrics: 8091,
 		},
 	},
 

docker/monolith/vector-client/vector.yaml

@@ -12,7 +12,7 @@ sources:
   prometheus_pegboard:
     type: prometheus_scrape
     endpoints:
-      - http://rivet-client:6090
+      - http://rivet-client:8091
     scrape_interval_secs: 15
 
   pegboard_manager:

packages/common/chirp-workflow/core/src/worker.rs

@@ -83,6 +83,14 @@ impl Worker {
 		loop {
 			tokio::select! {
 				_ = tick_interval.tick() => {},
+				res = wake_sub.next() => {
+					if res.is_none() {
+						return Err(WorkflowError::SubscriptionUnsubscribed.into());
+					}
+
+					tick_interval.reset();
+				},
+
 				res = &mut gc_handle => {
 					tracing::error!(?res, "metrics task unexpectedly stopped");
 					break;
@@ -91,13 +99,6 @@ impl Worker {
 					tracing::error!(?res, "metrics task unexpectedly stopped");
 					break;
 				},
-				res = wake_sub.next() => {
-					if res.is_none() {
-						return Err(WorkflowError::SubscriptionUnsubscribed.into());
-					}
-
-					tick_interval.reset();
-				},
 				_ = ctrl_c() => break,
 				_ = sigterm.recv() => break,
 			}

packages/common/metrics/src/server.rs

@@ -1,11 +1,12 @@
+use std::net::SocketAddr;
+
 use global_error::prelude::*;
 use hyper::{
 	header::CONTENT_TYPE,
 	service::{make_service_fn, service_fn},
 	Body, Request, Response, Server,
 };
 use prometheus::{Encoder, TextEncoder};
-use std::net::SocketAddr;
 
 // TODO: Record extra labels
 

packages/core/services/cluster/src/workflows/server/install/install_scripts/mod.rs

@@ -175,7 +175,7 @@ pub async fn gen_initialize(
 				"pegboard".into(),
 				components::vector::PrometheusTarget {
 					// Should match port from pb manager config
-					endpoint: "http://127.0.0.1:6090".into(),
+					endpoint: "http://127.0.0.1:8091".into(),
 					scrape_interval: 15,
 				},
 			);
@@ -190,7 +190,7 @@ pub async fn gen_initialize(
 				"pegboard".into(),
 				components::vector::PrometheusTarget {
 					// Should match port from pb manager config
-					endpoint: "http://127.0.0.1:6090".into(),
+					endpoint: "http://127.0.0.1:8091".into(),
 					scrape_interval: 15,
 				},
 			);

packages/edge/infra/client/config/Cargo.toml

@@ -6,10 +6,13 @@ license.workspace = true
 edition.workspace = true
 
 [dependencies]
+anyhow = "1.0"
 ipnet = { version = "2.10.1", features = ["serde"] }
+pegboard.workspace = true
+rivet-util.workspace = true
 schemars = { version = "0.8.21", features = ["url", "uuid1"] }
 serde = { version = "1.0.195", features = ["derive"] }
+serde_json = "1.0"
+tokio-util = { version = "0.7", features = ["codec"] }
 url = "2.2.2"
 uuid = { version = "1.6.1", features = ["v4"] }
-pegboard.workspace = true
-rivet-util.workspace = true

packages/edge/infra/client/config/src/manager.rs

@@ -85,13 +85,6 @@ pub struct Runner {
 	/// Whether or not to use a mount for actor file systems.
 	pub use_mounts: Option<bool>,
 
-	/// Address of the WebSocket server for runners. Should exist on a network interface that both the host
-	/// and containers can access.
-	pub ip: Option<IpAddr>,
-
-	/// WebSocket port for runners on this machine to connect to.
-	pub port: Option<u16>,
-
 	pub container_runner_binary_path: Option<PathBuf>,
 }
 
@@ -100,14 +93,6 @@ impl Runner {
 		self.use_mounts.unwrap_or(true)
 	}
 
-	pub fn ip(&self) -> IpAddr {
-		self.ip.unwrap_or(IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)))
-	}
-
-	pub fn port(&self) -> u16 {
-		self.port.unwrap_or(6080)
-	}
-
 	pub fn container_runner_binary_path(&self) -> PathBuf {
 		self.container_runner_binary_path
 			.clone()
@@ -247,7 +232,7 @@ pub struct Metrics {
 
 impl Metrics {
 	pub fn port(&self) -> u16 {
-		self.port.unwrap_or(6090)
+		self.port.unwrap_or(8091)
 	}
 }
 

packages/edge/infra/client/config/src/runner_protocol.rs

@@ -1,5 +1,9 @@
+use anyhow::*;
 use pegboard::protocol;
 use serde::{Deserialize, Serialize};
+use tokio_util::codec::LengthDelimitedCodec;
+use serde::{de::DeserializeOwned, Serialize};
+use bytes::Bytes;
 
 #[derive(Debug, Serialize, Deserialize)]
 #[serde(rename_all = "snake_case", deny_unknown_fields)]
@@ -38,3 +42,38 @@ pub enum ActorState {
 	Running,
 	Exited { exit_code: Option<i32> },
 }
+
+pub fn codec() -> LengthDelimitedCodec {
+	LengthDelimitedCodec::builder()
+		.length_field_type::<u32>()
+		.length_field_length(4)
+		// No offset
+		.length_field_offset(0)
+		// Header length is not included in the length calculation
+		.length_adjustment(4)
+		// Header is included in the returned bytes
+		.num_skip(0)
+		.new_codec()
+}
+
+pub fn encode_frame<T: Serialize>(payload: &T) -> Result<Vec<u8>> {
+	let mut buf = Vec::with_capacity(4);
+	buf.extend_from_slice(&[0u8; 4]); // header (currently unused)
+
+	let mut cursor = Cursor::new(&mut buf);
+	serde_json::to_writer(&mut cursor, payload)?;
+
+	Ok(buf)
+}
+
+fn decode_frame<T: DeserializeOwned>(frame: &Vec<u8>) -> Result<([u8; 4], T)> {
+	ensure!(frame.len() >= 4, "Frame too short");
+
+	// Extract the header (first 4 bytes)
+	let header = [frame[0], frame[1], frame[2], frame[3]];
+
+	// Deserialize the rest of the frame (payload after the header)
+	let payload = serde_json::from_slice(&frame[4..])?;
+
+	Ok((header, payload))
+}