diff --git a/.github/workflows/publish.yaml b/.github/workflows/publish.yaml index c2b3324074..5913ef76e2 100644 --- a/.github/workflows/publish.yaml +++ b/.github/workflows/publish.yaml @@ -664,6 +664,23 @@ jobs: --latest ${{ needs.context.outputs.latest }} \ --name container-runner + # The Inspector UI lives in frontend/dist, outside the rivetkit-core + # crate, so it must be built and staged into the crate before packaging. + # Otherwise the published crate embeds the empty fallback and serves + # inspector.ui_asset_not_found. Runs for every trigger so branch previews + # gate the crate the same way release does. + - name: Build Inspector UI bundle + env: + SKIP_NAPI_BUILD: "1" + SKIP_WASM_BUILD: "1" + run: npx turbo build:inspector-ui -F @rivetkit/engine-frontend + + - name: Stage Inspector UI into rivetkit-core + run: node rivetkit-rust/packages/rivetkit-core/scripts/stage-inspector-bundle.mjs + + - name: Verify Inspector UI ships in the crate + run: node rivetkit-rust/packages/rivetkit-core/scripts/verify-inspector-bundle.mjs + - name: Dry-run Rust crate publish if: needs.context.outputs.trigger != 'release' # Branch previews publish npm/R2/Docker artifacts only. Unlike npm, diff --git a/Cargo.lock b/Cargo.lock index d1d903ce6c..5ed497b518 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6232,6 +6232,7 @@ dependencies = [ "serde", "serde_json", "tokio", + "tokio-stream", "tokio-util", "tracing", "tracing-subscriber", diff --git a/engine/sdks/rust/envoy-client/src/callbacks.rs b/engine/sdks/rust/envoy-client/src/callbacks.rs index a59e146f83..0d332ba3d8 100644 --- a/engine/sdks/rust/envoy-client/src/callbacks.rs +++ b/engine/sdks/rust/envoy-client/src/callbacks.rs @@ -33,6 +33,13 @@ impl ActorStopHandle { } } + /// Test-only constructor that discards the completion signal. Not for production use. + #[doc(hidden)] + pub fn detached() -> Self { + let (tx, _rx) = oneshot::channel(); + Self::new(tx) + } + pub fn complete(self) -> bool { self.finish(Ok(())) } diff --git a/rivetkit-rust/packages/rivetkit-core/Cargo.toml b/rivetkit-rust/packages/rivetkit-core/Cargo.toml index 4aee22664c..b1c5d139f1 100644 --- a/rivetkit-rust/packages/rivetkit-core/Cargo.toml +++ b/rivetkit-rust/packages/rivetkit-core/Cargo.toml @@ -100,3 +100,7 @@ tracing-subscriber.workspace = true [[test]] name = "integration" path = "tests/integration.rs" + +[[test]] +name = "inspector_bundle" +path = "tests/inspector_bundle.rs" diff --git a/rivetkit-rust/packages/rivetkit-core/build.rs b/rivetkit-rust/packages/rivetkit-core/build.rs index ff085deae5..840d6935a6 100644 --- a/rivetkit-rust/packages/rivetkit-core/build.rs +++ b/rivetkit-rust/packages/rivetkit-core/build.rs @@ -2,14 +2,25 @@ use std::env; use std::fs; use std::path::Path; -use anyhow::Result; +use anyhow::{Context, Result}; use fs_extra::dir; -// Stages frontend/dist/inspector-ui/ into $OUT_DIR/inspector-ui/ and -// frontend/dist/inspector-tab/ into $OUT_DIR/inspector-tab/ so the inspector -// bundle module can embed both via include_dir!. Falls back to an empty -// directory when the frontend has not been built yet, so a missing bundle -// degrades to 404 at runtime instead of a compile error. +// Stages the inspector-ui and inspector-tab bundles into `$OUT_DIR` so the +// inspector bundle module can embed both via include_dir!. +// +// Source resolution order per bundle: +// 1. In-crate `inspector-dist//`, staged by +// `scripts/stage-inspector-bundle.mjs` before `cargo package` / publish. +// This is the only copy that ships to crates.io, because `frontend/dist` +// lives outside the crate and is therefore absent from the `.crate` +// archive. +// 2. Monorepo `../../../frontend/dist//`, produced by the frontend +// build. Used for in-workspace dev builds where staging has not run. +// 3. An empty placeholder, so a not-yet-built frontend degrades to a runtime +// 404 (`inspector.ui_asset_not_found`) instead of a compile error. +// +// A bundle only counts as present when its marker file exists, so an empty +// staging placeholder correctly falls through to the frontend build. fn main() -> Result<()> { let manifest_dir = env::var("CARGO_MANIFEST_DIR")?; let out_dir = env::var("OUT_DIR")?; @@ -18,43 +29,49 @@ fn main() -> Result<()> { // generic source changes, so the script itself must opt in explicitly. println!("cargo:rerun-if-changed=build.rs"); - stage_dir( - &manifest_dir, - &out_dir, - "../../../frontend/dist/inspector-ui", - "inspector-ui", - )?; - stage_dir( - &manifest_dir, - &out_dir, - "../../../frontend/dist/inspector-tab", - "inspector-tab", - )?; + stage_bundle(&manifest_dir, &out_dir, "inspector-ui", "index.html")?; + stage_bundle(&manifest_dir, &out_dir, "inspector-tab", "styles.css")?; Ok(()) } -fn stage_dir(manifest_dir: &str, out_dir: &str, source_rel: &str, staged_name: &str) -> Result<()> { - let source = Path::new(manifest_dir).join(source_rel); - let staged = Path::new(out_dir).join(staged_name); +fn stage_bundle(manifest_dir: &str, out_dir: &str, name: &str, marker: &str) -> Result<()> { + let manifest = Path::new(manifest_dir); + let in_crate = manifest.join("inspector-dist").join(name); + let monorepo = manifest.join("../../../frontend/dist").join(name); - println!("cargo:rerun-if-changed={}", source.display()); + // Rerun when either candidate bundle changes so a rebuild picks up staged + // or freshly built assets. + println!("cargo:rerun-if-changed={}", in_crate.display()); + println!("cargo:rerun-if-changed={}", monorepo.display()); + let source = if in_crate.join(marker).is_file() { + Some(in_crate) + } else if monorepo.join(marker).is_file() { + Some(monorepo) + } else { + None + }; + + let staged = Path::new(out_dir).join(name); if staged.exists() { fs::remove_dir_all(&staged)?; } fs::create_dir_all(&staged)?; - if source.exists() && source.is_dir() { - let mut opts = dir::CopyOptions::new(); - opts.content_only = true; - opts.overwrite = true; - dir::copy(&source, &staged, &opts) - .unwrap_or_else(|e| panic!("failed to copy {source_rel} into OUT_DIR: {e}")); - } else { - // Placeholder so include_dir! has something to embed even when the - // frontend has not been built yet. - fs::write(staged.join(".empty"), b"")?; + match source { + Some(source) => { + let mut opts = dir::CopyOptions::new(); + opts.content_only = true; + opts.overwrite = true; + dir::copy(&source, &staged, &opts) + .with_context(|| format!("failed to copy {} into OUT_DIR", source.display()))?; + } + None => { + // Placeholder so include_dir! has something to embed even when + // neither the staged crate bundle nor the frontend build exists yet. + fs::write(staged.join(".empty"), b"")?; + } } Ok(()) diff --git a/rivetkit-rust/packages/rivetkit-core/inspector-dist/README.md b/rivetkit-rust/packages/rivetkit-core/inspector-dist/README.md new file mode 100644 index 0000000000..76f35bdd59 --- /dev/null +++ b/rivetkit-rust/packages/rivetkit-core/inspector-dist/README.md @@ -0,0 +1,22 @@ +# inspector-dist + +Publish staging area for the Inspector UI bundle. + +`build.rs` embeds two directories into the crate via `include_dir!`: + +- `inspector-dist/inspector-ui/` — the built Inspector UI SPA (served at + `/inspector/ui/`). +- `inspector-dist/inspector-tab/` — the shared custom-tab stylesheet (served at + `/inspector/tab.css`). + +These assets are generated by the frontend build (`frontend/dist/...`), which +lives outside this crate and therefore never ships to crates.io. To make the +Inspector UI a real input to the published crate, `scripts/stage-inspector-bundle.mjs` +copies the built assets into this directory before `cargo package` / publish, +and `scripts/verify-inspector-bundle.mjs` asserts they landed in the `.crate` +archive and are served (not `inspector.ui_asset_not_found`). + +The generated `inspector-ui/` and `inspector-tab/` contents are intentionally +**not** committed: they are staged fresh during publish. For in-workspace dev +builds `build.rs` falls back to `../../../frontend/dist/...`, so nothing needs +to be staged locally. This directory only needs to exist. diff --git a/rivetkit-rust/packages/rivetkit-core/scripts/stage-inspector-bundle.mjs b/rivetkit-rust/packages/rivetkit-core/scripts/stage-inspector-bundle.mjs new file mode 100644 index 0000000000..2a5b3ba2a0 --- /dev/null +++ b/rivetkit-rust/packages/rivetkit-core/scripts/stage-inspector-bundle.mjs @@ -0,0 +1,79 @@ +#!/usr/bin/env node +/** + * Stage the built Inspector UI bundle into the rivetkit-core crate so it ships + * to crates.io. + * + * `frontend/dist/inspector-{ui,tab}` lives outside the crate and is absent from + * the `.crate` archive, so a published crate would embed the empty fallback and + * serve `inspector.ui_asset_not_found`. This copies the built assets into the + * in-crate `inspector-dist/` directory, which `build.rs` embeds via + * `include_dir!`. + * + * Run the frontend build first: + * pnpm turbo build:inspector-ui -F @rivetkit/engine-frontend + * then: + * node scripts/stage-inspector-bundle.mjs + * + * Source maps are stripped: they are debug-only, bloat the crate, and the crate + * has no need to serve `*.map` requests. + */ +import { cpSync, existsSync, mkdirSync, rmSync, statSync } from "node:fs"; +import { dirname, join, resolve } from "node:path"; +import { fileURLToPath } from "node:url"; + +const __dirname = dirname(fileURLToPath(import.meta.url)); +const crateDir = dirname(__dirname); +const repoRoot = resolve(crateDir, "../../.."); +const frontendDist = join(repoRoot, "frontend", "dist"); +const stageRoot = join(crateDir, "inspector-dist"); + +// Each bundle plus the marker file that proves it was actually built. +const BUNDLES = [ + { name: "inspector-ui", marker: "index.html" }, + { name: "inspector-tab", marker: "styles.css" }, +]; + +function fail(message) { + console.error(`stage-inspector-bundle: ${message}`); + process.exit(1); +} + +let strippedMaps = 0; +let strippedBytes = 0; + +for (const { name, marker } of BUNDLES) { + const source = join(frontendDist, name); + const markerPath = join(source, marker); + if (!existsSync(markerPath)) { + fail( + `missing ${name}/${marker} at ${markerPath}. Build it first:\n` + + " pnpm turbo build:inspector-ui -F @rivetkit/engine-frontend", + ); + } + + const dest = join(stageRoot, name); + rmSync(dest, { recursive: true, force: true }); + mkdirSync(dest, { recursive: true }); + + cpSync(source, dest, { + recursive: true, + filter: (src) => { + if (src.endsWith(".map")) { + try { + strippedBytes += statSync(src).size; + } catch {} + strippedMaps += 1; + return false; + } + return true; + }, + }); + + console.log(`staged ${name} -> ${dest}`); +} + +if (strippedMaps > 0) { + console.log( + `stripped ${strippedMaps} source map file(s) (${Math.round(strippedBytes / 1024)} KiB) from the published crate bundle`, + ); +} diff --git a/rivetkit-rust/packages/rivetkit-core/scripts/verify-inspector-bundle.mjs b/rivetkit-rust/packages/rivetkit-core/scripts/verify-inspector-bundle.mjs new file mode 100644 index 0000000000..dd371031b1 --- /dev/null +++ b/rivetkit-rust/packages/rivetkit-core/scripts/verify-inspector-bundle.mjs @@ -0,0 +1,100 @@ +#!/usr/bin/env node +/** + * Release check: prove the Inspector UI actually ships in the rivetkit-core + * `.crate` and is served, rather than degrading to + * `inspector.ui_asset_not_found`. + * + * Two independent assertions: + * 1. Archive inclusion: assert the Inspector UI index.html and the tab + * stylesheet are in the exact file list cargo would pack (`cargo package + * --list`). This catches the root-cause bug (asset missing from the + * published artifact) that the empty-bundle fallback otherwise hides. + * + * `--list` is used instead of building the `.crate`: full packaging strips + * path deps and resolves the exact-pinned sibling crates against + * crates.io, which are not published yet at this point in the ordered + * publish run. `--list` uses the workspace path deps, so it never touches + * the registry while reporting the same file set. + * 2. Runtime serving: build the crate and assert `GET /inspector/ui/` returns + * index.html, not the `ui_asset_not_found` JSON error, via the + * `inspector_bundle` integration test (gated on + * RIVETKIT_ASSERT_INSPECTOR_BUNDLE so ordinary `cargo test` runs without a + * built frontend still pass). + * + * Run `node scripts/stage-inspector-bundle.mjs` first. + */ +import { execFileSync } from "node:child_process"; +import { existsSync, readFileSync, statSync } from "node:fs"; +import { dirname, join, resolve } from "node:path"; +import { fileURLToPath } from "node:url"; + +const __dirname = dirname(fileURLToPath(import.meta.url)); +const crateDir = dirname(__dirname); +const repoRoot = resolve(crateDir, "../../.."); + +function fail(message) { + console.error(`verify-inspector-bundle: ${message}`); + process.exit(1); +} + +function assertHtmlIndex(path, label) { + if (!existsSync(path)) { + fail( + `${label} missing (${path}). The published crate would serve inspector.ui_asset_not_found.`, + ); + } + const html = readFileSync(path, "utf-8").toLowerCase(); + if (!html.includes(" l.trim())); + +for (const required of [ + "inspector-dist/inspector-ui/index.html", + "inspector-dist/inspector-tab/styles.css", +]) { + if (!files.has(required)) { + fail( + `${required} is not in the rivetkit-core package file list. ` + + "The published crate would serve inspector.ui_asset_not_found.", + ); + } +} +console.log("ok: rivetkit-core packages inspector-ui/index.html and inspector-tab/styles.css"); + +// --- 2. Runtime serving ---------------------------------------------------- + +console.log("asserting GET /inspector/ui/ serves index.html..."); +execFileSync( + "cargo", + ["test", "-p", "rivetkit-core", "--test", "inspector_bundle"], + { + cwd: repoRoot, + stdio: "inherit", + env: { ...process.env, RIVETKIT_ASSERT_INSPECTOR_BUNDLE: "1" }, + }, +); + +console.log("verify-inspector-bundle: ok"); diff --git a/rivetkit-rust/packages/rivetkit-core/src/lib.rs b/rivetkit-rust/packages/rivetkit-core/src/lib.rs index 2676b4e9cd..ae6fa5a94e 100644 --- a/rivetkit-rust/packages/rivetkit-core/src/lib.rs +++ b/rivetkit-rust/packages/rivetkit-core/src/lib.rs @@ -153,7 +153,7 @@ pub use actor::task_types::ShutdownKind; pub use actor::work_registry::{ActorWorkKind, ActorWorkPolicy}; pub use error::ActorLifecycle; pub use inspector::{Inspector, InspectorSnapshot}; -pub use registry::{CoreRegistry, EngineSpawnMode, ServeConfig}; +pub use registry::{CoreRegistry, EngineSpawnMode, RuntimeMode, ServeConfig}; pub use rivet_envoy_client::config::{ HTTP_BODY_MAX_CHUNK_SIZE, HTTP_BODY_STREAM_CHANNEL_CAPACITY, HttpRequestBodyStream, ResponseChunk, diff --git a/rivetkit-rust/packages/rivetkit-core/src/registry/envoy_callbacks.rs b/rivetkit-rust/packages/rivetkit-core/src/registry/envoy_callbacks.rs index 8ed87ce9f4..b4c3d45df2 100644 --- a/rivetkit-rust/packages/rivetkit-core/src/registry/envoy_callbacks.rs +++ b/rivetkit-rust/packages/rivetkit-core/src/registry/envoy_callbacks.rs @@ -53,7 +53,7 @@ impl EnvoyCallbacks for RegistryCallbacks { &self, _handle: EnvoyHandle, actor_id: String, - _generation: u32, + generation: u32, reason: protocol::StopActorReason, stop_handle: ActorStopHandle, ) -> EnvoyBoxFuture> { @@ -61,7 +61,9 @@ impl EnvoyCallbacks for RegistryCallbacks { Box::pin(async move { RuntimeSpawner::spawn( async move { - if let Err(error) = dispatcher.stop_actor(&actor_id, reason, stop_handle).await + if let Err(error) = dispatcher + .stop_actor(&actor_id, generation, reason, stop_handle) + .await { tracing::error!( ?error, diff --git a/rivetkit-rust/packages/rivetkit-core/src/registry/mod.rs b/rivetkit-rust/packages/rivetkit-core/src/registry/mod.rs index 24c897a7be..49fa6321d3 100644 --- a/rivetkit-rust/packages/rivetkit-core/src/registry/mod.rs +++ b/rivetkit-rust/packages/rivetkit-core/src/registry/mod.rs @@ -160,10 +160,23 @@ impl ActorInstanceState { #[derive(Clone)] struct PendingStop { + generation: u32, reason: protocol::StopActorReason, stop_handle: ActorStopHandle, } +/// Outcome of attempting to transition the actor instance under an id to stopping +/// for a specific generation. +enum TransitionResult { + /// The current instance matches the requested generation and was moved to stopping. + Transitioned(ActiveActorInstance), + /// An instance exists but for a different generation than the stop targets, so it + /// was left untouched. The stop must not be applied to it. + Stale, + /// No instance is registered for the actor id. + Vacant, +} + pub(crate) struct RegistryDispatcher { pub(crate) factories: HashMap>, actor_instances: SccHashMap, @@ -228,6 +241,26 @@ impl EngineSpawnMode { } } +/// Selects how `Registry::start` runs, mirroring the TypeScript +/// `RIVETKIT_RUNTIME_MODE` env var. `Envoy` holds one long-lived outbound +/// envoy for the process lifetime; `Serverless` runs an HTTP listener that +/// lazily starts and caches an envoy on the first request. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub enum RuntimeMode { + #[default] + Envoy, + Serverless, +} + +impl RuntimeMode { + pub fn from_env() -> Self { + match env::var("RIVETKIT_RUNTIME_MODE") { + Ok(value) if value.eq_ignore_ascii_case("serverless") => Self::Serverless, + _ => Self::Envoy, + } + } +} + #[derive(Clone, Debug, Default)] pub struct ServeConfig { pub version: u32, @@ -773,6 +806,11 @@ impl RegistryDispatcher { .starting_instances .insert_async(request.actor_id.clone(), startup_notify.clone()) .await; + // Test-only seam: lets a test hold a generation in the "starting" window so it + // can deterministically deliver a stop for a previous generation that parks + // under the actor id and gets consumed by this startup. + #[cfg(test)] + test_hooks::wait_for_startup_gate(&request.actor_id).await; let factory = self .factories .get(&request.actor_name) @@ -851,6 +889,19 @@ impl RegistryDispatcher { .remove_async(&request.actor_id.clone()) .await .map(|(_, pending_stop)| pending_stop); + // Only apply a parked stop if it targets the generation we just started. + // A stop parked for a previous generation is stale: complete its handle + // so teardown finalizes, but leave the new generation running. + let pending_stop = match pending_stop { + Some(pending_stop) if pending_stop.generation == request.generation => { + Some(pending_stop) + } + Some(stale_stop) => { + let _ = stale_stop.stop_handle.complete(); + None + } + None => None, + }; if let Some(pending_stop) = pending_stop { let actor_id = request.actor_id.clone(); let stop_reason = map_envoy_stop_reason(&pending_stop.reason); @@ -913,6 +964,16 @@ impl RegistryDispatcher { .starting_instances .remove_async(&request.actor_id.clone()) .await; + // A stop parked while this start was in flight would otherwise leak its + // ActorStopHandle in the map (the map holds the sender alive, hanging the + // caller). Drain and complete it since there is no instance to stop. + if let Some((_, pending_stop)) = self + .pending_stops + .remove_async(&request.actor_id.clone()) + .await + { + let _ = pending_stop.stop_handle.complete(); + } startup_notify.notify_waiters(); Err(error) } @@ -933,11 +994,19 @@ impl RegistryDispatcher { async fn transition_actor_to_stopping( &self, actor_id: &str, + generation: u32, reason: ShutdownKind, - ) -> Option { + ) -> TransitionResult { match self.actor_instances.entry_async(actor_id.to_owned()).await { SccEntry::Occupied(mut entry) => { let instance = entry.get().instance(); + // A stop is scoped to the generation it was issued for. If the currently + // registered instance is a different generation (e.g. a lost previous + // generation whose replacement is already running), the stop is stale and + // must not tear down the newer generation. + if instance.generation != generation { + return TransitionResult::Stale; + } if matches!(entry.get(), ActorInstanceState::Active(_)) { entry.insert(ActorInstanceState::Stopping { instance: instance.clone(), @@ -948,11 +1017,11 @@ impl RegistryDispatcher { .ctx .warn_work_sent_to_stopping_instance("stop_actor"); } - Some(instance) + TransitionResult::Transitioned(instance) } SccEntry::Vacant(entry) => { drop(entry); - None + TransitionResult::Vacant } } } @@ -1035,6 +1104,7 @@ impl RegistryDispatcher { async fn stop_actor( &self, actor_id: &str, + generation: u32, reason: protocol::StopActorReason, stop_handle: ActorStopHandle, ) -> Result<()> { @@ -1044,11 +1114,14 @@ impl RegistryDispatcher { .await .is_some() { + // The target generation is still starting. Park the stop with its generation + // so startup can decide whether it belongs to the generation being started. let _ = self .pending_stops .insert_async( actor_id.to_owned(), PendingStop { + generation, reason, stop_handle, }, @@ -1058,31 +1131,40 @@ impl RegistryDispatcher { } let task_stop_reason = map_envoy_stop_reason(&reason); - let instance = match self - .transition_actor_to_stopping(actor_id, task_stop_reason) + match self + .transition_actor_to_stopping(actor_id, generation, task_stop_reason) .await { - Some(instance) => instance, - None => { + TransitionResult::Transitioned(instance) => { + let result = self + .shutdown_started_instance(actor_id, instance.clone(), reason, stop_handle) + .await; + self.remove_stopping_actor_instance(actor_id, &instance) + .await; + result + } + TransitionResult::Stale => { + // The running instance is a different generation; this stop targets a + // generation that is already gone. Complete the handle so envoy-client + // finalizes teardown cleanly instead of warning about a dropped handle. + let _ = stop_handle.complete(); + Ok(()) + } + TransitionResult::Vacant => { let _ = self .pending_stops .insert_async( actor_id.to_owned(), PendingStop { + generation, reason, stop_handle, }, ) .await; - return Ok(()); + Ok(()) } - }; - let result = self - .shutdown_started_instance(actor_id, instance.clone(), reason, stop_handle) - .await; - self.remove_stopping_actor_instance(actor_id, &instance) - .await; - result + } } async fn shutdown_started_instance( @@ -1269,3 +1351,43 @@ fn map_envoy_stop_reason(reason: &protocol::StopActorReason) -> ShutdownKind { #[cfg(test)] #[path = "../../tests/registry.rs"] pub(crate) mod tests; + +// Test-only hooks used by the moved registry tests to deterministically drive the +// generation-stop race. Gated behind `cfg(test)` so there is no production impact. +#[cfg(test)] +pub(crate) mod test_hooks { + use std::sync::{Arc, OnceLock}; + + use scc::HashMap as SccHashMap; + use tokio::sync::Semaphore; + + static STARTUP_GATES: OnceLock>> = OnceLock::new(); + + fn gates() -> &'static SccHashMap> { + STARTUP_GATES.get_or_init(SccHashMap::new) + } + + /// Arms a gate so `start_actor` pauses for `actor_id` after registering as + /// starting, until `release_startup_gate` is called. + pub(crate) fn arm_startup_gate(actor_id: &str) { + let _ = gates().insert_sync(actor_id.to_owned(), Arc::new(Semaphore::new(0))); + } + + /// Releases a previously armed gate, letting the paused `start_actor` continue. + pub(crate) fn release_startup_gate(actor_id: &str) { + if let Some(sem) = gates().read_sync(actor_id, |_, sem| sem.clone()) { + sem.add_permits(1); + } + } + + /// Called from inside `start_actor`. Blocks only if a gate is armed for the + /// actor. Order-independent: a release before this runs still lets it through. + pub(crate) async fn wait_for_startup_gate(actor_id: &str) { + let sem = gates().read_sync(actor_id, |_, sem| sem.clone()); + if let Some(sem) = sem { + let permit = sem.acquire().await.expect("startup gate semaphore closed"); + permit.forget(); + let _ = gates().remove_sync(actor_id); + } + } +} diff --git a/rivetkit-rust/packages/rivetkit-core/tests/inspector_bundle.rs b/rivetkit-rust/packages/rivetkit-core/tests/inspector_bundle.rs new file mode 100644 index 0000000000..4502ae399f --- /dev/null +++ b/rivetkit-rust/packages/rivetkit-core/tests/inspector_bundle.rs @@ -0,0 +1,37 @@ +//! Release check that the embedded Inspector UI bundle is actually served. +//! +//! Gated on `RIVETKIT_ASSERT_INSPECTOR_BUNDLE`: the bundle is only staged during +//! publish (see `scripts/stage-inspector-bundle.mjs`), so ordinary `cargo test` +//! runs without a built frontend must still pass. `scripts/verify-inspector-bundle.mjs` +//! sets the env var during the release check. + +use rivetkit_core::inspector_bundle::serve_inspector_bundle; + +#[test] +fn embedded_bundle_serves_index_when_required() { + if std::env::var_os("RIVETKIT_ASSERT_INSPECTOR_BUNDLE").is_none() { + eprintln!( + "skipping: set RIVETKIT_ASSERT_INSPECTOR_BUNDLE=1 to require the embedded bundle" + ); + return; + } + + let resp = serve_inspector_bundle("GET", "/inspector/ui/") + .expect("/inspector/ui/ is a public bundle path"); + + let body = resp.body.expect("response has a body"); + let text = String::from_utf8_lossy(&body); + + assert!( + !text.contains("ui_asset_not_found"), + "embedded inspector bundle is empty: GET /inspector/ui/ served ui_asset_not_found", + ); + assert_eq!(resp.status, 200, "expected 200 for index.html, got {}", resp.status); + + let lower = text.to_ascii_lowercase(); + assert!( + lower.contains(") -> bool { + matches!(state, Some(ActorInstanceState::Active(_))) +} + +/// Regression test for the production incident where a `CommandStopActor` addressed +/// to a previous generation was applied to a freshly-started newer generation. +/// +/// A gen-49 `Lost` stop is parked before gen 50 starts. Because stops are now +/// generation-scoped, gen 50's startup must recognize the parked stop as stale and +/// leave gen 50 running instead of stopping itself. +#[tokio::test] +async fn stop_for_previous_generation_does_not_kill_freshly_started_generation() { + use crate::actor::context::ActorContext; + use rivet_envoy_client::config::ActorStopHandle; + use rivet_envoy_client::protocol::StopActorReason; + + let mut factories = HashMap::new(); + factories.insert( + "counter".to_owned(), + Arc::new(ActorFactory::new(ActorConfig::default(), |_start| { + Box::pin(async { Ok(()) }) + })), + ); + let dispatcher = Arc::new(RegistryDispatcher::new(factories, false)); + + let actor_id = "actor-preparked"; + + // gen 49's `Lost` stop arrives while there is no active instance and is parked. + dispatcher + .stop_actor( + actor_id, + 49, + StopActorReason::Lost, + ActorStopHandle::detached(), + ) + .await + .expect("parking a stop with no active instance returns Ok"); + assert!( + dispatcher + .pending_stops + .get_async(&actor_id.to_owned()) + .await + .is_some(), + "gen-49 stop should be parked under the actor id", + ); + + // gen 50 is started to take over the actor on the same runner. + let ctx = ActorContext::new(actor_id, "counter", Vec::new(), "local"); + dispatcher + .start_actor(StartActorRequest { + actor_id: actor_id.to_owned(), + generation: 50, + actor_name: "counter".to_owned(), + input: None, + ctx, + }) + .await + .expect("gen 50 should start successfully"); + + // The stale gen-49 stop is discarded during startup rather than applied to gen 50. + assert!( + dispatcher + .pending_stops + .get_async(&actor_id.to_owned()) + .await + .is_none(), + "stale gen-49 stop should be cleared during gen 50 startup", + ); + + // gen 50 survives: the stop for a previous generation did not kill it. + let is_active = is_actor_active( + dispatcher + .actor_instances + .get_async(&actor_id.to_owned()) + .await + .as_ref() + .map(|entry| entry.get()), + ); + assert!( + is_active, + "gen 50 must stay Active; a stop for gen 49 must not kill gen 50", + ); +} + +/// Regression test for the exact logged ordering: a gen-49 stop arrives *while gen +/// 50 is still starting*. It parks, and gen 50's startup must treat it as stale and +/// keep running. +/// +/// Uses a test-only startup gate (`start_actor` seam) to hold gen 50 in the +/// "starting" window so the stop is delivered mid-startup. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn parked_previous_generation_stop_does_not_kill_starting_generation() { + use std::time::Duration; + + use crate::actor::context::ActorContext; + use rivet_envoy_client::config::ActorStopHandle; + use rivet_envoy_client::protocol::StopActorReason; + + let mut factories = HashMap::new(); + factories.insert( + "counter".to_owned(), + Arc::new(ActorFactory::new(ActorConfig::default(), |_start| { + Box::pin(async { Ok(()) }) + })), + ); + let dispatcher = Arc::new(RegistryDispatcher::new(factories, false)); + + let actor_id = "actor-gated"; + + // Hold gen 50 in the starting window. + test_hooks::arm_startup_gate(actor_id); + let start_dispatcher = dispatcher.clone(); + let ctx = ActorContext::new(actor_id, "counter", Vec::new(), "local"); + let start_task = tokio::spawn(async move { + start_dispatcher + .start_actor(StartActorRequest { + actor_id: actor_id.to_owned(), + generation: 50, + actor_name: "counter".to_owned(), + input: None, + ctx, + }) + .await + }); + + // Wait until gen 50 has registered as starting (paused at the gate). + tokio::time::timeout(Duration::from_secs(5), async { + loop { + if dispatcher + .starting_instances + .get_async(&actor_id.to_owned()) + .await + .is_some() + { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("gen 50 should register as starting"); + + // gen 49's `Lost` stop arrives while gen 50 is starting: it parks under the actor id. + dispatcher + .stop_actor( + actor_id, + 49, + StopActorReason::Lost, + ActorStopHandle::detached(), + ) + .await + .expect("stop parks while gen 50 is starting"); + assert!( + dispatcher + .pending_stops + .get_async(&actor_id.to_owned()) + .await + .is_some(), + "gen-49 stop should be parked while gen 50 is starting", + ); + + // Release gen 50's startup; it must recognize the parked gen-49 stop as stale. + test_hooks::release_startup_gate(actor_id); + start_task + .await + .expect("start task joins") + .expect("gen 50 should start successfully"); + + assert!( + dispatcher + .pending_stops + .get_async(&actor_id.to_owned()) + .await + .is_none(), + "stale gen-49 stop should be cleared during gen 50 startup", + ); + let is_active = is_actor_active( + dispatcher + .actor_instances + .get_async(&actor_id.to_owned()) + .await + .as_ref() + .map(|entry| entry.get()), + ); + assert!( + is_active, + "gen 50 must stay Active; a gen-49 stop parked during its startup must not kill it", + ); +} + +/// Guards against over-correction: a stop for the *current* generation must still +/// stop it. Prevents the generation-scoping fix from turning legitimate stops into +/// no-ops. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn stop_for_current_generation_stops_it() { + use crate::actor::context::ActorContext; + use rivet_envoy_client::config::ActorStopHandle; + use rivet_envoy_client::protocol::StopActorReason; + + let mut factories = HashMap::new(); + factories.insert( + "counter".to_owned(), + Arc::new(ActorFactory::new(ActorConfig::default(), |_start| { + Box::pin(async { Ok(()) }) + })), + ); + let dispatcher = Arc::new(RegistryDispatcher::new(factories, false)); + + let actor_id = "actor-current"; + + let ctx = ActorContext::new(actor_id, "counter", Vec::new(), "local"); + dispatcher + .start_actor(StartActorRequest { + actor_id: actor_id.to_owned(), + generation: 50, + actor_name: "counter".to_owned(), + input: None, + ctx, + }) + .await + .expect("gen 50 should start successfully"); + assert!( + is_actor_active( + dispatcher + .actor_instances + .get_async(&actor_id.to_owned()) + .await + .as_ref() + .map(|entry| entry.get()), + ), + "gen 50 should be Active after starting", + ); + + // A stop for the matching generation (50) must stop the running instance. + dispatcher + .stop_actor( + actor_id, + 50, + StopActorReason::Lost, + ActorStopHandle::detached(), + ) + .await + .expect("stopping the current generation should succeed"); + + assert!( + !is_actor_active( + dispatcher + .actor_instances + .get_async(&actor_id.to_owned()) + .await + .as_ref() + .map(|entry| entry.get()), + ), + "a stop for the current generation must stop it", + ); +} diff --git a/rivetkit-rust/packages/rivetkit/Cargo.toml b/rivetkit-rust/packages/rivetkit/Cargo.toml index 10451df6d6..d08738259e 100644 --- a/rivetkit-rust/packages/rivetkit/Cargo.toml +++ b/rivetkit-rust/packages/rivetkit/Cargo.toml @@ -17,6 +17,8 @@ sqlite-local = ["rivetkit-core/sqlite-local"] [dependencies] anyhow.workspace = true async-trait.workspace = true +axum.workspace = true +bytes.workspace = true ciborium.workspace = true futures.workspace = true http.workspace = true @@ -27,6 +29,7 @@ parking_lot.workspace = true serde.workspace = true serde_json.workspace = true tokio.workspace = true +tokio-stream.workspace = true tokio-util.workspace = true tracing.workspace = true diff --git a/rivetkit-rust/packages/rivetkit/src/lib.rs b/rivetkit-rust/packages/rivetkit/src/lib.rs index f53050902c..18e126854f 100644 --- a/rivetkit-rust/packages/rivetkit/src/lib.rs +++ b/rivetkit-rust/packages/rivetkit/src/lib.rs @@ -8,6 +8,7 @@ pub mod persist; pub mod prelude; pub mod queue; pub mod registry; +mod serverless_listener; pub mod sqlite; pub mod start; pub mod test; @@ -46,7 +47,8 @@ pub use rivetkit_core::{ CompletableQueueMessage, ConnHandle, ConnId, EngineSpawnMode, EnqueueAndWaitOpts, HTTP_BODY_STREAM_CHANNEL_CAPACITY, KeepAwakeRegion, ListOpts, QueueMessage as CoreQueueMessage, QueueNextBatchOpts, QueueNextOpts, QueueTryNextBatchOpts, QueueTryNextOpts, QueueWaitOpts, - Request, RequestSaveOpts, Response, ResponseChunk, SaveStateOpts, SerializeStateReason, + Request, RequestSaveOpts, Response, ResponseChunk, RuntimeMode, SaveStateOpts, + SerializeStateReason, ServeConfig, SqliteBatchStatement, SqliteDb, SqliteTransaction, StateDelta, StreamingResponse, WebSocket, WsMessage, sqlite::{BindParam, ColumnValue, ExecResult, QueryResult}, diff --git a/rivetkit-rust/packages/rivetkit/src/registry.rs b/rivetkit-rust/packages/rivetkit/src/registry.rs index 8159a17aad..070c29dcdf 100644 --- a/rivetkit-rust/packages/rivetkit/src/registry.rs +++ b/rivetkit-rust/packages/rivetkit/src/registry.rs @@ -6,12 +6,14 @@ use rivetkit_core::metrics_endpoint::{RenderedMetrics, render_prometheus_metrics use rivetkit_core::registry::CoreEnvoyHandle; use rivetkit_core::serverless::CoreServerlessRuntime; use rivetkit_core::{ - ActorConfig, ActorFactory as CoreActorFactory, ActorStart, CoreRegistry, ServeConfig, + ActorConfig, ActorFactory as CoreActorFactory, ActorStart, CoreRegistry, RuntimeMode, + ServeConfig, }; use tokio_util::sync::CancellationToken; use crate::{ actor::Actor, + serverless_listener, start::{Start, run_actor, wrap_start}, }; @@ -130,7 +132,20 @@ impl Registry { } /// [`start`](Self::start) with an explicit [`ServeConfig`]. + /// + /// Selects the run mode from `RIVETKIT_RUNTIME_MODE` (see [`RuntimeMode`]), + /// mirroring the TypeScript `registry.start()`. `Envoy` (default) holds one + /// long-lived outbound envoy; `Serverless` runs an HTTP listener that lazily + /// starts and caches an envoy on the first request. pub async fn start_with_config(self, config: ServeConfig) -> Result<()> { + match RuntimeMode::from_env() { + RuntimeMode::Envoy => self.start_envoy(config).await, + RuntimeMode::Serverless => self.start_serverless(config).await, + } + } + + /// Persistent-envoy `start`: serves until SIGINT/SIGTERM, then drains. + async fn start_envoy(self, config: ServeConfig) -> Result<()> { let shutdown = CancellationToken::new(); let mut serve = tokio::spawn({ let shutdown = shutdown.clone(); @@ -146,6 +161,26 @@ impl Registry { shutdown.cancel(); serve.await? } + + /// Serverless `start`: runs the HTTP listener until SIGINT/SIGTERM, then + /// drains the cached envoy. + async fn start_serverless(self, config: ServeConfig) -> Result<()> { + let runtime = self.into_serverless_runtime(config).await?; + let shutdown = CancellationToken::new(); + let mut serve = tokio::spawn({ + let shutdown = shutdown.clone(); + async move { serverless_listener::serve(runtime, shutdown).await } + }); + + tokio::select! { + // Surface an early listener failure instead of waiting for a signal. + result = &mut serve => return result?, + _ = shutdown_signal() => {} + } + + shutdown.cancel(); + serve.await? + } } /// Resolves when the process receives SIGINT or, on Unix, SIGTERM. diff --git a/rivetkit-rust/packages/rivetkit/src/serverless_listener.rs b/rivetkit-rust/packages/rivetkit/src/serverless_listener.rs new file mode 100644 index 0000000000..31762b1af1 --- /dev/null +++ b/rivetkit-rust/packages/rivetkit/src/serverless_listener.rs @@ -0,0 +1,155 @@ +//! HTTP listener for `RIVETKIT_RUNTIME_MODE=serverless`. +//! +//! Mirrors the TypeScript `registry.listen()` serverless path: binds an HTTP +//! server and forwards every request to [`CoreServerlessRuntime::handle_request`], +//! which lazily starts and caches an envoy on the first request. The server +//! shuts down gracefully when the shutdown token is cancelled, then drains the +//! cached envoy via [`CoreServerlessRuntime::shutdown`]. + +use std::collections::HashMap; +use std::io; +use std::net::{Ipv4Addr, SocketAddr}; + +use anyhow::{Context, Result}; +use axum::body::{Body, to_bytes}; +use axum::extract::{Request, State}; +use axum::response::Response; +use axum::Router; +use bytes::Bytes; +use futures::StreamExt; +use http::StatusCode; +use rivetkit_core::serverless::{ + CoreServerlessRuntime, ServerlessRequest, ServerlessResponse, +}; +use tokio_stream::wrappers::UnboundedReceiverStream; +use tokio_util::sync::CancellationToken; + +/// Default listen port, matching the TypeScript `RIVET_PORT` fallback. +const DEFAULT_PORT: u16 = 3000; + +/// Runs the serverless HTTP listener until `shutdown` is cancelled, then drains +/// the runtime. Binds `0.0.0.0:$RIVET_PORT` (default 3000). +pub async fn serve(runtime: CoreServerlessRuntime, shutdown: CancellationToken) -> Result<()> { + let port = listen_port(); + let addr = SocketAddr::from((Ipv4Addr::UNSPECIFIED, port)); + + let app = Router::new().fallback(handle).with_state(runtime.clone()); + + let listener = tokio::net::TcpListener::bind(addr) + .await + .with_context(|| format!("bind serverless listener on {addr}"))?; + tracing::info!(%addr, "rivetkit serverless listener started"); + + let serve_result = axum::serve(listener, app) + .with_graceful_shutdown(async move { shutdown.cancelled().await }) + .await + .context("serverless listener failed"); + + // Drain the cached envoy regardless of how the server loop ended so an + // in-flight actor's `Stopped` reaches the engine before shutdown. + runtime.shutdown().await; + + serve_result +} + +/// Resolves the listen port from `RIVET_PORT`, falling back to [`DEFAULT_PORT`]. +fn listen_port() -> u16 { + std::env::var("RIVET_PORT") + .ok() + .and_then(|value| value.parse().ok()) + .unwrap_or(DEFAULT_PORT) +} + +/// Forwards a single HTTP request to the serverless runtime and streams the +/// response back. +async fn handle(State(runtime): State, request: Request) -> Response { + let (parts, body) = request.into_parts(); + + let headers: HashMap = parts + .headers + .iter() + .filter_map(|(name, value)| { + value + .to_str() + .ok() + .map(|value| (name.as_str().to_ascii_lowercase(), value.to_owned())) + }) + .collect(); + + // `route_path` parses an absolute URL, so reconstruct one from the request + // target and the Host header (origin-form requests carry no authority). + let host = headers + .get("host") + .map(String::as_str) + .unwrap_or("localhost"); + let path_and_query = parts + .uri + .path_and_query() + .map(|pq| pq.as_str()) + .unwrap_or("/"); + let url = format!("http://{host}{path_and_query}"); + + let body = match to_bytes(body, runtime.max_request_body_bytes()).await { + Ok(body) => body, + Err(_) => return into_response(runtime.incoming_too_long_response()), + }; + + // Cancelled when the response body is dropped (e.g. client disconnect), + // which tears down this in-flight request without touching the cached envoy. + let cancel_token = CancellationToken::new(); + + let response = runtime + .handle_request(ServerlessRequest { + method: parts.method.as_str().to_owned(), + url, + headers, + body: body.to_vec(), + cancel_token: cancel_token.clone(), + }) + .await; + + into_response_with_guard(response, cancel_token.drop_guard()) +} + +fn into_response(response: ServerlessResponse) -> Response { + build_response(response, None) +} + +fn into_response_with_guard( + response: ServerlessResponse, + guard: tokio_util::sync::DropGuard, +) -> Response { + build_response(response, Some(guard)) +} + +/// Builds an axum response that streams the runtime's response chunks. `guard`, +/// when present, is held for the lifetime of the stream so dropping the body +/// cancels the in-flight request. +fn build_response(response: ServerlessResponse, guard: Option) -> Response { + let ServerlessResponse { + status, + headers, + body, + } = response; + + let stream = UnboundedReceiverStream::new(body).map(move |item| { + // Touch the guard so it lives as long as the stream is polled. + let _ = &guard; + item.map(Bytes::from) + .map_err(|error| io::Error::new(io::ErrorKind::Other, error.message)) + }); + + let mut builder = Response::builder() + .status(StatusCode::from_u16(status).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR)); + for (name, value) in headers { + builder = builder.header(name, value); + } + + builder.body(Body::from_stream(stream)).unwrap_or_else(|error| { + tracing::error!(?error, "failed to build serverless response"); + Response::builder() + .status(StatusCode::INTERNAL_SERVER_ERROR) + .body(Body::empty()) + .expect("static error response is valid") + }) +} diff --git a/rivetkit-typescript/packages/rivetkit/src/client/actor-handle.ts b/rivetkit-typescript/packages/rivetkit/src/client/actor-handle.ts index 5e84d7ae39..e41aa6366d 100644 --- a/rivetkit-typescript/packages/rivetkit/src/client/actor-handle.ts +++ b/rivetkit-typescript/packages/rivetkit/src/client/actor-handle.ts @@ -78,6 +78,7 @@ export class ActorHandleRaw { #gatewayOptions: ActorGatewayOptions; #params: unknown; #getParams?: () => Promise; + #signal?: AbortSignal; #resolvedActorId?: string; #resolvingActorId?: Promise; #queueSendMutex = new AsyncMutex(); @@ -97,6 +98,7 @@ export class ActorHandleRaw { encoding: Encoding, actorResolutionState: ActorResolutionState, gatewayOptions: ActorGatewayOptions = {}, + signal?: AbortSignal, ) { this.#client = client; this.#driver = driver; @@ -105,6 +107,7 @@ export class ActorHandleRaw { this.#gatewayOptions = gatewayOptions; this.#params = params; this.#getParams = getParams; + this.#signal = signal; } async #resolveConnectionParams(): Promise { @@ -268,12 +271,17 @@ export class ActorHandleRaw { `Invalid action call: expected an options object { name, args }, got ${typeof opts}. Use handle.actionName(...args) for the shorthand API.`, ); } - const run = async () => (await this.#sendActionNow(opts)) as Response; + // Fall back to the handle-level signal from `get`/`getOrCreate`/etc. + // when no per-call signal is provided. + const signal = opts.signal ?? this.#signal; + const optsWithSignal = { ...opts, signal }; + const run = async () => + (await this.#sendActionNow(optsWithSignal)) as Response; if (opts.name === "destroy") { return await run(); } - return await retryOnLifecycleBoundary(run, { signal: opts.signal }); + return await retryOnLifecycleBoundary(run, { signal }); } async #sendActionNow( diff --git a/rivetkit-typescript/packages/rivetkit/src/client/client.ts b/rivetkit-typescript/packages/rivetkit/src/client/client.ts index 33262b0096..439403871e 100644 --- a/rivetkit-typescript/packages/rivetkit/src/client/client.ts +++ b/rivetkit-typescript/packages/rivetkit/src/client/client.ts @@ -236,6 +236,7 @@ export class ClientRaw { opts?.params, opts?.getParams, actorQuery, + opts?.signal, ); return createActorProxy(handle) as ActorHandle; } @@ -275,6 +276,7 @@ export class ClientRaw { opts?.params, opts?.getParams, actorQuery, + opts?.signal, ); return createActorProxy(handle) as ActorHandle; } @@ -318,6 +320,7 @@ export class ClientRaw { opts?.params, opts?.getParams, actorQuery, + opts?.signal, ); return createActorProxy(handle) as ActorHandle; } @@ -377,6 +380,7 @@ export class ClientRaw { opts?.params, opts?.getParams, getForIdQuery, + opts?.signal, ); const proxy = createActorProxy(handle) as ActorHandle; @@ -388,6 +392,7 @@ export class ClientRaw { params: unknown, getParams: (() => Promise) | undefined, actorQuery: ActorQuery, + signal?: AbortSignal, ): ActorHandleRaw { return new ActorHandleRaw( this, @@ -397,6 +402,7 @@ export class ClientRaw { this.#encodingKind, actorQuery, this.#gatewayOptions, + signal, ); } diff --git a/rivetkit-typescript/packages/rivetkit/tests/driver/actor-handle.test.ts b/rivetkit-typescript/packages/rivetkit/tests/driver/actor-handle.test.ts index e450e07e91..1c9ffa39ec 100644 --- a/rivetkit-typescript/packages/rivetkit/tests/driver/actor-handle.test.ts +++ b/rivetkit-typescript/packages/rivetkit/tests/driver/actor-handle.test.ts @@ -324,5 +324,65 @@ describeDriverMatrix("Actor Handle", (driverTestConfig) => { ).toBe(2); }); }); + + describe("Abort Signal", () => { + test("get(): already-aborted signal rejects the action before it runs", async (c) => { + const { client } = await setupDriverTest(c, driverTestConfig); + + const key = ["abort-get-preaborted", crypto.randomUUID()]; + // Actor exists, so the action would otherwise succeed. + await client.counter.create(key); + + const controller = new AbortController(); + const reason = new Error("aborted before request"); + controller.abort(reason); + + const handle = client.counter.get(key, { + signal: controller.signal, + }); + + // The handle-level signal must short-circuit the action with + // the exact abort reason. + await expect(handle.increment(1)).rejects.toBe(reason); + + // The aborted action never reached the actor. + const count = await client.counter.get(key).getCount(); + expect(count).toBe(0); + }); + + test("getOrCreate(): already-aborted signal rejects the action before it runs", async (c) => { + const { client } = await setupDriverTest(c, driverTestConfig); + + const controller = new AbortController(); + const reason = new Error("aborted before request"); + controller.abort(reason); + + const handle = client.counter.getOrCreate( + ["abort-get-or-create-preaborted", crypto.randomUUID()], + { signal: controller.signal }, + ); + + await expect(handle.increment(1)).rejects.toBe(reason); + }); + + test("get(): mid-flight abort rejects the in-flight action", async (c) => { + const { client } = await setupDriverTest(c, driverTestConfig); + + const key = ["abort-get-midflight", crypto.randomUUID()]; + const controller = new AbortController(); + + const handle = client.concurrentActionActor.getOrCreate(key, { + signal: controller.signal, + }); + + const promise = handle.runWithDelay("slow", 10_000); + + // Give the request time to reach the actor before aborting. + await new Promise((resolve) => setTimeout(resolve, 150)); + controller.abort(); + + await expect(promise).rejects.toThrow(); + }); + }); }); });