feat(pb): get multi-actors working e2e, add resources for builds

Author: MasterPtato

Cargo.lock

@@ -23,4 +23,4 @@ RUN chown -R rivet:rivet /app/dist
 USER rivet
 
 # Start the server
-CMD ["node", "dist/src/container/main.js"]
+CMD ["node", "dist/src/main.js"]

docs-internal/infrastructure/pegboard/ISOLATE_RUNNER.md

@@ -1,9 +1,6 @@
 {
 	"builds": {
-		// "ws-isolate": {
-		// 	"script": "src/isolate/main.ts"
-		// },
-		"ws-container": {
+		"ws": {
 			"dockerfile": "Dockerfile"
 		}
 	}

examples/system-test-actor/Dockerfile

@@ -3,7 +3,7 @@ import type { UpgradeWebSocket } from "hono/ws";
 
 type GetUpgradeWebSocketFn = (app: Hono) => UpgradeWebSocket;
 
-export function createAndStartServer(
+export function createAndStartHttpServer(
 	getUpgradeWebSocket: GetUpgradeWebSocketFn,
 ): { app: Hono; port: number } {
 	// Setup auto-exit timer
@@ -12,8 +12,7 @@ export function createAndStartServer(
 			"Actor should've been destroyed by now. Automatically exiting.",
 		);
 
-		if (typeof Deno !== "undefined") Deno.exit(1);
-		else process.exit(1);
+		process.exit(1);
 	}, 60 * 1000);
 
 	let tickIndex = 0;
@@ -23,10 +22,7 @@ export function createAndStartServer(
 	}, 1000);
 
 	// Get port from environment
-	const portEnv =
-		typeof Deno !== "undefined"
-			? Deno.env.get("PORT_HTTP")
-			: process.env.PORT_HTTP;
+	const portEnv = process.env.PORT_HTTP;
 	if (!portEnv) {
 		throw new Error("missing PORT_HTTP");
 	}
@@ -41,8 +37,7 @@ export function createAndStartServer(
 		const query = c.req.query("code");
 		const exitCode = query ? Number(query) : 0;
 
-		if (typeof Deno != "undefined") Deno.exit(exitCode);
-		else process.exit(exitCode);
+		process.exit(exitCode);
 
 		return c.text("unreachable");
 	});

examples/system-test-actor/rivet.jsonc

@@ -0,0 +1,22 @@
+import { serve } from "@hono/node-server";
+import { createNodeWebSocket } from "@hono/node-ws";
+import { createAndStartHttpServer } from "./httpServer.js";
+import { createAndStartUdpServer } from "./udpServer.js";
+import { connectToManager } from "./managerClient.js";
+
+let injectWebSocket: any;
+const { app, port } = createAndStartHttpServer((app) => {
+	// Get Node.js WebSocket handler
+	const result = createNodeWebSocket({ app });
+	injectWebSocket = result.injectWebSocket;
+	return result.upgradeWebSocket;
+});
+
+const server = serve({ fetch: app.fetch, port });
+injectWebSocket(server);
+
+createAndStartUdpServer();
+
+if (process.env.MULTI) {
+	connectToManager();
+}

examples/system-test-actor/src/container/main.ts

@@ -0,0 +1,85 @@
+import WebSocket from "ws";
+
+export function connectToManager() {
+	let managerIp = process.env.RIVET_MANAGER_IP;
+	let managerPort = process.env.RIVET_MANAGER_PORT;
+	let pingInterval: NodeJS.Timeout;
+
+	if (!managerIp || !managerPort) {
+		console.error("Missing RIVET_MANAGER_IP or RIVET_MANAGER_PORT environment variables");
+		return;
+	}
+
+	let wsUrl = `ws://${managerIp}:${managerPort}`;
+	console.log(`Connecting to manager WebSocket at ${wsUrl}`);
+
+	let ws = new WebSocket(wsUrl);
+
+	ws.on("open", () => {
+		console.log("Connected to manager WebSocket");
+
+		let message = {
+			init: {
+				runner_id: process.env.RIVET_RUNNER_ID
+			}
+		};
+		let buffer = Buffer.from(JSON.stringify(message));
+		ws.send(buffer);
+
+		// Start ping loop to keep connection alive
+		pingInterval = setInterval(() => {
+			if (ws.readyState === WebSocket.OPEN) {
+				ws.ping();
+			}
+		}, 2000);
+	});
+
+	ws.on("message", (data) => {
+		let json = data.toString();
+
+		console.log("Received message from manager:", json);
+
+		let packet = JSON.parse(json);
+
+		if (packet.start_actor) {
+			let message = {
+				actor_state_update: {
+					actor_id: packet.start_actor.actor_id,
+					generation: packet.start_actor.generation,
+					state: {
+						running: null,
+					},
+				}
+			};
+			let buffer = Buffer.from(JSON.stringify(message));
+			ws.send(buffer);
+		} else if (packet.signal_actor) {
+			let message = {
+				actor_state_update: {
+					actor_id: packet.start_actor.actor_id,
+					generation: packet.start_actor.generation,
+					state: {
+						exited: {
+							exit_code: 0,
+						}
+					},
+				}
+			};
+			let buffer = Buffer.from(JSON.stringify(message));
+			ws.send(buffer);
+		}
+	});
+
+	ws.on("error", (error) => {
+		console.error("WebSocket error:", error);
+	});
+
+	ws.on("close", code => {
+		console.log("WebSocket connection closed, attempting to reconnect...", code);
+
+		// Clear ping interval when connection closes
+		if (pingInterval) clearInterval(pingInterval);
+
+		setTimeout(connectToManager, 5000);
+	});
+}

examples/system-test-actor/src/shared/server.ts renamed to examples/system-test-actor/src/httpServer.ts

@@ -0,0 +1,36 @@
+import dgram from 'dgram';
+
+export function createAndStartUdpServer() {
+	// Get port from environment
+	const portEnv = process.env.PORT_UDP;
+
+	if (portEnv) {
+		// Create a UDP socket
+		const udpServer = dgram.createSocket('udp4');
+
+		// Listen for incoming messages
+		udpServer.on('message', (msg, rinfo) => {
+			console.log(`UDP server received: ${msg} from ${rinfo.address}:${rinfo.port}`);
+
+			// Echo the message back to the sender
+			udpServer.send(msg, rinfo.port, rinfo.address, (err) => {
+				if (err) console.error('Failed to send UDP response:', err);
+			});
+		});
+
+		// Handle errors
+		udpServer.on('error', (err) => {
+			console.error('UDP server error:', err);
+			udpServer.close();
+		});
+
+
+		const port2 = Number.parseInt(portEnv);
+
+		udpServer.bind(port2, () => {
+			console.log(`UDP echo server running on port ${port2}`);
+		});
+	} else {
+		console.warn("missing PORT_UDP");
+	}
+}