feat: actor messages & queues

Author: NathanFlurry

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

@@ -0,0 +1,70 @@
+import { actor } from "rivetkit";
+import type { registry } from "./registry";
+
+export const queueActor = actor({
+	state: {},
+	actions: {
+		receiveOne: async (
+			c,
+			name: string,
+			opts?: { count?: number; timeout?: number },
+		) => {
+			const message = await c.queue.next(name, opts);
+			if (!message) {
+				return null;
+			}
+			return { name: message.name, body: message.body };
+		},
+		receiveMany: async (
+			c,
+			names: string[],
+			opts?: { count?: number; timeout?: number },
+		) => {
+			const messages = await c.queue.next(names, opts);
+			return (messages ?? []).map(
+				(message: { name: string; body: unknown }) => ({
+					name: message.name,
+					body: message.body,
+				}),
+			);
+		},
+		receiveRequest: async (
+			c,
+			request: {
+				name: string | string[];
+				count?: number;
+				timeout?: number;
+			},
+		) => {
+			const messages = await c.queue.next(request);
+			return (messages ?? []).map(
+				(message: { name: string; body: unknown }) => ({
+					name: message.name,
+					body: message.body,
+				}),
+			);
+		},
+		sendToSelf: async (c, name: string, body: unknown) => {
+			const client = c.client<typeof registry>();
+			const handle = client.queueActor.getForId(c.actorId);
+			await handle.queue[name].send(body);
+			return true;
+		},
+		waitForAbort: async (c) => {
+			setTimeout(() => {
+				c.destroy();
+			}, 10);
+			await c.queue.next("abort", { timeout: 10_000 });
+			return true;
+		},
+	},
+});
+
+export const queueLimitedActor = actor({
+	state: {},
+	actions: {},
+	options: {
+		maxQueueSize: 1,
+		maxQueueMessageSize: 64,
+	},
+});

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

@@ -26,6 +26,7 @@ import { kvActor } from "./kv";
 import { largePayloadActor, largePayloadConnActor } from "./large-payloads";
 import { counterWithLifecycle } from "./lifecycle";
 import { metadataActor } from "./metadata";
