feat(pegboard): add local image cache and overlay fs mounts

Author: MasterPtato

examples/system-test-actor/rivet.jsonc

@@ -1,10 +1,13 @@
 {
 	"builds": {
-		"ws-isolate": {
-			"script": "src/isolate/main.ts"
-		},
+		// "ws-isolate": {
+		// 	"script": "src/isolate/main.ts"
+		// },
 		"ws-container": {
-			"dockerfile": "Dockerfile"
+			"dockerfile": "Dockerfile",
+			"unstable": {
+				"compression": "none"
+			}
 		}
 	}
 }

packages/core/services/cluster/src/workflows/server/install/install_scripts/files/pegboard_configure.sh

@@ -353,6 +353,8 @@ ExecStart=/usr/local/bin/rivet-client -c /etc/rivet-client/config.json
 Restart=always
 RestartSec=2
 
+# High scheduling priority
+Nice=-15
 # Real time service
 CPUSchedulingPolicy=fifo
 # High CPU priority

packages/core/services/cluster/src/workflows/server/install/install_scripts/files/rivet_guard_configure.sh

@@ -26,6 +26,8 @@ ExecStart=/usr/local/bin/rivet-guard
 Restart=always
 RestartSec=2
 
+# High scheduling priority
+Nice=-15
 # Real time service
 CPUSchedulingPolicy=fifo
 # High CPU priority

packages/core/services/cluster/src/workflows/server/install/install_scripts/files/rivet_worker_configure.sh

@@ -28,6 +28,8 @@ ExecStart=/usr/local/bin/rivet-edge-server start --skip-provision
 Restart=always
 RestartSec=2
 
+# High scheduling priority
+Nice=-15
 # Real time service
 CPUSchedulingPolicy=fifo
 # High CPU priority

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

@@ -32,7 +32,7 @@ pub fn run(
 		.context("empty `actor_path`")?
 		.to_string_lossy()
 		.to_string();
-	let fs_path = actor_path.join("fs");
+	let fs_path = actor_path.join("fs").join("upper");
 	let oci_bundle_config_json = fs_path.join("config.json");
 
 	// Validate OCI bundle

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

@@ -2,9 +2,12 @@ use tiny_http::{Response, Server, StatusCode};
 
 // TODO: This can't pick up SIGTERM
 fn main() {
-	let port = std::env::var("PORT")
-		.ok()
-		.unwrap_or_else(|| "8080".to_string());
+	println!("Env:");
+	for (key, value) in std::env::vars() {
+		println!("  {}: {}", key, value);
+	}
+
+	let port = std::env::var("PORT_MAIN").expect("no PORT_MAIN");
 	let addr = format!("0.0.0.0:{port}");
 	let server = Server::http(&addr).unwrap();
 	println!("Listening on {addr}");

packages/edge/infra/client/isolate-v8-runner/src/isolate.rs

@@ -153,7 +153,7 @@ pub async fn run_inner(
 	tracing::info!(?actor_id, ?generation, "isolate kv initialized");
 
 	// Should match the path from `Actor::make_fs` in manager/src/actor/setup.rs
-	let index = actor_path.join("fs").join("index.js");
+	let index = actor_path.join("fs").join("upper").join("index.js");
 
 	// Load index.js
 	let index_script_content = match fs::read_to_string(&index).await {
@@ -560,7 +560,8 @@ mod tests {
 
 		let fs_path = actors_path
 			.join(format!("{actor_id}-{generation}"))
-			.join("fs");
+			.join("fs")
+			.join("upper");
 		std::fs::create_dir_all(&fs_path)?;
 
 		std::fs::copy(

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

@@ -30,6 +30,7 @@ rand = "0.8"
 rand_chacha = "0.3.1"
 reqwest = { version = "0.12", default-features = false, features = ["stream", "rustls-tls", "json"] }
 rivet-logs.workspace = true
+scc = "2.3.4"
 serde = { version = "1.0.195", features = ["derive"] }
 serde_json = "1.0.111"
 serde_yaml = "0.9.34"

packages/edge/infra/client/manager/src/actor/mod.rs

@@ -1,14 +1,15 @@
 use std::{
 	result::Result::{Err, Ok},
 	sync::Arc,
-	time::Duration,
+	time::{Duration, Instant},
 };
 
 use anyhow::*;
 use indoc::indoc;
 use nix::sys::signal::Signal;
 use pegboard::protocol;
 use pegboard_config::runner_protocol;
+use sqlx::Acquire;
 use tokio::{fs, sync::Mutex};
 use uuid::Uuid;
 
@@ -75,16 +76,18 @@ impl Actor {
 					actor_id,
 					generation,
 					config,
-					start_ts
+					start_ts,
+					image_id
 				)
-				VALUES (?1, ?2, ?3, ?4)
+				VALUES (?1, ?2, ?3, ?4, ?5)
 				ON CONFLICT (actor_id, generation) DO NOTHING
 				",
 			))
 			.bind(self.actor_id)
 			.bind(self.generation as i64)
 			.bind(&config_json)
 			.bind(utils::now())
+			.bind(self.config.image.id)
 			.execute(&mut *ctx.sql().await?)
 			.await
 		})
@@ -128,7 +131,7 @@ impl Actor {
 		self: &Arc<Self>,
 		ctx: &Arc<Ctx>,
 	) -> Result<protocol::HashableMap<String, protocol::ProxiedPort>> {
-		let setup_timer = std::time::Instant::now();
+		let setup_timer = Instant::now();
 		tracing::info!(actor_id=?self.actor_id, generation=?self.generation, "setting up actor");
 
 		let actor_path = ctx.actor_path(self.actor_id, self.generation);
@@ -146,19 +149,13 @@ impl Actor {
 				protocol::ImageKind::DockerImage | protocol::ImageKind::OciBundle
 			) && matches!(self.config.network_mode, protocol::NetworkMode::Bridge);
 
-		// Parallelize two independent jobs:
-		//
-		// - `download_image` takes a long time to download. `download_image` is dependent on
-		//   `make_fs`
-		// - `setup_cni_network` takes a long time. `setup_cni_network` is dependent on
-		//   `bind_ports`.
 		tracing::info!(actor_id=?self.actor_id, generation=?self.generation, "starting parallel setup tasks");
-		let parallel_timer = std::time::Instant::now();
+		let parallel_timer = Instant::now();
 
 		let (_, ports) = tokio::try_join!(
 			async {
-				self.make_fs(&ctx).await?;
 				self.download_image(&ctx).await?;
+				self.make_fs(&ctx).await?;
 				Result::<(), anyhow::Error>::Ok(())
 			},
 			async {
@@ -205,7 +202,12 @@ impl Actor {
 		let mut runner_env = vec![
 			(
 				"ROOT_USER_ENABLED",
-				if self.config.root_user_enabled { "1" } else { "0" }.to_string(),
+				if self.config.root_user_enabled {
+					"1"
+				} else {
+					"0"
+				}
+				.to_string(),
 			),
 			("ACTOR_ID", self.actor_id.to_string()),
 		];
@@ -443,7 +445,10 @@ impl Actor {
 		// Update stop_ts
 		if matches!(signal, Signal::SIGTERM | Signal::SIGKILL) || !has_runner {
 			let stop_ts_set = utils::sql::query(|| async {
-				sqlx::query_as::<_, (bool,)>(indoc!(
+				let mut conn = ctx.sql().await?;
+				let mut tx = conn.begin().await?;
+
+				let res = sqlx::query_as::<_, (bool,)>(indoc!(
 					"
 					UPDATE actors
 					SET stop_ts = ?3
@@ -457,11 +462,27 @@ impl Actor {
 				.bind(self.actor_id)
 				.bind(self.generation as i64)
 				.bind(utils::now())
-				.fetch_optional(&mut *ctx.sql().await?)
-				.await
+				.fetch_optional(&mut *tx)
+				.await?;
+
+				// Update LRU cache
+				sqlx::query(indoc!(
+					"
+					UPDATE images_cache
+					SET last_used_ts = ?2
+					WHERE image_id = ?1
+					",
+				))
+				.bind(self.config.image.id)
+				.bind(utils::now())
+				.execute(&mut *tx)
+				.await?;
+
+				tx.commit().await?;
+
+				Ok(res.is_some())
 			})
-			.await?
-			.is_some();
+			.await?;
 
 			// Emit event if not stopped before
 			if stop_ts_set {

packages/edge/infra/client/manager/src/actor/oci_config.rs

@@ -25,9 +25,7 @@ pub fn config(opts: ConfigOpts) -> Result<serde_json::Value> {
 	//
 	// Corresponds to cpu.weight in cgroups. Must be [1, 10_000]
 	//
-	// We divide by 8 in order to make sure the CPU shares are within bounds. `cpu` is measured in
-	// millishares, so 1_000 = 1 core. For a range of 32d1 (32_000) to 1d16 (62), we divide by 8
-	// to make the range 3_200 to 6.
+	// We divide by 10 in order to make sure the CPU shares are within bounds.
 	let mut cpu_shares = opts.cpu / 10;
 	if cpu_shares > 10_000 {
 		cpu_shares = 10_000;