diff --git a/docs-internal/engine/napi-bridge.md b/docs-internal/engine/napi-bridge.md index 1bbda8b1bb..0fc19c9cd8 100644 --- a/docs-internal/engine/napi-bridge.md +++ b/docs-internal/engine/napi-bridge.md @@ -25,7 +25,7 @@ Rules for `rivetkit-typescript/packages/rivetkit-napi/`. The bridge is pure plum ## Payload + error conventions - `#[napi(object)]` bridge payloads stay plain-data only. If TypeScript needs to cancel native work, use primitives or JS-side polling instead of trying to pass a `#[napi]` class instance through an object field. -- N-API structured errors cross the JS<->Rust boundary by prefix-encoding `{ group, code, message, metadata }` into `napi::Error.reason`, then normalizing that prefix back into a `RivetError` on the other side. +- N-API structured errors cross the JS<->Rust boundary by prefix-encoding `{ group, code, message, metadata, rayId }` into `napi::Error.reason`, then normalizing that prefix back into a `RivetError` on the other side. - N-API bridge debug logs use stable `kind` plus compact payload summaries, never raw buffers or full request bodies. ## Receive-loop state lifecycle diff --git a/frontend/src/app/dialogs/connect-manual-serverless-frame.tsx b/frontend/src/app/dialogs/connect-manual-serverless-frame.tsx index f5badc696e..ee539aedde 100644 --- a/frontend/src/app/dialogs/connect-manual-serverless-frame.tsx +++ b/frontend/src/app/dialogs/connect-manual-serverless-frame.tsx @@ -188,6 +188,7 @@ export const buildServerlessConfig = async ( slotsPerRunner: values.slotsPerRunner ?? 1, runnersMargin: values.runnerMargin ?? 0, minRunners: values.minRunners ?? 0, + drainGracePeriod: values.drainGracePeriod, }; const resolvedProvider = provider || "custom"; const isCustom = diff --git a/frontend/src/app/dialogs/connect-quick-vercel-frame.tsx b/frontend/src/app/dialogs/connect-quick-vercel-frame.tsx index 71ec78dca7..4dd69f5745 100644 --- a/frontend/src/app/dialogs/connect-quick-vercel-frame.tsx +++ b/frontend/src/app/dialogs/connect-quick-vercel-frame.tsx @@ -18,7 +18,6 @@ import { buildServerlessConfig, ConfigurationAccordion, } from "./connect-manual-serverless-frame"; -import { VERCEL_SERVERLESS_MAX_DURATION } from "./connect-vercel-frame"; const { stepper } = ConnectVercelForm; @@ -91,7 +90,8 @@ function FormStepper({ provider, { ...values, - requestLifespan: VERCEL_SERVERLESS_MAX_DURATION - 5, + requestLifespan: + ConnectVercelForm.VERCEL_REQUEST_LIFESPAN, }, { provider: "vercel" }, ); @@ -104,7 +104,7 @@ function FormStepper({ defaultValues={{ runnerName: "default", headers: [], - drainGracePeriod: 0, + drainGracePeriod: ConnectVercelForm.VERCEL_DRAIN_GRACE_PERIOD, plan: "hobby", datacenters: Object.fromEntries( datacenters.map((dc) => [dc.name, true]), diff --git a/frontend/src/app/dialogs/connect-vercel-frame.tsx b/frontend/src/app/dialogs/connect-vercel-frame.tsx index 255e04fe00..97e3b50601 100644 --- a/frontend/src/app/dialogs/connect-vercel-frame.tsx +++ b/frontend/src/app/dialogs/connect-vercel-frame.tsx @@ -19,8 +19,6 @@ import { const { stepper } = ConnectVercelForm; -export const VERCEL_SERVERLESS_MAX_DURATION = 300; - interface CreateProjectFrameContentProps extends DialogContentProps {} export default function CreateProjectFrameContent({ @@ -93,7 +91,8 @@ function FormStepper({ provider, { ...values, - requestLifespan: VERCEL_SERVERLESS_MAX_DURATION - 5, + requestLifespan: + ConnectVercelForm.VERCEL_REQUEST_LIFESPAN, }, { provider: "vercel" }, ); @@ -106,7 +105,7 @@ function FormStepper({ defaultValues={{ plan: "hobby", runnerName: "default", - drainGracePeriod: 0, + drainGracePeriod: ConnectVercelForm.VERCEL_DRAIN_GRACE_PERIOD, headers: [], datacenters: Object.fromEntries( datacenters.map((dc) => [dc.name, true]), diff --git a/frontend/src/app/forms/connect-quick-vercel-form.tsx b/frontend/src/app/forms/connect-quick-vercel-form.tsx index 0617703ab4..3839255509 100644 --- a/frontend/src/app/forms/connect-quick-vercel-form.tsx +++ b/frontend/src/app/forms/connect-quick-vercel-form.tsx @@ -1,10 +1,12 @@ import z from "zod"; import * as ConnectVercelForm from "@/app/forms/connect-vercel-form"; import { defineStepper } from "@/components/ui/stepper"; -import { - configurationSchema, - deploymentSchema, -} from "./connect-manual-serverless-form"; +import { deploymentSchema } from "./connect-manual-serverless-form"; + +export const VERCEL_REQUEST_LIFESPAN = + ConnectVercelForm.VERCEL_REQUEST_LIFESPAN; +export const VERCEL_DRAIN_GRACE_PERIOD = + ConnectVercelForm.VERCEL_DRAIN_GRACE_PERIOD; export const stepper = defineStepper( { @@ -20,7 +22,7 @@ export const stepper = defineStepper( assist: true, next: "Done", schema: z.object({ - ...configurationSchema.omit({ requestLifespan: true }).shape, + ...ConnectVercelForm.configurationSchema.shape, ...deploymentSchema.shape, plan: z.string().min(1, "Please select a Vercel plan"), }), diff --git a/frontend/src/app/forms/connect-vercel-form.tsx b/frontend/src/app/forms/connect-vercel-form.tsx index e5225378bb..60ef1e2ff5 100644 --- a/frontend/src/app/forms/connect-vercel-form.tsx +++ b/frontend/src/app/forms/connect-vercel-form.tsx @@ -24,6 +24,25 @@ import { useRivetDsn, } from "../env-variables"; +export const VERCEL_SERVERLESS_MAX_DURATION = 300; +export const VERCEL_REQUEST_LIFESPAN = VERCEL_SERVERLESS_MAX_DURATION - 5; +export const VERCEL_DRAIN_GRACE_PERIOD = 5; + +export const configurationSchema = + ConnectManualServerlessForm.configurationSchema + .omit({ requestLifespan: true }) + .extend({ + drainGracePeriod: z.coerce + .number() + .min(0) + .max( + VERCEL_REQUEST_LIFESPAN - 1, + `Must be less than ${VERCEL_REQUEST_LIFESPAN}`, + ) + .optional() + .default(VERCEL_DRAIN_GRACE_PERIOD), + }); + export const stepper = defineStepper( { id: "api-route", @@ -55,9 +74,7 @@ export const stepper = defineStepper( next: "Done", schema: z.object({ ...ConnectManualServerlessForm.deploymentSchema.shape, - ...ConnectManualServerlessForm.configurationSchema.omit({ - requestLifespan: true, - }).shape, + ...configurationSchema.shape, plan: z.string().min(1, "Please select a Vercel plan"), }), }, diff --git a/rivetkit-typescript/packages/effect/src/RivetError.ts b/rivetkit-typescript/packages/effect/src/RivetError.ts index d2b2732f02..bad9521dfb 100644 --- a/rivetkit-typescript/packages/effect/src/RivetError.ts +++ b/rivetkit-typescript/packages/effect/src/RivetError.ts @@ -31,6 +31,9 @@ export class Forbidden extends Schema.TaggedErrorClass( get public() { return this.cause.public; } + get rayId() { + return this.cause.rayId; + } get isRetryable(): boolean { return false; } @@ -63,6 +66,9 @@ export class ActorNotFound extends Schema.TaggedErrorClass( get public() { return this.cause.public; } + get rayId() { + return this.cause.rayId; + } get isRetryable(): boolean { return false; } @@ -95,6 +101,9 @@ export class ActorStopping extends Schema.TaggedErrorClass( get public() { return this.cause.public; } + get rayId() { + return this.cause.rayId; + } get isRetryable(): boolean { return true; } @@ -127,6 +136,9 @@ export class ActorRestarting extends Schema.TaggedErrorClass( get public() { return this.cause.public; } + get rayId() { + return this.cause.rayId; + } get isRetryable(): boolean { return true; } @@ -167,6 +179,9 @@ export class ActionNotFound extends Schema.TaggedErrorClass( get public() { return this.cause.public; } + get rayId() { + return this.cause.rayId; + } get isRetryable(): boolean { return false; } @@ -197,6 +212,9 @@ export class ActionTimedOut extends Schema.TaggedErrorClass( get public() { return this.cause.public; } + get rayId() { + return this.cause.rayId; + } get isRetryable(): boolean { return true; } @@ -229,6 +247,9 @@ export class ActionAborted extends Schema.TaggedErrorClass( get public() { return this.cause.public; } + get rayId() { + return this.cause.rayId; + } get isRetryable(): boolean { return false; } @@ -261,6 +282,9 @@ export class ActorOverloaded extends Schema.TaggedErrorClass( get public() { return this.cause.public; } + get rayId() { + return this.cause.rayId; + } get isRetryable(): boolean { return true; } @@ -293,6 +317,9 @@ export class IncomingMessageTooLong extends Schema.TaggedErrorClass( get public() { return this.cause.public; } + get rayId() { + return this.cause.rayId; + } get isRetryable(): boolean { return false; } @@ -389,6 +422,9 @@ export class InvalidRequest extends Schema.TaggedErrorClass( get public() { return this.cause.public; } + get rayId() { + return this.cause.rayId; + } get isRetryable(): boolean { return false; } @@ -421,6 +457,9 @@ export class GuardActorReadyTimeout extends Schema.TaggedErrorClass( get public() { return this.cause.public; } + get rayId() { + return this.cause.rayId; + } get isRetryable(): boolean { return false; } @@ -742,6 +808,9 @@ export class ActionErrorDecodeFailed extends Schema.TaggedErrorClass( get public() { return this.cause.public; } + get rayId() { + return this.cause.rayId; + } get isRetryable(): boolean { return false; } @@ -829,6 +901,11 @@ export class UnknownError extends Schema.TaggedErrorClass( ? this.cause.public : undefined; } + get rayId() { + return this.cause instanceof RivetkitErrors.RivetError + ? this.cause.rayId + : undefined; + } get isRetryable(): boolean { return false; } @@ -987,6 +1064,11 @@ export class RivetError extends Schema.TaggedErrorClass( return "public" in this.reason ? this.reason.public : undefined; } + /** Delegates to the underlying reason's `rayId` if present. */ + get rayId(): string | undefined { + return "rayId" in this.reason ? this.reason.rayId : undefined; + } + /** Delegates to the underlying reason's `isRetryable` getter. */ get isRetryable(): boolean { return this.reason.isRetryable; diff --git a/rivetkit-typescript/packages/rivetkit-napi/src/actor_context.rs b/rivetkit-typescript/packages/rivetkit-napi/src/actor_context.rs index b85649a507..49478854b7 100644 --- a/rivetkit-typescript/packages/rivetkit-napi/src/actor_context.rs +++ b/rivetkit-typescript/packages/rivetkit-napi/src/actor_context.rs @@ -377,6 +377,7 @@ impl ActorContext { message: Some(message), public_: Some(true), status_code: Some(401), + ray_id: None, })) }) } diff --git a/rivetkit-typescript/packages/rivetkit-napi/src/actor_factory.rs b/rivetkit-typescript/packages/rivetkit-napi/src/actor_factory.rs index 40946ea1ed..2462f5aa37 100644 --- a/rivetkit-typescript/packages/rivetkit-napi/src/actor_factory.rs +++ b/rivetkit-typescript/packages/rivetkit-napi/src/actor_factory.rs @@ -288,6 +288,8 @@ struct BridgeRivetErrorPayload { code: String, message: String, metadata: Option, + #[serde(rename = "rayId")] + ray_id: Option, #[serde(rename = "public")] public_: Option, #[serde(rename = "statusCode")] @@ -300,6 +302,7 @@ pub(crate) struct BridgeRivetErrorContext { pub message: Option, pub public_: Option, pub status_code: Option, + pub ray_id: Option, } impl std::fmt::Display for BridgeRivetErrorContext { @@ -1047,6 +1050,7 @@ fn parse_bridge_rivet_error(reason: &str) -> Option { message: Some(message), public_: payload.public_, status_code: payload.status_code, + ray_id: payload.ray_id, })) } diff --git a/rivetkit-typescript/packages/rivetkit-napi/src/lib.rs b/rivetkit-typescript/packages/rivetkit-napi/src/lib.rs index 51e164eab6..d4df87d2bb 100644 --- a/rivetkit-typescript/packages/rivetkit-napi/src/lib.rs +++ b/rivetkit-typescript/packages/rivetkit-napi/src/lib.rs @@ -97,6 +97,7 @@ fn anyhow_to_bridge_rivet_error_payload(error: anyhow::Error) -> serde_json::Val "code": error.code(), "message": error.message(), "metadata": error.metadata(), + "rayId": bridge_context.and_then(|context| context.ray_id.as_deref()), "public": public_, "statusCode": status_code, "actor": error.actor(), diff --git a/rivetkit-typescript/packages/rivetkit-napi/tests/actor_factory.rs b/rivetkit-typescript/packages/rivetkit-napi/tests/actor_factory.rs index 69d2705e38..393afc7092 100644 --- a/rivetkit-typescript/packages/rivetkit-napi/tests/actor_factory.rs +++ b/rivetkit-typescript/packages/rivetkit-napi/tests/actor_factory.rs @@ -68,6 +68,7 @@ mod moved_tests { "code": "same_code", "message": "same message", "metadata": { "count": 1 }, + "rayId": "ray-123", }) ); @@ -78,6 +79,12 @@ mod moved_tests { assert!(transport_error(&first).schema().is_none()); assert_eq!(transport_error(&second).group(), "actor"); assert_eq!(transport_error(&second).code(), "same_code"); + + let payload = crate::anyhow_to_bridge_rivet_error_payload(first); + assert_eq!( + payload.get("rayId").and_then(|value| value.as_str()), + Some("ray-123") + ); } #[test] diff --git a/rivetkit-typescript/packages/rivetkit/src/actor/errors.ts b/rivetkit-typescript/packages/rivetkit/src/actor/errors.ts index 71ef418821..3aff6cdb52 100644 --- a/rivetkit-typescript/packages/rivetkit/src/actor/errors.ts +++ b/rivetkit-typescript/packages/rivetkit/src/actor/errors.ts @@ -12,6 +12,8 @@ export interface RivetErrorOptions extends ErrorOptions { public?: boolean; /** Metadata associated with this error. */ metadata?: unknown; + /** Request identifier used to correlate this error with engine logs. */ + rayId?: string; /** Explicit HTTP status override for router responses. */ statusCode?: number; /** Actor context associated with this error. */ @@ -31,6 +33,7 @@ export interface RivetErrorLike { code: string; message: string; metadata?: unknown; + rayId?: string; public?: boolean; statusCode?: number; actor?: ActorSpecifier; @@ -38,6 +41,14 @@ export interface RivetErrorLike { export interface BridgeRivetErrorPayload extends RivetErrorLike {} +/** + * Shape as it arrives from the native bridge, where an absent ray ID serializes + * to null. Normalized to a BridgeRivetErrorPayload before validation. + */ +interface NativeBridgeErrorPayload extends Omit { + rayId?: string | null; +} + export interface UserErrorOptions extends ErrorOptions { /** * Machine readable code for this error. Useful for catching different types of @@ -59,6 +70,7 @@ function looksLikeRivetErrorOptions( value !== null && ("public" in value || "metadata" in value || + "rayId" in value || "statusCode" in value || "actor" in value || "cause" in value) @@ -94,6 +106,9 @@ export function isRivetErrorLike( typeof error.code === "string" && "message" in error && typeof error.message === "string" && + (!("rayId" in error) || + error.rayId === undefined || + typeof error.rayId === "string") && (!("__type" in error) || isTypedErrorTag(error.__type)) ); } @@ -128,6 +143,7 @@ export class RivetError extends Error { public public: boolean; public metadata?: unknown; + public readonly rayId?: string; public statusCode: number; public actor?: ActorSpecifier; public readonly group: string; @@ -161,6 +177,7 @@ export class RivetError extends Error { this.code = code; this.public = normalized.public ?? false; this.metadata = normalized.metadata; + this.rayId = normalized.rayId ?? undefined; this.statusCode = normalized.statusCode ?? (this.public ? 400 : 500); this.actor = normalized.actor; } @@ -205,6 +222,7 @@ export function toRivetError( public: error.public, statusCode: error.statusCode, metadata: error.metadata, + rayId: error.rayId, actor: error.actor, cause: error instanceof Error ? error.cause : undefined, }); @@ -218,6 +236,7 @@ export function toRivetError( public: fallback?.public, statusCode: fallback?.statusCode, metadata: fallback?.metadata, + rayId: fallback?.rayId, actor: fallback?.actor, cause: error instanceof Error ? error : undefined, }, @@ -230,6 +249,7 @@ export function encodeBridgeRivetError(error: RivetErrorLike): string { code: error.code, message: error.message, metadata: error.metadata, + rayId: error.rayId, public: error.public, statusCode: error.statusCode, actor: error.actor, @@ -244,9 +264,14 @@ export function decodeBridgeRivetErrorPayload( } try { - const payload = JSON.parse( + const raw = JSON.parse( value.slice(BRIDGE_RIVET_ERROR_PREFIX.length), - ) as BridgeRivetErrorPayload; + ) as NativeBridgeErrorPayload; + // Coerce the native bridge's null ray ID to undefined before validation. + const payload: BridgeRivetErrorPayload = { + ...raw, + rayId: raw.rayId ?? undefined, + }; if (!isRivetErrorLike(payload)) { return undefined; } @@ -272,6 +297,7 @@ export function decodeBridgeRivetError(value: string): RivetError | undefined { return new RivetError(payload.group, payload.code, payload.message, { metadata: payload.metadata, + rayId: payload.rayId, public: payload.public, statusCode: payload.statusCode, actor: payload.actor ?? undefined, @@ -303,6 +329,7 @@ export function internalError( public: options?.public, statusCode: options?.statusCode, metadata: options?.metadata, + rayId: options?.rayId, actor: options?.actor, cause: options?.cause, }, diff --git a/rivetkit-typescript/packages/rivetkit/src/client/actor-conn.ts b/rivetkit-typescript/packages/rivetkit/src/client/actor-conn.ts index 545a7dde63..db8edb885f 100644 --- a/rivetkit-typescript/packages/rivetkit/src/client/actor-conn.ts +++ b/rivetkit-typescript/packages/rivetkit/src/client/actor-conn.ts @@ -874,7 +874,7 @@ export class ActorConnRaw { const parsed = parseWebSocketCloseReason(reason); if (parsed) { - const { group, code } = parsed; + const { group, code, rayId } = parsed; if (this.#shouldReconnectForStaleActor(group, code)) { this.#clearResolvedActorIdentity(); @@ -883,7 +883,7 @@ export class ActorConnRaw { group, code, `Connection closed: ${reason}`, - undefined, + { rayId }, ), ); return; @@ -897,6 +897,7 @@ export class ActorConnRaw { this.#actorId, this.#actorResolutionState, this.#driver, + rayId, ); if (schedulingError) { error = schedulingError; @@ -905,7 +906,7 @@ export class ActorConnRaw { group, code, `Connection closed: ${reason}`, - undefined, + { rayId }, ); } } else { @@ -913,7 +914,7 @@ export class ActorConnRaw { group, code, `Connection closed: ${reason}`, - undefined, + { rayId }, ); } diff --git a/rivetkit-typescript/packages/rivetkit/src/client/actor-handle.ts b/rivetkit-typescript/packages/rivetkit/src/client/actor-handle.ts index b911baa39c..37bc4a9a34 100644 --- a/rivetkit-typescript/packages/rivetkit/src/client/actor-handle.ts +++ b/rivetkit-typescript/packages/rivetkit/src/client/actor-handle.ts @@ -166,7 +166,7 @@ export class ActorHandleRaw { }, }).send(name, body, options as any); } catch (err) { - const { group, code, message, metadata, actor } = + const { group, code, message, metadata, rayId, actor } = deconstructError(err, true); if ( @@ -189,6 +189,7 @@ export class ActorHandleRaw { actorId, attempt, maxAttempts, + rayId, ) ) { useQueryTarget = true; @@ -227,6 +228,7 @@ export class ActorHandleRaw { throw new ActorError(group, code, message, { metadata, + rayId, actor, }); } @@ -357,7 +359,7 @@ export class ActorHandleRaw { } return output; } catch (err) { - const { group, code, message, metadata, actor } = + const { group, code, message, metadata, rayId, actor } = deconstructError(err, true); if ( @@ -367,6 +369,7 @@ export class ActorHandleRaw { actorId, attempt, maxAttempts, + rayId, ) ) { useQueryTarget = true; @@ -398,7 +401,7 @@ export class ActorHandleRaw { "actor", "not_found", "The actor does not exist or was destroyed.", - { metadata, actor }, + { metadata, rayId, actor }, ); } @@ -419,7 +422,11 @@ export class ActorHandleRaw { continue; } - throw new ActorError(group, code, message, { metadata, actor }); + throw new ActorError(group, code, message, { + metadata, + rayId, + actor, + }); } } @@ -514,6 +521,7 @@ export class ActorHandleRaw { actorId: string | undefined, attempt: number, maxAttempts: number, + rayId?: string, ): Promise { if ( !isDynamicActorQuery(this.#actorResolutionState) || @@ -530,6 +538,7 @@ export class ActorHandleRaw { actorId, this.#actorResolutionState, this.#driver, + rayId, ); if (schedulingError) { throw schedulingError; @@ -679,7 +688,7 @@ export class ActorHandleRaw { } return response; } catch (err) { - const { group, code, message, metadata, actor } = + const { group, code, message, metadata, rayId, actor } = deconstructError(err, true); if ( @@ -689,6 +698,7 @@ export class ActorHandleRaw { actorId, attempt, maxAttempts, + rayId, ) ) { useQueryTarget = true; @@ -725,7 +735,11 @@ export class ActorHandleRaw { continue; } - throw new ActorError(group, code, message, { metadata, actor }); + throw new ActorError(group, code, message, { + metadata, + rayId, + actor, + }); } } @@ -750,7 +764,7 @@ export class ActorHandleRaw { return null; } - const { group, code } = error; + const { group, code, rayId } = error; if ( await this.#shouldRetrySchedulingError( @@ -759,6 +773,7 @@ export class ActorHandleRaw { actorId, attempt, maxAttempts, + rayId, ) ) { return { @@ -802,6 +817,7 @@ export class ActorHandleRaw { code: string; message: string; metadata?: unknown; + rayId?: string; actor?: ActorSpecifier; } | null> { if (response.ok) { @@ -814,7 +830,7 @@ export class ActorHandleRaw { : this.#encoding; try { - return deserializeWithEncoding< + const error = deserializeWithEncoding< protocol.HttpResponseError, HttpResponseErrorJson, { @@ -822,6 +838,7 @@ export class ActorHandleRaw { code: string; message: string; metadata?: unknown; + rayId?: string; actor?: ActorSpecifier; } >( @@ -854,6 +871,10 @@ export class ActorHandleRaw { : undefined, }), ); + return { + ...error, + rayId: response.headers.get("x-rivet-ray-id") ?? undefined, + }; } catch { return null; } @@ -927,7 +948,9 @@ export class ActorHandleRaw { "actor", "reload_failed", `reload failed with status ${response.status}: ${body}`, - {}, + { + rayId: response.headers.get("x-rivet-ray-id") ?? undefined, + }, ); } } diff --git a/rivetkit-typescript/packages/rivetkit/src/client/actor-query.ts b/rivetkit-typescript/packages/rivetkit/src/client/actor-query.ts index c9ebd90456..5a659a30b3 100644 --- a/rivetkit-typescript/packages/rivetkit/src/client/actor-query.ts +++ b/rivetkit-typescript/packages/rivetkit/src/client/actor-query.ts @@ -67,6 +67,7 @@ export async function checkForSchedulingError( actorId: string, query: ActorQuery, driver: EngineControlClient, + rayId?: string, ): Promise { const name = getActorNameFromQuery(query); @@ -79,7 +80,13 @@ export async function checkForSchedulingError( actorId, error: actor.error, }); - return actorSchedulingError(group, code, actorId, actor.error); + return actorSchedulingError( + group, + code, + actorId, + actor.error, + rayId, + ); } } catch (err) { logger().warn({ diff --git a/rivetkit-typescript/packages/rivetkit/src/client/errors.ts b/rivetkit-typescript/packages/rivetkit/src/client/errors.ts index a72952086f..1303de57c7 100644 --- a/rivetkit-typescript/packages/rivetkit/src/client/errors.ts +++ b/rivetkit-typescript/packages/rivetkit/src/client/errors.ts @@ -51,12 +51,13 @@ export function actorSchedulingError( code: string, actorId: string, details: unknown, + rayId?: string, ): RivetError { return new RivetError( group, code, `Actor failed to start (${actorId}): ${JSON.stringify(details)}`, - { metadata: { actorId, details } }, + { metadata: { actorId, details }, rayId }, ); } diff --git a/rivetkit-typescript/packages/rivetkit/src/client/raw-utils.ts b/rivetkit-typescript/packages/rivetkit/src/client/raw-utils.ts index 23c2fd70be..111e13bce8 100644 --- a/rivetkit-typescript/packages/rivetkit/src/client/raw-utils.ts +++ b/rivetkit-typescript/packages/rivetkit/src/client/raw-utils.ts @@ -99,11 +99,9 @@ export async function rawHttpFetch( return driver.sendRequest(target, proxyRequest, options); } catch (err) { // Standardize to ClientActorError instead of the native backend error - const { group, code, message, metadata, actor } = deconstructError( - err, - true, - ); - throw new ActorError(group, code, message, { metadata, actor }); + const { group, code, message, metadata, rayId, actor } = + deconstructError(err, true); + throw new ActorError(group, code, message, { metadata, rayId, actor }); } } diff --git a/rivetkit-typescript/packages/rivetkit/src/client/utils.ts b/rivetkit-typescript/packages/rivetkit/src/client/utils.ts index 3d1da3fbd2..2e79b18235 100644 --- a/rivetkit-typescript/packages/rivetkit/src/client/utils.ts +++ b/rivetkit-typescript/packages/rivetkit/src/client/utils.ts @@ -215,6 +215,7 @@ export async function sendHttpRequest< code: responseData.code, message: responseData.message, metadata: responseData.metadata, + rayId, actorId: responseData.actor?.actorId, generation: responseData.actor?.generation, actorKey: responseData.actor?.key, @@ -226,6 +227,7 @@ export async function sendHttpRequest< responseData.message, { metadata: responseData.metadata, + rayId: rayId ?? undefined, actor: responseData.actor, }, ); diff --git a/rivetkit-typescript/packages/rivetkit/src/common/utils.ts b/rivetkit-typescript/packages/rivetkit/src/common/utils.ts index edb2494af5..f8ac56b2b9 100644 --- a/rivetkit-typescript/packages/rivetkit/src/common/utils.ts +++ b/rivetkit-typescript/packages/rivetkit/src/common/utils.ts @@ -44,6 +44,7 @@ export interface DeconstructedError { code: string; message: string; metadata?: unknown; + rayId?: string; actor?: errors.ActorSpecifier; } @@ -80,6 +81,7 @@ export function deconstructError( let code: string; let message: string; let metadata: unknown; + let rayId: string | undefined; let actor: errors.ActorSpecifier | undefined; // Structured errors from core or from pre-built `RivetError` instances are canonical. // Only unstructured errors go through the classifier below. @@ -96,6 +98,7 @@ export function deconstructError( code = error.code; message = error.message; metadata = error.metadata; + rayId = error.rayId; actor = error.actor; } else if (errors.ActorError.isActorError(error) && error.public) { // Check if error has statusCode (could be ActorError instance or DeconstructedError) @@ -107,6 +110,7 @@ export function deconstructError( code = error.code; message = getErrorMessage(error); metadata = error.metadata; + rayId = error.rayId; actor = error.actor; } else if (exposeInternalError) { if (errors.ActorError.isActorError(error)) { @@ -116,6 +120,7 @@ export function deconstructError( code = error.code; message = getErrorMessage(error); metadata = error.metadata; + rayId = error.rayId; actor = error.actor; } else { statusCode = 500; @@ -146,6 +151,7 @@ export function deconstructError( code, message, metadata, + rayId, actor, }; } diff --git a/rivetkit-typescript/packages/rivetkit/src/registry/native.ts b/rivetkit-typescript/packages/rivetkit/src/registry/native.ts index 613a9213fc..5aafcd9a57 100644 --- a/rivetkit-typescript/packages/rivetkit/src/registry/native.ts +++ b/rivetkit-typescript/packages/rivetkit/src/registry/native.ts @@ -835,6 +835,7 @@ function encodeNativeCallbackError(error: unknown): Error { group: structuredError.group, code: structuredError.code, metadata: structuredError.metadata, + rayId: structuredError.rayId, }); } diff --git a/rivetkit-typescript/packages/rivetkit/tests/rivet-error.test.ts b/rivetkit-typescript/packages/rivetkit/tests/rivet-error.test.ts index ed964b6be0..9195815e67 100644 --- a/rivetkit-typescript/packages/rivetkit/tests/rivet-error.test.ts +++ b/rivetkit-typescript/packages/rivetkit/tests/rivet-error.test.ts @@ -1,16 +1,20 @@ import { describe, expect, test } from "vitest"; import { + BRIDGE_RIVET_ERROR_PREFIX, decodeBridgeRivetError, encodeBridgeRivetError, RivetError, toRivetError, } from "../src/actor/errors"; +import { createClientWithDriver } from "../src/client/client"; import { deconstructError } from "../src/common/utils"; +import type { EngineControlClient } from "../src/engine-client/driver"; describe("RivetError bridge helpers", () => { test("round trips structured bridge payloads", () => { const error = new RivetError("user", "boom", "typed failure", { metadata: { source: "native" }, + rayId: "ray-123", public: true, actor: { actorId: "actor-123", @@ -27,6 +31,7 @@ describe("RivetError bridge helpers", () => { code: "boom", message: "typed failure", metadata: { source: "native" }, + rayId: "ray-123", actor: { actorId: "actor-123", generation: 7, @@ -35,6 +40,31 @@ describe("RivetError bridge helpers", () => { }); }); + test("decodes bridge payloads whose ray ID serialized to null", () => { + // The native bridge emits `"rayId": null` when no ray ID exists; the + // payload must still decode as a structured error rather than degrade. + const encoded = `${BRIDGE_RIVET_ERROR_PREFIX}${JSON.stringify({ + group: "actor", + code: "boom", + message: "typed failure", + metadata: { source: "native" }, + rayId: null, + public: false, + statusCode: 500, + })}`; + + const decoded = decodeBridgeRivetError(encoded); + + expect(decoded).toBeInstanceOf(RivetError); + expect(decoded).toMatchObject({ + group: "actor", + code: "boom", + message: "typed failure", + metadata: { source: "native" }, + }); + expect(decoded?.rayId).toBeUndefined(); + }); + test("wraps plain errors with actor/internal_error defaults", () => { const error = toRivetError(new Error("plain failure"), { group: "actor", @@ -57,6 +87,7 @@ describe("RivetError bridge helpers", () => { public: true, statusCode: 408, metadata: { source: "core" }, + rayId: "ray-456", }, ); @@ -69,9 +100,23 @@ describe("RivetError bridge helpers", () => { code: "action_timed_out", message: "Action timed out", metadata: { source: "core" }, + rayId: "ray-456", }); }); + test("keeps ray ID separate from application metadata", () => { + const metadata = { rayId: "application-value", source: "user" }; + const error = toRivetError( + new RivetError("user", "boom", "typed failure", { + metadata, + rayId: "engine-value", + }), + ); + + expect(error.rayId).toBe("engine-value"); + expect(error.metadata).toBe(metadata); + }); + test("does not treat plain objects as structured errors", () => { const result = deconstructError({ group: "foo", @@ -103,3 +148,45 @@ describe("RivetError bridge helpers", () => { }); }); }); + +describe("RivetError HTTP diagnostics", () => { + test("exposes response ray ID from an actor action to the user", async () => { + const metadata = { source: "engine" }; + const driver = { + sendRequest: async () => + new Response( + JSON.stringify({ + group: "core", + code: "internal_error", + message: "An internal error occurred", + metadata, + }), + { + status: 500, + headers: { + "content-type": "application/json", + "x-rivet-ray-id": "ray-http-123", + }, + }, + ), + } as EngineControlClient; + const client = createClientWithDriver(driver, { encoding: "json" }); + const handle = client.getForId("test-actor", "actor-123"); + let thrown: unknown; + + try { + await handle.action({ name: "fail", args: [] }); + } catch (error) { + thrown = error; + } + + expect(thrown).toBeInstanceOf(RivetError); + expect(thrown).toMatchObject({ + group: "core", + code: "internal_error", + message: "An internal error occurred", + metadata, + rayId: "ray-http-123", + }); + }); +}); diff --git a/rivetkit-typescript/packages/workflow-engine/src/context.ts b/rivetkit-typescript/packages/workflow-engine/src/context.ts index c4ee278bf7..098a39aef5 100644 --- a/rivetkit-typescript/packages/workflow-engine/src/context.ts +++ b/rivetkit-typescript/packages/workflow-engine/src/context.ts @@ -27,6 +27,7 @@ import { appendName, emptyLocation, isLocationPrefix, + isLoopIterationMarker, locationToKey, registerName, } from "./location.js"; @@ -498,6 +499,81 @@ export class WorkflowContextImpl implements WorkflowContextInterface { this.visitedKeys.add(key); } + /** + * Mark history entries for a loop's already-completed iterations visited so + * an enclosing branch's validateComplete does not reject the iterations this + * run intentionally skips. The resumed iteration and later replay normally. + */ + private markCompletedLoopIterationsVisited( + loopLocation: Location, + resumedIteration: number, + ): void { + if (resumedIteration <= 0) { + return; + } + + const loopSegment = loopLocation[loopLocation.length - 1]; + if (typeof loopSegment !== "number") { + throw new Error("Expected loop location to end with a name index"); + } + + for (const [key, entry] of this.storage.history.entries) { + if (!isLocationPrefix(loopLocation, entry.location)) { + continue; + } + + const iterationSegment = entry.location[loopLocation.length]; + if ( + !iterationSegment || + !isLoopIterationMarker(iterationSegment) || + // Defensive: redundant under the prefix invariant, guards against marking a foreign loop's entries if it ever changes. + iterationSegment.loop !== loopSegment || + iterationSegment.iteration >= resumedIteration + ) { + continue; + } + + this.markVisited(key); + } + } + + /** + * Find the earliest loop iteration still present in history. Bounded loop + * history prunes earlier iterations, so rollback replay must start here to + * avoid aborting on a missing entry. + */ + private firstRetainedLoopIteration( + loopLocation: Location, + maxIteration: number, + ): number { + const loopSegment = loopLocation[loopLocation.length - 1]; + if (typeof loopSegment !== "number") { + throw new Error("Expected loop location to end with a name index"); + } + + let first = maxIteration; + for (const [, entry] of this.storage.history.entries) { + if (!isLocationPrefix(loopLocation, entry.location)) { + continue; + } + + const iterationSegment = entry.location[loopLocation.length]; + if ( + !iterationSegment || + !isLoopIterationMarker(iterationSegment) || + iterationSegment.loop !== loopSegment + ) { + continue; + } + + if (iterationSegment.iteration < first) { + first = iterationSegment.iteration; + } + } + + return first; + } + /** * Check if a name has already been used at the current location in this execution. * Throws HistoryDivergedError if duplicate detected. @@ -1173,7 +1249,59 @@ export class WorkflowContextImpl implements WorkflowContextInterface { ); if (rollbackMode) { - if (loopData.output !== undefined) { + if (metadata.status === "completed") { + // Re-walk the completed loop's retained iterations so nested + // steps register their rollback handlers. Keyed on completion + // status, not output, so a Loop.break(undefined) still + // replays. Bounded loop history may have pruned early + // iterations; start at the first retained one to avoid + // aborting on a missing entry. Only the final loop state is + // persisted, so once the prefix is pruned we can reconstruct + // only the last iteration. This collects; it does not persist. + const firstRetained = this.firstRetainedLoopIteration( + location, + loopData.iteration, + ); + let rollbackIteration: number; + let rollbackState: S; + if (firstRetained === 0) { + rollbackIteration = 0; + rollbackState = config.state as S; + } else { + rollbackIteration = loopData.iteration; + rollbackState = loopData.state as S; + } + for ( + ; + rollbackIteration <= loopData.iteration; + rollbackIteration++ + ) { + const iterationLocation = appendLoopIteration( + this.storage, + location, + config.name, + rollbackIteration, + ); + const branchCtx = this.createBranch(iterationLocation); + const iterationResult = await config.run( + branchCtx, + rollbackState, + ); + branchCtx.validateComplete(); + const result = + iterationResult === undefined + ? ({ + continue: true, + state: rollbackState, + } as LoopResult) + : iterationResult; + if ("break" in result && result.break) { + break; + } + if ("continue" in result && result.continue) { + rollbackState = result.state; + } + } return loopData.output as T; } rollbackSingleIteration = true; @@ -1182,11 +1310,19 @@ export class WorkflowContextImpl implements WorkflowContextInterface { } if (metadata.status === "completed") { + this.markCompletedLoopIterationsVisited( + location, + loopData.iteration + 1, + ); return loopData.output as T; } // Loop already completed if (loopData.output !== undefined) { + this.markCompletedLoopIterationsVisited( + location, + loopData.iteration + 1, + ); return loopData.output as T; } @@ -1194,6 +1330,7 @@ export class WorkflowContextImpl implements WorkflowContextInterface { entry = existing; state = loopData.state as S; iteration = loopData.iteration; + this.markCompletedLoopIterationsVisited(location, iteration); if (rollbackMode) { rollbackOutput = loopData.output as T | undefined; rollbackIterationRan = rollbackOutput !== undefined; @@ -1381,6 +1518,7 @@ export class WorkflowContextImpl implements WorkflowContextInterface { if ( !iterationSegment || typeof iterationSegment === "number" || + // Defensive: redundant under the prefix invariant, guards against pruning a foreign loop's entries if it ever changes. iterationSegment.loop !== loopSegment || iterationSegment.iteration < fromIteration || iterationSegment.iteration >= keepFrom diff --git a/rivetkit-typescript/packages/workflow-engine/tests/loops.test.ts b/rivetkit-typescript/packages/workflow-engine/tests/loops.test.ts index 2a18dbebf2..f46e45e4bf 100644 --- a/rivetkit-typescript/packages/workflow-engine/tests/loops.test.ts +++ b/rivetkit-typescript/packages/workflow-engine/tests/loops.test.ts @@ -232,6 +232,196 @@ for (const mode of modes) { expect(result.output).toEqual(["a", "b", "c"]); }); + it("should resume an inner loop that suspends mid-iteration inside a parent loop", async () => { + // An inner loop suspends mid-iteration, persists its iteration, and + // resumes on the next run without re-visiting iteration 0. The enclosing + // parent loop's validateComplete() must still treat it as visited. + const ticks: number[] = []; + + const workflow = async (ctx: WorkflowContextInterface) => { + return await ctx.loop({ + name: "outer", + state: { done: false }, + run: async (outerCtx, outerState) => { + if (outerState.done) { + return Loop.break(ticks.length); + } + + await outerCtx.loop({ + name: "inner", + state: { count: 0 }, + run: async (innerCtx, innerState) => { + if (innerState.count >= 2) { + return Loop.break(undefined); + } + + const message = await innerCtx.queue.next<{ + n: number; + }>("tick", { names: ["tick"] }); + + await innerCtx.step( + `record-${innerState.count}`, + async () => { + ticks.push(message.body.n); + }, + ); + + return Loop.continue({ + count: innerState.count + 1, + }); + }, + }); + + return Loop.break(ticks.length); + }, + }); + }; + + if (mode === "yield") { + await driver.messageDriver.addMessage({ + id: "tick-1", + name: "tick", + data: { n: 1 }, + sentAt: Date.now(), + }); + + // First run: inner iteration 0 consumes tick-1 and continues; + // inner iteration 1 finds no message and suspends, persisting + // the inner loop at iteration 1. + const firstRun = await runWorkflow( + "wf-1", + workflow, + undefined, + driver, + { mode }, + ).result; + + expect(firstRun.state).toBe("sleeping"); + expect(ticks).toEqual([1]); + + await driver.messageDriver.addMessage({ + id: "tick-2", + name: "tick", + data: { n: 2 }, + sentAt: Date.now(), + }); + + // Second run: replay resumes the inner loop at iteration 1 + // (iteration 0 is never re-visited), then the parent branch is + // validated against the full subtree. + const secondRun = await runWorkflow( + "wf-1", + workflow, + undefined, + driver, + { mode }, + ).result; + + expect(secondRun.state).toBe("completed"); + expect(secondRun.output).toBe(2); + expect(ticks).toEqual([1, 2]); + return; + } + + const handle = runWorkflow("wf-1", workflow, undefined, driver, { + mode, + }); + + await handle.message("tick", { n: 1 }); + await handle.message("tick", { n: 2 }); + + const result = await handle.result; + expect(result.state).toBe("completed"); + expect(result.output).toBe(2); + expect(ticks).toEqual([1, 2]); + }); + + it("should replay a completed inner loop beside a suspending one inside a parent loop", async () => { + // Inner loop A completes while inner loop B suspends in the same parent + // iteration. On resume A returns its saved output early, so its iteration + // entries must still be marked visited or validateComplete() rejects them. + const recordsA: number[] = []; + const recordsB: number[] = []; + + const workflow = async (ctx: WorkflowContextInterface) => { + return await ctx.loop({ + name: "outer", + state: { done: false }, + run: async (outerCtx) => { + await outerCtx.loop({ + name: "innerA", + state: { count: 0 }, + run: async (innerCtx, innerState) => { + if (innerState.count >= 1) { + return Loop.break(undefined); + } + await innerCtx.step( + `record-a-${innerState.count}`, + async () => { + recordsA.push(innerState.count); + }, + ); + return Loop.continue({ count: innerState.count + 1 }); + }, + }); + + await outerCtx.loop({ + name: "innerB", + state: { count: 0 }, + run: async (innerCtx, innerState) => { + if (innerState.count >= 1) { + return Loop.break(undefined); + } + const message = await innerCtx.queue.next<{ n: number }>( + "tickB", + { names: ["tickB"] }, + ); + await innerCtx.step("record-b", async () => { + recordsB.push(message.body.n); + }); + return Loop.continue({ count: innerState.count + 1 }); + }, + }); + + return Loop.break(recordsB.length); + }, + }); + }; + + if (mode === "yield") { + // First run: innerA completes; innerB suspends waiting for tickB. + const firstRun = await runWorkflow("wf-1", workflow, undefined, driver, { + mode, + }).result; + expect(firstRun.state).toBe("sleeping"); + expect(recordsA).toEqual([0]); + + await driver.messageDriver.addMessage({ + id: "tick-b", + name: "tickB", + data: { n: 7 }, + sentAt: Date.now(), + }); + + // Second run: replay returns innerA's saved output early, then + // innerB resumes and the parent branch validates the full subtree. + const secondRun = await runWorkflow("wf-1", workflow, undefined, driver, { + mode, + }).result; + expect(secondRun.state).toBe("completed"); + expect(secondRun.output).toBe(1); + expect(recordsB).toEqual([7]); + return; + } + + const handle = runWorkflow("wf-1", workflow, undefined, driver, { mode }); + await handle.message("tickB", { n: 7 }); + const result = await handle.result; + expect(result.state).toBe("completed"); + expect(result.output).toBe(1); + expect(recordsB).toEqual([7]); + }); + it("should resume nested joins across parent loop iterations", async () => { const processed: string[] = []; diff --git a/rivetkit-typescript/packages/workflow-engine/tests/rollback.test.ts b/rivetkit-typescript/packages/workflow-engine/tests/rollback.test.ts index 5b7314deb9..51e7200e29 100644 --- a/rivetkit-typescript/packages/workflow-engine/tests/rollback.test.ts +++ b/rivetkit-typescript/packages/workflow-engine/tests/rollback.test.ts @@ -191,5 +191,147 @@ for (const mode of modes) { expect(loopEntry.kind.data.iteration).toBe(1); expect(rollbacks).toContain("outside"); }); + + it("should roll back a completed loop's nested steps", async () => { + // A loop completes with a defined output, then a later step fails and + // triggers rollback. The completed loop's nested step rollbacks must + // still be collected, not skipped by the completed-loop early return. + const rollbacks: string[] = []; + + const workflow = async (ctx: WorkflowContextInterface) => { + await ctx.rollbackCheckpoint("checkpoint"); + await ctx.loop({ + name: "loop", + state: 0, + run: async (loopCtx, state) => { + await loopCtx.step({ + name: `step-${state}`, + run: async () => `step-${state}`, + rollback: async () => { + rollbacks.push(`loop-${state}`); + }, + }); + if (state === 0) { + return Loop.continue(1); + } + return Loop.break("loop-done"); + }, + }); + await ctx.step({ + name: "after", + run: async () => "after", + rollback: async () => { + rollbacks.push("after"); + }, + }); + throw new Error("boom"); + }; + + await expect( + runWorkflow("wf-1", workflow, undefined, driver, { mode }) + .result, + ).rejects.toThrow("boom"); + + // Control: the post-loop step's rollback proves rollback ran. + expect(rollbacks).toContain("after"); + // The completed loop's nested step rollbacks must also run. + expect(rollbacks).toContain("loop-0"); + expect(rollbacks).toContain("loop-1"); + }); + + it("should roll back nested steps of a loop that breaks with no output", async () => { + // Loop.break(undefined) completes the loop with output === undefined; + // the rollback replay is keyed on completion status, not output, so + // nested step handlers must still be registered. + const rollbacks: string[] = []; + + const workflow = async (ctx: WorkflowContextInterface) => { + await ctx.rollbackCheckpoint("checkpoint"); + await ctx.loop({ + name: "loop", + state: 0, + run: async (loopCtx, state) => { + await loopCtx.step({ + name: `step-${state}`, + run: async () => `step-${state}`, + rollback: async () => { + rollbacks.push(`loop-${state}`); + }, + }); + if (state === 0) { + return Loop.continue(1); + } + return Loop.break(undefined); + }, + }); + await ctx.step({ + name: "after", + run: async () => "after", + rollback: async () => { + rollbacks.push("after"); + }, + }); + throw new Error("boom"); + }; + + await expect( + runWorkflow("wf-1", workflow, undefined, driver, { mode }) + .result, + ).rejects.toThrow("boom"); + + expect(rollbacks).toContain("after"); + expect(rollbacks).toContain("loop-0"); + expect(rollbacks).toContain("loop-1"); + }); + + it("should roll back retained iterations when loop history is pruned", async () => { + // Bounded loop history prunes early iterations. The rollback replay + // must start at the first retained iteration instead of aborting on + // the pruned one, so retained iterations and post-loop steps still + // register their rollback handlers. + const rollbacks: string[] = []; + + const workflow = async (ctx: WorkflowContextInterface) => { + await ctx.rollbackCheckpoint("checkpoint"); + await ctx.loop({ + name: "loop", + state: 0, + historyPruneInterval: 1, + historySize: 1, + run: async (loopCtx, state) => { + await loopCtx.step({ + name: `step-${state}`, + run: async () => `step-${state}`, + rollback: async () => { + rollbacks.push(`loop-${state}`); + }, + }); + if (state === 0) { + return Loop.continue(1); + } + return Loop.break("done"); + }, + }); + await ctx.step({ + name: "after", + run: async () => "after", + rollback: async () => { + rollbacks.push("after"); + }, + }); + throw new Error("boom"); + }; + + await expect( + runWorkflow("wf-1", workflow, undefined, driver, { mode }) + .result, + ).rejects.toThrow("boom"); + + // The retained tail iteration and post-loop step roll back; the + // pruned iteration 0 cannot (its step output is gone). + expect(rollbacks).toContain("after"); + expect(rollbacks).toContain("loop-1"); + expect(rollbacks).not.toContain("loop-0"); + }); }); }