+import { queueActor, queueLimitedActor } from "./queue";
 import {
 	rawHttpActor,
 	rawHttpHonoActor,
@@ -75,6 +76,9 @@ export const registry = setup({
 		inlineClientActor,
 		// From kv.ts
 		kvActor,
+		// From queue.ts
+		queueActor,
+		queueLimitedActor,
 		// From action-inputs.ts
 		inputActor,
 		// From action-timeout.ts

rivetkit-typescript/packages/rivetkit/schemas/actor-inspector/v1.bare

@@ -101,6 +101,7 @@ type Init struct {
 	isStateEnabled: bool
 	rpcs: list<str>
 	isDatabaseEnabled: bool
+	queueSize: uint
 }
 
 type ConnectionsResponse struct {
@@ -132,6 +133,10 @@ type EventsUpdated struct {
 	events: list<Event>
 }
 
+type QueueUpdated struct {
+	queueSize: uint
+}
+
 type RpcsListResponse struct {
 	rid: uint
 	rpcs: list<str>
@@ -151,6 +156,7 @@ type ToClientBody union {
 	ActionResponse |
 	ConnectionsUpdated |
 	EventsUpdated |
+	QueueUpdated |
 	StateUpdated |
 	RpcsListResponse |
 	Error |

rivetkit-typescript/packages/rivetkit/schemas/client-protocol/v1.bare

@@ -67,6 +67,17 @@ type HttpActionResponse struct {
 	output: data
 }
 
+# MARK: HTTP Queue
+
+type HttpQueueSendRequest struct {
+	name: str
+	body: data
+}
+
+type HttpQueueSendResponse struct {
+	ok: bool
+}
+
 # MARK: HTTP Error
 type HttpResponseError struct {
 	group: str

rivetkit-typescript/packages/rivetkit/schemas/client-protocol/v2.bare

@@ -66,6 +66,17 @@ type HttpActionResponse struct {
 	output: data
 }
 
+# MARK: HTTP Queue
+
+type HttpQueueSendRequest struct {
+	name: str
+	body: data
+}
+
+type HttpQueueSendResponse struct {
+	ok: bool
+}
+
 # MARK: HTTP Error
 type HttpResponseError struct {
 	group: str

rivetkit-typescript/packages/rivetkit/src/actor/config.ts

@@ -83,6 +83,11 @@ export const ActorConfigSchema = z
 				connectionLivenessInterval: z.number().positive().default(5000),
 				noSleep: z.boolean().default(false),
 				sleepTimeout: z.number().positive().default(30_000),
+				maxQueueSize: z.number().positive().default(1000),
+				maxQueueMessageSize: z
+					.number()
+					.positive()
+					.default(1024 * 1024),
 				/**
 				 * Can hibernate WebSockets for onWebSocket.
 				 *
@@ -689,6 +694,18 @@ export const DocActorOptionsSchema = z
 			.describe(
 				"Time in ms of inactivity before the actor sleeps. Default: 30000",
 			),
+		maxQueueSize: z
+			.number()
+			.optional()
+			.describe(
+				"Maximum number of queue messages before rejecting new messages. Default: 1000",
+			),
+		maxQueueMessageSize: z
+			.number()
+			.optional()
+			.describe(
+				"Maximum size of each queue message in bytes. Default: 1048576",
+			),
 		canHibernateWebSocket: z
 			.boolean()
 			.optional()

rivetkit-typescript/packages/rivetkit/src/actor/contexts/base/actor.ts

@@ -8,6 +8,7 @@ import type { ActorDefinition, AnyActorDefinition } from "../../definition";
 import type { ActorInstance, SaveStateOptions } from "../../instance/mod";
 import { ActorKv } from "../../instance/kv";
 import type { Schedule } from "../../schedule";
+import type { QueueContext } from "../queue";
 
 /**
  * ActorContext class that provides access to actor methods and state
@@ -91,6 +92,20 @@ export class ActorContext<
 		return this.#actor.log;
 	}
 
+	/**
+	 * Access to queue receive helpers.
+	 */
+	get queue(): QueueContext<
+		TState,
+		TConnParams,
+		TConnState,
+		TVars,
+		TInput,
+		TDatabase
+	> {
+		return this.#actor.queue;
+	}
+
 	/**
 	 * Gets actor ID.
 	 */

rivetkit-typescript/packages/rivetkit/src/actor/contexts/index.ts

@@ -1,10 +1,10 @@
 // Base contexts
-export { ActorContext, type ActorContextOf } from "./base/actor";
-export { ConnContext, type ConnContextOf } from "./base/conn";
-export { ConnInitContext, type ConnInitContextOf } from "./base/conn-init";
 
 // Lifecycle contexts
 export { ActionContext, type ActionContextOf } from "./action";
+export { ActorContext, type ActorContextOf } from "./base/actor";
+export { ConnContext, type ConnContextOf } from "./base/conn";
+export { ConnInitContext, type ConnInitContextOf } from "./base/conn-init";
 export {
 	BeforeActionResponseContext,
 	type BeforeActionResponseContextOf,
@@ -22,6 +22,7 @@ export {
 export { CreateVarsContext, type CreateVarsContextOf } from "./create-vars";
 export { DestroyContext, type DestroyContextOf } from "./destroy";
 export { DisconnectContext, type DisconnectContextOf } from "./disconnect";
+export { QueueContext, type QueueContextOf } from "./queue";
 export { RequestContext, type RequestContextOf } from "./request";
 export { SleepContext, type SleepContextOf } from "./sleep";
 export {

rivetkit-typescript/packages/rivetkit/src/actor/contexts/queue.ts

@@ -0,0 +1,98 @@
+import type { AnyDatabaseProvider } from "../database";
+import type { ActorDefinition, AnyActorDefinition } from "../definition";
+import type { ActorInstance } from "../instance/mod";
+import type { QueueMessage } from "../instance/queue-manager";
+
+export interface QueueReceiveOptions {
+	count?: number;
+	timeout?: number;
+}
+
+export interface QueueReceiveRequest extends QueueReceiveOptions {
+	name: string | string[];
+}
+
+export class QueueContext<
+	TState,
+	TConnParams,
+	TConnState,
+	TVars,
+	TInput,
+	TDatabase extends AnyDatabaseProvider,
+> {
+	#actor: ActorInstance<
+		TState,
+		TConnParams,
+		TConnState,
+		TVars,
+		TInput,
+		TDatabase
+	>;
+
+	constructor(
+		actor: ActorInstance<
+			TState,
+			TConnParams,
+			TConnState,
+			TVars,
+			TInput,
+			TDatabase
+		>,
+	) {
+		this.#actor = actor;
+	}
+
+	next(
+		name: string,
+		opts?: QueueReceiveOptions,
+	): Promise<QueueMessage | undefined>;
+	next(
+		name: string[],
+		opts?: QueueReceiveOptions,
+	): Promise<QueueMessage[] | undefined>;
+	next(request: QueueReceiveRequest): Promise<QueueMessage[] | undefined>;
+	async next(
+		nameOrRequest: string | string[] | QueueReceiveRequest,
+		opts: QueueReceiveOptions = {},
+	): Promise<QueueMessage | QueueMessage[] | undefined> {
+		const request =
+			typeof nameOrRequest === "object" && !Array.isArray(nameOrRequest)
+				? nameOrRequest
+				: { name: nameOrRequest };
+		const mergedOptions = request === nameOrRequest ? request : opts;
+		const names = Array.isArray(request.name)
+			? request.name
+			: [request.name];
+		const count = mergedOptions.count ?? 1;
+
+		const messages = await this.#actor.queueManager.receive(
+			names,
+			count,
+			mergedOptions.timeout,
+			this.#actor.abortSignal,
+		);
+
+		if (Array.isArray(request.name)) {
+			return messages;
+		}
+
+		if (!messages || messages.length === 0) {
+			return undefined;
+		}
+
+		return messages[0];
+	}
+}
+
+export type QueueContextOf<AD extends AnyActorDefinition> =
+	AD extends ActorDefinition<
+		infer S,
+		infer CP,
+		infer CS,
+		infer V,
+		infer I,
+		infer DB extends AnyDatabaseProvider,
+		any
+	>
+		? QueueContext<S, CP, CS, V, I, DB>
+		: never;

rivetkit-typescript/packages/rivetkit/src/actor/errors.ts

@@ -188,6 +188,45 @@ export class Unsupported extends ActorError {
 	}
 }
 
+export class QueueFull extends ActorError {
+	constructor(limit: number) {
+		super("queue", "full", `Queue is full. Limit is ${limit} messages.`, {
+			public: true,
+			metadata: { limit },
+		});
+	}
+}
+
+export class QueueMessageTooLarge extends ActorError {
+	constructor(size: number, limit: number) {
+		super(
+			"queue",
+			"message_too_large",
+			`Queue message too large (${size} bytes). Limit is ${limit} bytes.`,
+			{ public: true, metadata: { size, limit } },
+		);
+	}
+}
+
+export class QueueMessageInvalid extends ActorError {
+	constructor(path?: string) {
+		super(
+			"queue",
+			"message_invalid",
+			path
+				? `Queue message body contains unsupported type at ${path}.`
+				: "Queue message body contains unsupported type.",
+			{ public: true, metadata: path ? { path } : undefined },
+		);
+	}
+}
+
+export class ActorAborted extends ActorError {
+	constructor() {
+		super("actor", "aborted", "Actor aborted.", { public: true });
+	}
+}
+
 /**
  * Options for the UserError class.
  */