Skip to content

feat(rivetkit): streamline workflow host APIs - #5578

Open
NathanFlurry wants to merge 3 commits into
mainfrom
feat/workflows-public-host-apis
Open

feat(rivetkit): streamline workflow host APIs#5578
NathanFlurry wants to merge 3 commits into
mainfrom
feat/workflows-public-host-apis

Conversation

@NathanFlurry

@NathanFlurry NathanFlurry commented Aug 19, 2026

Copy link
Copy Markdown
Member
  • Make embedded c.db available by default and add experimental transactions that atomically include actor state.
  • Add the experimental Inspector, queue, and wake capabilities required by extracted workflows across native and wasm runtimes.
  • Preserve the existing workflow SQLite format and add upgrade and rollback compatibility coverage.
  • Remove the workflow-specific storage API in favor of the standard embedded database.

@railway-app

railway-app Bot commented Aug 19, 2026

Copy link
Copy Markdown

🚅 Deployed to the actors-pr-5578 environment in rivet-frontend

Service Status Web Updated (UTC)
kitchen-sink 😴 Sleeping (View Logs) Web Aug 20, 2026 at 6:53 am
website ❌ Build Failed (View Logs) Web Aug 20, 2026 at 6:43 am
frontend-cloud 😴 Sleeping (View Logs) Web Aug 20, 2026 at 6:43 am
frontend-inspector 😴 Sleeping (View Logs) Web Aug 20, 2026 at 6:41 am
ladle ✅ Success (View Logs) Web Aug 19, 2026 at 5:53 pm
mcp-hub ✅ Success (View Logs) Web Aug 19, 2026 at 5:52 pm

@claude

claude Bot commented Aug 19, 2026

Copy link
Copy Markdown
Contributor

Review: feat(workflows): expose public workflow host APIs

Big PR (60 files, ~4.3k additions) that exposes workflow host storage/queue/alarm/run-handler/inspector APIs across native and wasm runtimes, plus a new run_wake_at durable-wake mechanism and a TS RunHandlerCoordinator. Test coverage is genuinely strong — lifecycle, replay-concurrency, query-efficiency, and runtime-parity tests all landed alongside the feature, including explicit regression tests for several of the race conditions this kind of change usually introduces (stale alarm-ack races, exclusive-replay generation tracking, schema backward-compat).

That said, I found two correctness issues in the parts of this PR the tests don't cover (wasm-only concurrency, and a specific failure-path interleaving), plus a few smaller cleanup items.

Correctness

1. wasm RunWake dispatch can invoke a second concurrent run() while the first is still executing (rivetkit-typescript/packages/rivetkit-wasm/src/lib.rs:804, start_run_handler at line 586)

ActorEvent::RunWake => start_run_handler(callbacks, ctx),

start_run_handler just calls ctx.inner.begin_run_handler() (an activity counter, not a mutex — see sleep.rs:200) and spawn_locals a fresh invocation of the user's run callback. There's no check for an already-active handler and no cancellation of one, unlike the NAPI path (rivetkit-typescript/packages/rivetkit-napi/src/napi_actor_events.rs:337-347), which aborts the previous JoinHandle via attach_run_restart before respawning.

If a workflow's run() is still executing a step when its persisted run_wake_at deadline also elapses, dispatch_event fires RunWake fire-and-forget (no await), so a second concurrent run() invocation starts for the same actor while the first is still live — two workflow loops racing the same actor state/history. This directly contradicts the single-writer actor invariant in the root CLAUDE.md ("a Rivet Actor is the single writer for both KV and SQLite") and the wasm/NAPI parity requirement in rivetkit-typescript/CLAUDE.md ("Keep NAPI and wasm serverless registry lifecycle semantics aligned"). It's not caught by runtime-parity.test.ts because that suite exercises the TS RunHandlerCoordinator against fakes, not the actual Rust-level abort/no-abort difference between the two bindings.

2. fire_due_alarms's failure-path restore can clobber a newer run_wake_at with the stale value it just consumed (rivetkit-rust/packages/rivetkit-core/src/actor/task.rs:1414-1449)

let due_run_wake = self.ctx.consume_due_run_wake().await?; // clears run_wake_at, returns T1
if let Err(error) = self.ctx.drain_overdue_scheduled_events().await {
    if ... && let Some(wake_at) = due_run_wake
        && let Err(restore_error) = self.ctx.set_run_wake_at(Some(wake_at)).await
    { ... }
    return Err(error);
}
...
if let Some(wake_at) = due_run_wake
    && let Err(error) = self.ctx.try_send_actor_event(ActorEvent::RunWake, "run_wake")
{
    self.ctx.set_run_wake_at(Some(wake_at)).await?; // unconditional restore of T1
    return Err(error).context("dispatch due run wake");
}

