feat: add configurable timeouts for actor lifecycle hooks (#885)

Author: NathanFlurry

.github/workflows/test.yml

@@ -3,6 +3,18 @@
 name: 'Test'
 on: 
   pull_request:
+    paths:
+      - 'packages/**'
+      - 'clients/**'
+      - 'examples/**'
+      - '.github/workflows/**'
+      - 'package.json'
+      - 'yarn.lock'
+      - 'tsconfig*.json'
+      - 'turbo.json'
+      - 'tsup.base.ts'
+      - 'biome.json'
+      - 'Cargo.toml'
   workflow_dispatch:
 
 jobs:

packages/actor-core/package.json

@@ -2,7 +2,13 @@
   "name": "actor-core",
   "version": "0.7.9",
   "license": "Apache-2.0",
-  "files": ["dist", "src", "deno.json", "bun.json", "package.json"],
+  "files": [
+    "dist",
+    "src",
+    "deno.json",
+    "bun.json",
+    "package.json"
+  ],
   "type": "module",
   "bin": "./dist/cli/mod.cjs",
   "exports": {

packages/actor-core/src/actor/config.ts

@@ -26,6 +26,14 @@ export const ActorConfigSchema = z
 		createVars: z.function().optional(),
 		options: z
 			.object({
+				lifecycle: z
+					.object({
+						createVarsTimeout: z.number().positive().default(5000),
+						createConnStateTimeout: z.number().positive().default(5000),
+						onConnectTimeout: z.number().positive().default(5000),
+					})
+					.strict()
+					.default({}),
 				state: z
 					.object({
 						saveInterval: z.number().positive().default(10_000),
@@ -81,7 +89,11 @@ export interface OnConnectOptions<CP> {
 // This must have only one or the other or else S will not be able to be inferred
 type CreateState<S, CP, CS, V> =
 	| { state: S }
-	| { createState: (c: ActorContext<undefined, undefined, undefined, undefined>) => S | Promise<S> }
+	| {
+			createState: (
+				c: ActorContext<undefined, undefined, undefined, undefined>,
+			) => S | Promise<S>;
+	  }
 	| Record<never, never>;
 
 // Creates connection state config
@@ -114,7 +126,10 @@ type CreateVars<S, CP, CS, V> =
 			/**
 			 * @experimental
 			 */
-			createVars: (c: ActorContext<undefined, undefined, undefined, undefined>, driverCtx: unknown) => V | Promise<V>;
+			createVars: (
+				c: ActorContext<undefined, undefined, undefined, undefined>,
+				driverCtx: unknown,
+			) => V | Promise<V>;
 	  }
 	| Record<never, never>;
 

packages/actor-core/src/actor/instance.ts

@@ -13,14 +13,18 @@ import * as errors from "./errors";
 import { processMessage } from "./protocol/message/mod";
 import { instanceLogger, logger } from "./log";
 import type { ActionContext } from "./action";
-import { Lock, deadline } from "./utils";
+import { DeadlineError, Lock, deadline } from "./utils";
 import { Schedule } from "./schedule";
 import type * as wsToServer from "@/actor/protocol/message/to-server";
 import { CachedSerializer } from "./protocol/serde";
 import { ActorInspector } from "@/inspector/actor";
 import { ActorContext } from "./context";
 import invariant from "invariant";
-import type { PersistedActor, PersistedConn, PersistedScheduleEvents } from "./persisted";
+import type {
+	PersistedActor,
+	PersistedConn,
+	PersistedScheduleEvents,
+} from "./persisted";
 
 /**
  * Options for the `_saveState` method.
@@ -160,9 +164,6 @@ export class ActorInstance<S, CP, CS, V> {
 
 		// TODO: Exit process if this errors
 		if (this.#varsEnabled) {
-			// TODO: Move to config
-			const CREATE_VARS_TIMEOUT = 5000; // 5 seconds
-
 			let vars: V | undefined = undefined;
 			if ("createVars" in this.#config) {
 				const dataOrPromise = this.#config.createVars(
@@ -175,7 +176,10 @@ export class ActorInstance<S, CP, CS, V> {
 					this.#actorDriver.getContext(this.#actorId),
 				);
 				if (dataOrPromise instanceof Promise) {
-					vars = await deadline(dataOrPromise, CREATE_VARS_TIMEOUT);
+					vars = await deadline(
+						dataOrPromise,
+						this.#config.options.lifecycle.createVarsTimeout,
+					);
 				} else {
 					vars = dataOrPromise;
 				}
@@ -214,10 +218,10 @@ export class ActorInstance<S, CP, CS, V> {
 			ar: args,
 		};
 
-		this.actorContext.log.info("scheduling event", { 
-			event: eventId, 
-			timestamp, 
-			action: fn 
+		this.actorContext.log.info("scheduling event", {
+			event: eventId,
+			timestamp,
+			action: fn,
 		});
 
 		// Insert event in to index
@@ -239,7 +243,10 @@ export class ActorInstance<S, CP, CS, V> {
 
 	async onAlarm() {
 		const now = Date.now();
-		this.actorContext.log.debug("alarm triggered", { now, events: this.#persist.e.length });
+		this.actorContext.log.debug("alarm triggered", {
+			now,
+			events: this.#persist.e.length,
+		});
 
 		// Remove events from schedule that we're about to run
 		const runIndex = this.#persist.e.findIndex((x) => x.t <= now);
@@ -248,7 +255,9 @@ export class ActorInstance<S, CP, CS, V> {
 			return;
 		}
 		const scheduleEvents = this.#persist.e.splice(0, runIndex + 1);
-		this.actorContext.log.debug("running events", { count: scheduleEvents.length });
+		this.actorContext.log.debug("running events", {
+			count: scheduleEvents.length,
+		});
 
 		// Set alarm for next event
 		if (this.#persist.e.length > 0) {
@@ -258,13 +267,13 @@ export class ActorInstance<S, CP, CS, V> {
 		// Iterate by event key in order to ensure we call the events in order
 		for (const event of scheduleEvents) {
 			try {
-				this.actorContext.log.info("running action for event", { 
-					event: event.e, 
-					timestamp: event.t, 
-					action: event.a, 
-					args: event.ar 
+				this.actorContext.log.info("running action for event", {
+					event: event.e,
+					timestamp: event.t,
+					action: event.a,
+					args: event.ar,
 				});
-				
+
 				// Look up function
 				const fn: unknown = this.#config.actions[event.a];
 				if (!fn) throw new Error(`Missing action for alarm ${event.a}`);
@@ -586,7 +595,6 @@ export class ActorInstance<S, CP, CS, V> {
 	): Promise<CS> {
 		// Authenticate connection
 		let connState: CS | undefined = undefined;
-		const PREPARE_CONNECT_TIMEOUT = 5000; // 5 seconds
 
 		const onBeforeConnectOpts = {
 			request,
@@ -612,7 +620,10 @@ export class ActorInstance<S, CP, CS, V> {
 					onBeforeConnectOpts,
 				);
 				if (dataOrPromise instanceof Promise) {
-					connState = await deadline(dataOrPromise, PREPARE_CONNECT_TIMEOUT);
+					connState = await deadline(
+						dataOrPromise,
+						this.#config.options.lifecycle.createConnStateTimeout,
+					);
 				} else {
 					connState = dataOrPromise;
 				}
@@ -676,12 +687,14 @@ export class ActorInstance<S, CP, CS, V> {
 		this.inspector.onConnChange(this.#connections);
 
 		// Handle connection
-		const CONNECT_TIMEOUT = 5000; // 5 seconds
 		if (this.#config.onConnect) {
 			try {
 				const result = this.#config.onConnect(this.actorContext, conn);
 				if (result instanceof Promise) {
-					deadline(result, CONNECT_TIMEOUT).catch((error) => {
+					deadline(
+						result,
+						this.#config.options.lifecycle.onConnectTimeout,
+					).catch((error) => {
 						logger().error("error in `onConnect`, closing socket", {
 							error,
 						});
@@ -842,13 +855,22 @@ export class ActorInstance<S, CP, CS, V> {
 		// TODO: Manually call abortable for better error handling
 		// Call the function on this object with those arguments
 		try {
+			// Log when we start executing the action
+			logger().debug("executing action", { actionName: rpcName, args });
+
 			const outputOrPromise = rpcFunction.call(undefined, ctx, ...args);
 			let output: unknown;
 			if (outputOrPromise instanceof Promise) {
+				// Log that we're waiting for an async action
+				logger().debug("awaiting async action", { actionName: rpcName });
+
 				output = await deadline(
 					outputOrPromise,
 					this.#config.options.action.timeout,
 				);
+
+				// Log that async action completed
+				logger().debug("async action completed", { actionName: rpcName });
 			} else {
 				output = outputOrPromise;
 			}
@@ -863,7 +885,13 @@ export class ActorInstance<S, CP, CS, V> {
 						output,
 					);
 					if (processedOutput instanceof Promise) {
+						logger().debug("awaiting onBeforeActionResponse", {
+							actionName: rpcName,
+						});
 						output = await processedOutput;
+						logger().debug("onBeforeActionResponse completed", {
+							actionName: rpcName,
+						});
 					} else {
 						output = processedOutput;
 					}
@@ -874,11 +902,22 @@ export class ActorInstance<S, CP, CS, V> {
 				}
 			}
 
+			// Log the output before returning
+			logger().debug("action completed", {
+				actionName: rpcName,
+				outputType: typeof output,
+				isPromise: output instanceof Promise,
+			});
+
 			return output;
 		} catch (error) {
-			if (error instanceof DOMException && error.name === "TimeoutError") {
+			if (error instanceof DeadlineError) {
 				throw new errors.ActionTimedOut();
 			}
+			logger().error("action error", {
+				actionName: rpcName,
+				error: stringifyError(error),
+			});
 			throw error;
 		} finally {
 			this.#savePersistThrottled();

packages/actor-core/src/actor/protocol/message/mod.ts

@@ -101,9 +101,22 @@ export async function processMessage<S, CP, CS, V>(
 			rpcId = id;
 			rpcName = name;
 
+			logger().debug("processing RPC request", { id, name, argsCount: args.length });
+			
 			const ctx = new ActionContext<S, CP, CS, V>(actor.actorContext, conn);
+			
+			// Process the RPC request and wait for the result
+			// This will wait for async actions to complete
 			const output = await handler.onExecuteRpc(ctx, name, args);
-
+			
+			logger().debug("sending RPC response", { 
+				id, 
+				name, 
+				outputType: typeof output, 
+				isPromise: output instanceof Promise 
+			});
+
+			// Send the response back to the client
 			conn._sendMessage(
 				new CachedSerializer<wsToClient.ToClient>({
 					b: {
@@ -114,6 +127,8 @@ export async function processMessage<S, CP, CS, V>(
 					},
 				}),
 			);
+			
+			logger().debug("RPC response sent", { id, name });
 		} else if ("sr" in message.b) {
 			// Subscription request
 
@@ -125,12 +140,15 @@ export async function processMessage<S, CP, CS, V>(
 			}
 
 			const { e: eventName, s: subscribe } = message.b.sr;
+			logger().debug("processing subscription request", { eventName, subscribe });
 
 			if (subscribe) {
 				await handler.onSubscribe(eventName, conn);
 			} else {
 				await handler.onUnsubscribe(eventName, conn);
 			}
+			
+			logger().debug("subscription request completed", { eventName, subscribe });
 		} else {
 			assertUnreachable(message.b);
 		}
@@ -141,6 +159,13 @@ export async function processMessage<S, CP, CS, V>(
 			rpcName,
 		});
 
+		logger().debug("sending error response", {
+			rpcId,
+			rpcName,
+			code,
+			message
+		});
+
 		// Build response
 		if (rpcId !== undefined) {
 			conn._sendMessage(
@@ -168,5 +193,7 @@ export async function processMessage<S, CP, CS, V>(
 				}),
 			);
 		}
+		
+		logger().debug("error response sent", { rpcId, rpcName });
 	}
 }

packages/actor-core/src/actor/utils.ts

@@ -33,6 +33,12 @@ export const throttle = <
 	};
 };
 
+export class DeadlineError extends Error {
+	constructor() {
+		super("Promise did not complete before deadline.")
+	}
+}
+
 export function deadline<T>(promise: Promise<T>, timeout: number): Promise<T> {
 	const controller = new AbortController();
 	const signal = controller.signal;
@@ -44,7 +50,7 @@ export function deadline<T>(promise: Promise<T>, timeout: number): Promise<T> {
 		promise,
 		new Promise<T>((_, reject) => {
 			signal.addEventListener("abort", () =>
-				reject(new Error("Operation timed out")),
+				reject(new DeadlineError()),
 			);
 		}),
 	]).finally(() => {