fix(rivetkit): wire CancellationToken through waitForNamesAvailable to eliminate KV busy-polling

Author: abcxff

rivetkit-typescript/packages/rivetkit-napi/src/queue.rs

@@ -162,9 +162,10 @@ impl Queue {
 		&self,
 		names: Vec<String>,
 		options: Option<JsQueueWaitOptions>,
+		signal: Option<&CancellationToken>,
 	) -> napi::Result<()> {
 		self.inner
-			.wait_for_names_available(names, queue_wait_opts(options, None)?)
+			.wait_for_names_available(names, queue_wait_opts(options, signal)?)
 			.await
 			.map_err(napi_anyhow_error)
 	}

rivetkit-typescript/packages/rivetkit/src/registry/napi-runtime.ts

@@ -677,10 +677,11 @@ export class NapiCoreRuntime implements CoreRuntime {
 		ctx: ActorContextHandle,
 		names: string[],
 		options?: RuntimeQueueWaitOptions | undefined | null,
+		signal?: CancellationTokenHandle | undefined | null,
 	): Promise<void> {
 		await asNativeActorContext(ctx)
 			.queue()
-			.waitForNamesAvailable(names, options);
+			.waitForNamesAvailable(names, options, signal);
 	}
 
 	async actorQueueEnqueueAndWait(

rivetkit-typescript/packages/rivetkit/src/registry/native.ts

@@ -1649,65 +1649,24 @@ class NativeQueueAdapter {
 			signal?: AbortSignal;
 		},
 	) {
-		if (!options?.signal) {
+		const { token, cleanup } = await createCancellationTokenHandle(
+			this.#runtime,
+			options?.signal,
+		);
+
+		try {
 			await callNative(() =>
 				this.#runtime.actorQueueWaitForNamesAvailable(
 					this.#ctx,
 					[...names],
 					{
 						timeoutMs: options?.timeout,
 					},
+					token,
 				),
 			);
-			return;
-		}
-
-		const deadline =
-			options.timeout === undefined
-				? undefined
-				: Date.now() + options.timeout;
-
-		for (;;) {
-			if (options.signal.aborted) {
-				throw actorAbortedError();
-			}
-
-			const remainingTimeout =
-				deadline === undefined
-					? undefined
-					: Math.max(0, deadline - Date.now());
-			const sliceTimeout =
-				remainingTimeout === undefined
-					? 100
-					: Math.min(remainingTimeout, 100);
-
-			try {
-				await callNative(() =>
-					this.#runtime.actorQueueWaitForNamesAvailable(
-						this.#ctx,
-						[...names],
-						{
-							timeoutMs: sliceTimeout,
-						},
-					),
-				);
-				return;
-			} catch (error) {
-				if (
-					(error as { group?: string; code?: string }).group ===
-						"queue" &&
-					(error as { group?: string; code?: string }).code ===
-						"timed_out"
-				) {
-					if (
-						remainingTimeout === undefined ||
-						remainingTimeout > 100
-					) {
-						continue;
-					}
-				}
-				throw error;
-			}
+		} finally {
+			cleanup?.();
 		}
 	}
 

rivetkit-typescript/packages/rivetkit/src/registry/runtime.ts

@@ -491,6 +491,7 @@ export interface CoreRuntime {
 		ctx: ActorContextHandle,
 		names: string[],
 		options?: RuntimeQueueWaitOptions | undefined | null,
+		signal?: CancellationTokenHandle | undefined | null,
 	): Promise<void>;
 	actorQueueEnqueueAndWait(
 		ctx: ActorContextHandle,

rivetkit-typescript/packages/rivetkit/src/registry/wasm-runtime.ts

@@ -764,9 +764,10 @@ export class WasmCoreRuntime implements CoreRuntime {
 		ctx: ActorContextHandle,
 		names: string[],
 		options?: RuntimeQueueWaitOptions | undefined | null,
+		signal?: CancellationTokenHandle | undefined | null,
 	): Promise<void> {
 		const queue = childHandle(asWasmActorContext(ctx), "queue");
-		await callHandleAsync(queue, "waitForNamesAvailable", names, options);
+		await callHandleAsync(queue, "waitForNamesAvailable", names, options, signal);
 	}
 
 	async actorQueueEnqueueAndWait(