wip: fix whatever is going on with envoys

Author: NathanFlurry

engine/packages/pegboard-envoy/src/conn.rs

@@ -132,10 +132,23 @@ pub async fn handle_init(
 			bypass_cache: false,
 		}),
 		// TODO: Move to op
-		udb.run(|tx| {
+		udb.run(|root_tx| {
 			let init = init.clone();
 			async move {
-				let tx = tx.with_subspace(pegboard::keys::subspace());
+				// Write protocol version to runner config so the actor create op
+				// knows to use the v2 workflow for this pool.
+				{
+					let ns_tx = root_tx.with_subspace(namespace::keys::subspace());
+					ns_tx.write(
+						&pegboard::keys::runner_config::ProtocolVersionKey::new(
+							namespace_id,
+							pool_name.clone(),
+						),
+						protocol_version,
+					)?;
+				}
+
+				let tx = root_tx.with_subspace(pegboard::keys::subspace());
 
 				let create_ts_key =
 					pegboard::keys::envoy::CreateTsKey::new(namespace_id, envoy_key.clone());

pnpm-lock.yaml

@@ -2,7 +2,7 @@
   "asyncapi": "3.0.0",
   "info": {
     "title": "RivetKit WebSocket Protocol",
-    "version": "2.2.0",
+    "version": "2.2.1",
     "description": "WebSocket protocol for bidirectional communication between RivetKit clients and actors"
   },
   "channels": {

rivetkit-asyncapi/asyncapi.json

@@ -563,7 +563,7 @@ export function buildManagerRouter(
 				try {
 					// Forward the request using the inline client driver
 					const response = await managerDriver.sendRequest(
-						actorId,
+						{ directId: actorId },
 						new Request(`http://actor/${pathWithQuery}`, {
 							method: c.req.method,
 							headers: c.req.raw.headers,
@@ -616,7 +616,7 @@ export function buildManagerRouter(
 				try {
 					// Send a special request to the actor to force disconnect the connection
 					const response = await managerDriver.sendRequest(
-						actorId,
+						{ directId: actorId },
 						new Request(
 							`http://actor/.test/force-disconnect?conn=${connId}`,
 							{

rivetkit-typescript/packages/rivetkit/src/manager/router.ts

@@ -52,6 +52,30 @@ const driverTestConfig = {
 					);
 				}
 
+				// Create runner config for the namespace so the engine
+				// knows how to schedule actors for this pool.
+				const rcResponse = await fetch(
+					`${namespaceEndpoint}/runner-configs/${runnerName}?namespace=${namespace}`,
+					{
+						method: "PUT",
+						headers: {
+							"Content-Type": "application/json",
+							Authorization: "Bearer dev",
+						},
+						body: JSON.stringify({
+							datacenters: {
+								default: { normal: {} },
+							},
+						}),
+					},
+				);
+				if (!rcResponse.ok) {
+					const errorBody = await rcResponse.text().catch(() => "");
+					throw new Error(
+						`Create runner config failed: ${rcResponse.status} ${rcResponse.statusText} ${errorBody}`,
+					);
+				}
+
 				// Create driver config.
 				const driverConfig = createEngineDriver();
 
@@ -83,43 +107,43 @@ const driverTestConfig = {
 					inlineClient,
 				);
 
-				// Wait for runner registration so tests do not race actor creation
-				// against asynchronous runner connect.
-				const runnersUrl = new URL(
-					`${endpoint.replace(/\/$/, "")}/runners`,
+				// Wait for envoy registration so tests do not race actor creation
+				// against asynchronous envoy connect.
+				const envoysUrl = new URL(
+					`${endpoint.replace(/\/$/, "")}/envoys`,
 				);
-				runnersUrl.searchParams.set("namespace", namespace);
-				runnersUrl.searchParams.set("name", runnerName);
+				envoysUrl.searchParams.set("namespace", namespace);
+				envoysUrl.searchParams.set("name", runnerName);
 				let probeError: unknown;
 				for (let attempt = 0; attempt < 120; attempt++) {
 					try {
-						const runnerResponse = await fetch(runnersUrl, {
+						const envoyResponse = await fetch(envoysUrl, {
 							method: "GET",
 							headers: {
 								Authorization: `Bearer ${token}`,
 							},
 						});
-						if (!runnerResponse.ok) {
-							const errorBody = await runnerResponse
+						if (!envoyResponse.ok) {
+							const errorBody = await envoyResponse
 								.text()
 								.catch(() => "");
 							probeError = new Error(
-								`List runners failed: ${runnerResponse.status} ${runnerResponse.statusText} ${errorBody}`,
+								`List envoys failed: ${envoyResponse.status} ${envoyResponse.statusText} ${errorBody}`,
 							);
 						} else {
 							const responseJson =
-								(await runnerResponse.json()) as {
-									runners?: Array<{ name?: string }>;
+								(await envoyResponse.json()) as {
+									envoys?: Array<{ pool_name?: string }>;
 								};
-							const hasRunner = !!responseJson.runners?.some(
-								(runner) => runner.name === runnerName,
+							const hasEnvoy = !!responseJson.envoys?.some(
+								(envoy) => envoy.pool_name === runnerName,
 							);
-							if (hasRunner) {
+							if (hasEnvoy) {
 								probeError = undefined;
 								break;
 							}
 							probeError = new Error(
-								`Runner ${runnerName} not registered yet`,
+								`Envoy ${runnerName} not registered yet`,
 							);
 						}
 					} catch (err) {