fix(rivetkit): remove top-level promise errors

Author: NathanFlurry

rivetkit-typescript/packages/cloudflare-workers/src/actor-driver.ts

@@ -13,6 +13,7 @@ import type {
 	ManagerDriver,
 } from "rivetkit/driver-helpers";
 import { promiseWithResolvers } from "rivetkit/utils";
+import { logger } from "./log";
 import { kvDelete, kvGet, kvListPrefix, kvPut } from "./actor-kv";
 import { GLOBAL_KV_KEYS } from "./global-kv";
 import { getCloudflareAmbientEnv } from "./handler";
@@ -142,7 +143,7 @@ export class CloudflareActorsActorDriver implements ActorDriver {
 		// Create new actor state if it doesn't exist
 		if (!actorState) {
 			actorState = new ActorGlobalState();
-			actorState.actorPromise = promiseWithResolvers();
+			actorState.actorPromise = promiseWithResolvers((reason) => logger().warn({ msg: "unhandled actor promise rejection", reason }));
 			this.#globalState.setActorState(doState.ctx, actorState);
 		} else if (actorState.actorPromise) {
 			// Another request is already loading this actor, wait for it

rivetkit-typescript/packages/rivetkit/fixtures/driver-test-suite/sleep.ts

@@ -52,7 +52,7 @@ export const sleepWithLongRpc = actor({
 		},
 		longRunningRpc: async (c) => {
 			c.log.info("starting long running rpc");
-			c.vars.longRunningResolve = promiseWithResolvers();
+			c.vars.longRunningResolve = promiseWithResolvers((reason) => c.log.warn({ msg: "unhandled long running rpc rejection", reason }));
 			c.broadcast("waiting");
 			await c.vars.longRunningResolve.promise;
 			c.log.info("finished long running rpc");

rivetkit-typescript/packages/rivetkit/src/actor/instance/queue-manager.ts

@@ -6,6 +6,7 @@ import {
 	QUEUE_METADATA_VERSIONED,
 } from "@/schemas/actor-persist/versioned";
 import { promiseWithResolvers } from "@/utils";
+import { loggerWithoutContext } from "@/actor/log";
 import type { AnyDatabaseProvider } from "../database";
 import type { ActorDriver } from "../driver";
 import * as errors from "../errors";
@@ -260,7 +261,7 @@ export class QueueManager<S, CP, CS, V, I, DB extends AnyDatabaseProvider> {
 		}
 
 		const { promise, resolve, reject } =
-			promiseWithResolvers<QueueMessage[]>();
+			promiseWithResolvers<QueueMessage[]>((reason) => loggerWithoutContext().warn({ msg: "unhandled queue message waiter rejection", reason }));
 		const waiterId = crypto.randomUUID();
 		const waiter: QueueWaiter = {
 			id: waiterId,
@@ -318,7 +319,7 @@ export class QueueManager<S, CP, CS, V, I, DB extends AnyDatabaseProvider> {
 		timeout?: number,
 	): Promise<QueueCompletionResult> {
 		const { promise, resolve, reject } =
-			promiseWithResolvers<QueueCompletionResult>();
+			promiseWithResolvers<QueueCompletionResult>((reason) => loggerWithoutContext().warn({ msg: "unhandled queue completion waiter rejection", reason }));
 		const waiterId = crypto.randomUUID();
 
 		const waiter: QueueCompletionWaiter = {

rivetkit-typescript/packages/rivetkit/src/actor/instance/state-manager.ts

@@ -7,6 +7,7 @@ import {
 	CONN_VERSIONED,
 } from "@/schemas/actor-persist/versioned";
 import { promiseWithResolvers, SinglePromiseQueue } from "@/utils";
+import { loggerWithoutContext } from "@/actor/log";
 import { type AnyConn, CONN_STATE_MANAGER_SYMBOL } from "../conn/mod";
 import { convertConnToBarePersistedConn } from "../conn/persisted";
 import type { ActorDriver } from "../driver";
@@ -210,7 +211,7 @@ export class StateManager<S, CP, CS, I> {
 			} else {
 				// Create promise for waiting
 				if (!this.#onPersistSavedPromise) {
-					this.#onPersistSavedPromise = promiseWithResolvers();
+					this.#onPersistSavedPromise = promiseWithResolvers((reason) => loggerWithoutContext().warn({ msg: "unhandled persist saved promise rejection", reason }));
 				}
 
 				// Save throttled

rivetkit-typescript/packages/rivetkit/src/actor/router-websocket-endpoints.ts

@@ -86,7 +86,7 @@ export async function routeWebSocket(
 		});
 
 		// Promise used to wait for the websocket close in `disconnect`
-		const closePromiseResolvers = promiseWithResolvers<void>();
+		const closePromiseResolvers = promiseWithResolvers<void>((reason) => loggerWithoutContext().warn({ msg: "unhandled websocket close promise rejection", reason }));
 
 		// Strip query parameters from requestPath for routing purposes.
 		// This handles paths like "/websocket?query=value" which should route

rivetkit-typescript/packages/rivetkit/src/client/actor-conn.ts

@@ -228,7 +228,7 @@ export class ActorConnRaw {
 		const { promise, resolve, reject } = promiseWithResolvers<{
 			id: bigint;
 			output: unknown;
-		}>();
+		}>((reason) => logger().warn({ msg: "unhandled action promise rejection", reason }));
 		this.#actionsInFlight.set(actionId, {
 			name: opts.name,
 			resolve,
@@ -359,7 +359,7 @@ export class ActorConnRaw {
 			// Create promise for open
 			if (this.#onOpenPromise)
 				throw new Error("#onOpenPromise already defined");
-			this.#onOpenPromise = promiseWithResolvers();
+			this.#onOpenPromise = promiseWithResolvers((reason) => logger().warn({ msg: "unhandled open promise rejection", reason }));
 
 			await this.#connectWebSocket();
 
@@ -1189,7 +1189,7 @@ export class ActorConnRaw {
 				ws.readyState !== 2 /* CLOSING */ &&
 				ws.readyState !== 3 /* CLOSED */
 			) {
-				const { promise, resolve } = promiseWithResolvers();
+				const { promise, resolve } = promiseWithResolvers((reason) => logger().warn({ msg: "unhandled websocket close promise rejection", reason }));
 				ws.addEventListener("close", () => resolve(undefined));
 				ws.close(1000, "Disposed");
 				await promise;

rivetkit-typescript/packages/rivetkit/src/drivers/engine/actor-driver.ts

@@ -82,8 +82,8 @@ export class EngineActorDriver implements ActorDriver {
 	#actors: Map<string, ActorHandler> = new Map();
 	#actorRouter: ActorRouter;
 
-	#runnerStarted: PromiseWithResolvers<undefined> = promiseWithResolvers();
-	#runnerStopped: PromiseWithResolvers<undefined> = promiseWithResolvers();
+	#runnerStarted: PromiseWithResolvers<undefined> = promiseWithResolvers((reason) => logger().warn({ msg: "unhandled runner started promise rejection", reason }));
+	#runnerStopped: PromiseWithResolvers<undefined> = promiseWithResolvers((reason) => logger().warn({ msg: "unhandled runner stopped promise rejection", reason }));
 	#isRunnerStopped: boolean = false;
 
 	// HACK: Track actor stop intent locally since the runner protocol doesn't
@@ -426,7 +426,7 @@ export class EngineActorDriver implements ActorDriver {
 			// async operations to avoid race conditions where multiple calls might try to
 			// create the same handler simultaneously.
 			handler = {
-				actorStartPromise: promiseWithResolvers(),
+				actorStartPromise: promiseWithResolvers((reason) => logger().warn({ msg: "unhandled actor start promise rejection", reason })),
 			};
 			this.#actors.set(actorId, handler);
 		}
@@ -435,40 +435,50 @@ export class EngineActorDriver implements ActorDriver {
 		invariant(actorConfig.key, "actor should have a key");
 		const key = deserializeActorKey(actorConfig.key);
 
-		// Initialize storage
-		const [persistDataBuffer] = await this.#runner.kvGet(actorId, [
-			KEYS.PERSIST_DATA,
-		]);
-		if (persistDataBuffer === null) {
-			const initialKvState = getInitialActorKvState(input);
-			await this.#runner.kvPut(actorId, initialKvState);
-			logger().debug({
-				msg: "initialized persist data for new actor",
-				actorId,
-			});
-		} else {
-			logger().debug({
-				msg: "found existing persist data for actor",
-				actorId,
-				dataSize: persistDataBuffer.byteLength,
-			});
-		}
+		try {
+			// Initialize storage
+			const [persistDataBuffer] = await this.#runner.kvGet(actorId, [
+				KEYS.PERSIST_DATA,
+			]);
+			if (persistDataBuffer === null) {
+				const initialKvState = getInitialActorKvState(input);
+				await this.#runner.kvPut(actorId, initialKvState);
+				logger().debug({
+					msg: "initialized persist data for new actor",
+					actorId,
+				});
+			} else {
+				logger().debug({
+					msg: "found existing persist data for actor",
+					actorId,
+					dataSize: persistDataBuffer.byteLength,
+				});
+			}
 
-		// Create actor instance
-		const definition = lookupInRegistry(this.#config, actorConfig.name);
-		handler.actor = await definition.instantiate();
+			// Create actor instance
+			const definition = lookupInRegistry(this.#config, actorConfig.name);
+			handler.actor = await definition.instantiate();
 
-		// Start actor
-		await handler.actor.start(
-			this,
-			this.#inlineClient,
-			actorId,
-			name,
-			key,
-			"unknown", // TODO: Add regions
-		);
+			// Start actor
+			await handler.actor.start(
+				this,
+				this.#inlineClient,
+				actorId,
+				name,
+				key,
+				"unknown", // TODO: Add regions
+			);
 
-		logger().debug({ msg: "runner actor started", actorId, name, key });
+			logger().debug({ msg: "runner actor started", actorId, name, key });
+		} catch (innerError) {
+			const error = new Error(
+				`Failed to start actor ${actorId}: ${innerError}`,
+				{ cause: innerError },
+			);
+			handler.actorStartPromise?.reject(error);
+			handler.actorStartPromise = undefined;
+			throw error;
+		}
 	}
 
 	async #runnerOnActorStop(

rivetkit-typescript/packages/rivetkit/src/drivers/file-system/global-state.ts

@@ -434,7 +434,7 @@ export class FileSystemGlobalState {
 			return;
 		}
 		actor.lifecycleState = ActorLifecycleState.STARTING_SLEEP;
-		actor.stopPromise = promiseWithResolvers();
+		actor.stopPromise = promiseWithResolvers((reason) => logger().warn({ msg: "unhandled actor sleep stop promise rejection", reason }));
 
 		// Wait for actor to fully start before stopping it to avoid race conditions
 		if (actor.loadPromise) await actor.loadPromise.catch();
@@ -486,7 +486,7 @@ export class FileSystemGlobalState {
 			return;
 		}
 		actor.lifecycleState = ActorLifecycleState.STARTING_DESTROY;
-		actor.stopPromise = promiseWithResolvers();
+		actor.stopPromise = promiseWithResolvers((reason) => logger().warn({ msg: "unhandled actor destroy stop promise rejection", reason }));
 
 		// Wait for actor to fully start before stopping it to avoid race conditions
 		if (actor.loadPromise) await actor.loadPromise.catch();
@@ -658,7 +658,7 @@ export class FileSystemGlobalState {
 		invariant(entry, "actor entry does not exist");
 
 		const previousWrite = entry.pendingWriteResolver;
-		const currentWrite = promiseWithResolvers<void>();
+		const currentWrite = promiseWithResolvers<void>((reason) => logger().warn({ msg: "unhandled kv write promise rejection", reason }));
 		entry.pendingWriteResolver = currentWrite;
 
 		if (previousWrite) {
@@ -895,7 +895,7 @@ export class FileSystemGlobalState {
 		}
 
 		// Create start promise
-		entry.startPromise = promiseWithResolvers();
+		entry.startPromise = promiseWithResolvers((reason) => logger().warn({ msg: "unhandled actor start promise rejection", reason }));
 
 		try {
 			// Create actor

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

@@ -425,7 +425,7 @@ export async function createTestWebSocketProxy(
 		promise: clientToProxyWsPromise,
 		resolve: clientToProxyWsResolve,
 		reject: clientToProxyWsReject,
-	} = promiseWithResolvers<WSContext>();
+	} = promiseWithResolvers<WSContext>((reason) => logger().warn({ msg: "unhandled client websocket promise rejection", reason }));
 	try {
 		// Resolve the client WebSocket promise
 		logger().debug({ msg: "awaiting client websocket promise" });

rivetkit-typescript/packages/rivetkit/src/utils.ts

@@ -94,7 +94,7 @@ export type LongTimeoutHandle = { abort: () => void };
  *
  * This is specifically for Cloudflare Workers. Their implementation of Promise.withResolvers does not work correctly.
  */
-export function promiseWithResolvers<T>(): {
+export function promiseWithResolvers<T>(onReject: (reason?: any) => void): {
 	promise: Promise<T>;
 	resolve: (value: T | PromiseLike<T>) => void;
 	reject: (reason?: any) => void;
@@ -105,6 +105,7 @@ export function promiseWithResolvers<T>(): {
 		resolve = res;
 		reject = rej;
 	});
+	promise.catch(onReject);
 	return { promise, resolve, reject };
 }
 
@@ -155,7 +156,7 @@ export class SinglePromiseQueue {
 
 		// Ensure a shared resolver exists for all callers in this cycle
 		if (!this.#pending) {
-			this.#pending = promiseWithResolvers<void>();
+			this.#pending = promiseWithResolvers<void>((reason) => logger().warn({ msg: "unhandled single promise queue rejection", reason }));
 		}
 
 		const waitForThisCycle = this.#pending.promise;