consume_due_run_wake() pops the deadline (releasing schedule_mutation_lock) before either of these failure branches runs. If, concurrently, the actor's live run() handler calls ctx.run.setWakeAt(T2) for a legitimately later deadline in that window, and then drain_overdue_scheduled_events/try_send_actor_event fails, the restore unconditionally writes back the stale T1, silently clobbering T2. set_run_wake_at has no compare-and-swap against the current persisted value, so there's no way to detect this happened. Worth either a CAS-style restore (only restore if the current value is still None/unset) or a comment explaining why this can't race in practice, if it truly can't.

Convention

3. New test uses _ => {} fallthrough on an ActorEvent match (rivetkit-rust/packages/rivetkit-core/tests/task.rs:2795-2800, in startup_delivers_a_persisted_run_wake_that_became_overdue)
CLAUDE.md is explicit: "Never use a _ => fall-through arm when matching on a Rust enum... Enumerate every variant explicitly so adding a new variant later is a compile error instead of a silent behavior change." This new test matches RunWake/BeginSleep/FinalizeSleep/Destroy then falls through the rest with _ => {}. Easy fix, just enumerate the remaining variants.

Cleanup (dead code left behind by the refactor)

  • isRunHandlerActive() / nativeRunHandlerActiveByActorId in rivetkit-typescript/packages/rivetkit/src/registry/native.ts (~3541, ~3794) has no remaining call sites — its only consumer (the guard in workflow/mod.ts's replayFromStep) was replaced by RunHandlerCoordinator.withInactive(...) in this PR. The map is still populated/cleared but nothing reads it anymore.
  • NAPI WorkflowStorage::flush_with_state (rivetkit-typescript/packages/rivetkit-napi/src/workflow_storage.rs:107) is unreachable from TS: storage.ts's flushWithState (native.ts:2825) calls saveStateAndWorkflowBatch instead, confirmed by tests/workflow-host-api.test.ts asserting exactly that. Either wire it up or drop the NAPI binding.

Design question (not blocking, but worth a second look)

run_wake_at is implemented as an entirely separate one-row scheduling/redelivery mechanism — new meta key, new versioned type (RunWakeAt in versioned.rs is a byte-for-byte structural clone of the existing LastPushedAlarm), a min_deadline() combinator merging it with the existing alarm path, and duplicated consume/restore-on-failure logic in task.rs (see finding #2) — rather than a sentinel row in the existing _rivet_schedule_events table, which already provides due-alarm dispatch and MIN(trigger_at) computation with correct redelivery semantics. Given the finding above came directly out of that duplication, it might be worth folding run_wake_at into the existing schedule-events machinery rather than maintaining two parallel "next alarm" tracks long-term. Not asking for a rewrite in this PR, just flagging the maintenance cost.

Smaller: WorkflowState is defined identically in three places (actor/config.ts, inspector/workflow.ts, and the canonical @rivetkit/workflow-engine), with no shared import — a future new state added to one copy won't fail typecheck in the others.

Performance (minor, non-blocking)

  • load_actor_snapshot (internal_storage/mod.rs:87) awaits the actor row, then load_last_pushed_alarm, then load_run_wake_at sequentially — three independent reads with no data dependency, now one query longer per actor startup than before this PR. Worth a tokio::try_join!.
  • complete_persisted_message (queue.rs:520) does a SELECT (verify_persisted_message_identity_unlocked) then a separate DELETE, both under the same lock — could combine into one round trip on the queue-completion hot path.

Nit

workflow_storage_batch_put's row/byte budget check (internal_storage/mod.rs:872-897) duplicates the arithmetic and bail message in validate_atomic_workflow_flush (~1002-1017) rather than sharing one validator — low risk today, but if KV_TX_MAX_ROWS/KV_TX_MAX_PAYLOAD_BYTES semantics change later, only one copy might get updated.

@NathanFlurry
NathanFlurry force-pushed the feat/workflows-public-host-apis branch from 2884e4e to 0ff6164 Compare August 19, 2026 19:00
@NathanFlurry NathanFlurry changed the title feat(workflows): expose public workflow host APIs feat(rivetkit): streamline workflow host APIs Aug 20, 2026
@NathanFlurry
NathanFlurry force-pushed the feat/workflows-public-host-apis branch from 817111f to 9546652 Compare August 20, 2026 06:32
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant