chore: fix driver test suite

Author: NathanFlurry

CLAUDE.md

@@ -284,7 +284,10 @@ let error_with_meta = ApiRateLimited { limit: 100, reset_at: 1234567890 }.build(
 - Connection pooling through `packages/common/pools/`
 
 **Performance**
-- ALWAYS prefer a dedicated concurrency container like `scc::HashMap<_, _>` with its async api or `moka::Cache` over `Arc<Mutex<HashMap<_, _>>>`. `Arc<Mutex<_>>` is very slow for containers.
+- Never use `Mutex<HashMap<...>>` or `RwLock<HashMap<...>>`.
+- Use `scc::HashMap` (preferred), `moka::Cache` (for TTL/bounded), or `DashMap` for concurrent maps.
+- Use `scc::HashSet` instead of `Mutex<HashSet<...>>` for concurrent sets.
+- `scc` async methods do not hold locks across `.await` points. Use `entry_async` for atomic read-then-write.
 
 ### Code Style
 - Hard tabs for Rust formatting (see `rustfmt.toml`)

Cargo.lock

@@ -46,6 +46,7 @@ members = [
   "engine/packages/universaldb",
   "engine/packages/universalpubsub",
   "engine/packages/util",
+  "engine/packages/util-serde",
   "engine/packages/util-id",
   "engine/packages/workflow-worker",
   "engine/sdks/rust/api-full",
@@ -488,6 +489,9 @@ members = [
     package = "rivet-util"
     path = "engine/packages/util"
 
+    [workspace.dependencies.rivet-util-serde]
+    path = "engine/packages/util-serde"
+
     [workspace.dependencies.rivet-util-id]
     path = "engine/packages/util-id"
 

Cargo.toml

@@ -33,10 +33,6 @@ When changing a versioned VBARE schema, follow the existing migration pattern.
 - When adding fields to epoxy workflow state structs, mark them `#[serde(default)]` so Gasoline can replay older serialized state.
 - Epoxy integration tests that spin up `tests/common::TestCtx` must call `shutdown()` before returning.
 
-## Concurrent containers
-
-Never use `Mutex<HashMap<...>>` or `RwLock<HashMap<...>>`. Use `scc::HashMap` (preferred), `moka::Cache` (for TTL/bounded), or `DashMap`. Same for sets: use `scc::HashSet` instead of `Mutex<HashSet<...>>`. Note that `scc` async methods do not hold locks across `.await` points. Use `entry_async` for atomic read-then-write.
-
 ## Test snapshots
 
 Use `test-snapshot-gen` to generate and load RocksDB snapshots of the full UDB KV store for migration and integration tests. Scenarios produce per-replica RocksDB checkpoints stored under `engine/packages/test-snapshot-gen/snapshots/` (git LFS tracked). In tests, use `test_snapshot::SnapshotTestCtx::from_snapshot("scenario-name")` to boot a cluster from snapshot data. See `docs-internal/engine/TEST_SNAPSHOTS.md` for the full guide.

engine/CLAUDE.md

@@ -207,7 +207,27 @@ async fn resolve_query_target_dc_label(
 }
 
 fn serialize_actor_key(key: &[String]) -> Result<String> {
-	serde_json::to_string(key).context("failed to serialize actor key")
+	const EMPTY_KEY: &str = "/";
+	const KEY_SEPARATOR: char = '/';
+
+	if key.is_empty() {
+		return Ok(EMPTY_KEY.to_string());
+	}
+
+	let mut escaped_parts = Vec::with_capacity(key.len());
+	for part in key {
+		if part.is_empty() {
+			escaped_parts.push(String::from("\\0"));
+			continue;
+		}
+
+		let escaped = part
+			.replace('\\', "\\\\")
+			.replace(KEY_SEPARATOR, "\\/");
+		escaped_parts.push(escaped);
+	}
+
+	Ok(escaped_parts.join(EMPTY_KEY))
 }
 
 fn is_duplicate_key_error(err: &anyhow::Error) -> bool {

engine/packages/guard/src/routing/pegboard_gateway/resolve_actor_query.rs

@@ -47,7 +47,10 @@ pub async fn pegboard_actor_create(ctx: &OperationCtx, input: &Input) -> Result<
 		ctx.subscribe::<crate::workflows::actor2::DestroyStarted>(("actor_id", input.actor_id)),
 		ctx.op(crate::ops::runner_config::get::Input {
 			runners: vec![(input.namespace_id, input.runner_name_selector.clone())],
-			bypass_cache: false,
+			// Actor creation needs the latest protocol version immediately after
+			// serverless metadata refresh so fresh pools can start on actor2
+			// without waiting for the runner-config cache to expire.
+			bypass_cache: true,
 		}),
 	)?;
 

engine/packages/pegboard/src/ops/actor/create.rs

@@ -3,6 +3,20 @@
 
 /* auto-generated by NAPI-RS */
 
+export interface JsBindParam {
+  kind: string
+  intValue?: number
+  floatValue?: number
+  textValue?: string
+  blobValue?: Buffer
+}
+export interface ExecuteResult {
+  changes: number
+}
+export interface QueryResult {
+  columns: Array<string>
+  rows: Array<Array<any>>
+}
 /** Open a native SQLite database backed by the envoy's KV channel. */
 export declare function openDatabaseFromEnvoy(jsHandle: JsEnvoyHandle, actorId: string): Promise<JsNativeDatabase>
 /** Configuration for starting the native envoy client. */
@@ -44,7 +58,12 @@ export declare function startEnvoySyncJs(config: JsEnvoyConfig, eventCallback: (
 /** Start the native envoy client asynchronously. */
 export declare function startEnvoyJs(config: JsEnvoyConfig, eventCallback: (event: any) => void): JsEnvoyHandle
 /** Native SQLite database handle exposed to JavaScript. */
-export declare class JsNativeDatabase { }
+export declare class JsNativeDatabase {
+  run(sql: string, params?: Array<JsBindParam> | undefined | null): Promise<ExecuteResult>
+  query(sql: string, params?: Array<JsBindParam> | undefined | null): Promise<QueryResult>
+  exec(sql: string): Promise<QueryResult>
+  close(): Promise<void>
+}
 /** Native envoy handle exposed to JavaScript via N-API. */
 export declare class JsEnvoyHandle {
   started(): Promise<void>
@@ -64,10 +83,10 @@ export declare class JsEnvoyHandle {
   kvDrop(actorId: string): Promise<void>
   restoreHibernatingRequests(actorId: string, requests: Array<HibernatingRequestEntry>): void
   sendHibernatableWebSocketMessageAck(gatewayId: Buffer, requestId: Buffer, clientMessageIndex: number): void
-  startServerless(payload: Buffer): Promise<void>
-  /** Send a message on an open WebSocket connection. */
-  sendWsMessage(gatewayId: Buffer, requestId: Buffer, data: Buffer, binary: boolean): void
+  /** Send a message on an open WebSocket connection identified by messageIdHex. */
+  sendWsMessage(gatewayId: Buffer, requestId: Buffer, data: Buffer, binary: boolean): Promise<void>
   /** Close an open WebSocket connection. */
-  closeWebsocket(gatewayId: Buffer, requestId: Buffer, code?: number | undefined | null, reason?: string | undefined | null): void
+  closeWebsocket(gatewayId: Buffer, requestId: Buffer, code?: number | undefined | null, reason?: string | undefined | null): Promise<void>
+  startServerless(payload: Buffer): Promise<void>
   respondCallback(responseId: string, data: any): Promise<void>
 }