Skip to content

feat: wait for external events via .waitForEvent + engine.sendEvent#55

Open
danfry1 wants to merge 2 commits into
feat/durable-sleepfrom
feat/wait-for-event
Open

feat: wait for external events via .waitForEvent + engine.sendEvent#55
danfry1 wants to merge 2 commits into
feat/durable-sleepfrom
feat/wait-for-event

Conversation

@danfry1

@danfry1 danfry1 commented Jun 22, 2026

Copy link
Copy Markdown
Owner

Summary

Adds event-driven durable suspension — the second half of durable suspend, alongside .sleep(). A workflow can pause until an external signal arrives (a webhook, an approval, a payment confirmation, a human action):

.step('create-payment-intent', async ({ input }) => ({ intentId: await stripe.createIntent(input.orderId) }))
.waitForEvent('payment.succeeded', { schema: z.object({ chargeId: z.string() }), timeoutMs: 30 * 60 * 1000 })
.step('fulfil', async ({ prev, input }) => ship(input.orderId, prev.chargeId)) // prev = the event payload

// ...later, from a webhook handler:
await engine.sendEvent(runId, 'payment.succeeded', { chargeId: event.data.object.id })

A run reaching the wait is persisted as waiting with its lease released, and resumes — on any engine instance — when the event is delivered. The (optionally schema-validated) payload becomes the next step's prev.

Design

  • New waitForEvent execution unit (mirrors .sleep), so it fits the replay model with no determinism constraints.
  • Durable, order-independent delivery. Events are persisted (a workflow_events table / in-memory buffer). An event sent before the run reaches the wait is buffered and consumed when it gets there.
  • Race-free. A deliver-during-suspend race is closed by having waitRun check, in its own transaction, for an already-buffered event and stay reclaimable if one exists.
  • Timeouts reuse the wake_at machinery; on expiry the run fails with the new WaitTimeoutError.
  • engine.sendEvent validates the payload against the wait's schema before delivering.

Surface

  • RunStatus / StepStatus gain waiting. New exported WaitTimeoutError.
  • StorageAdapter gains waitRun / deliverEvent / takeEvent; claimNextRun now also wakes due waiting runs. Implemented across all four adapters (SQLite adds a workflow_events table, created via CREATE TABLE IF NOT EXISTS — no migration needed).

Tests

  • wait-for-event.test.ts (10): suspend/resume, early-arrival buffering, timeout → WaitTimeoutError, in-time delivery, schema validation, unknown-event / missing-run, cross-engine crash recovery, sequential waits, cancel-while-waiting, duplicate-name.
  • Adapter contract tests for waitRun / deliverEvent / takeEvent + claim-wake across memory / better-sqlite3 / node:sqlite, plus the buffered-event race case. Bun smoke test extended for the new methods.

404 tests pass; typecheck, build, coverage (85.4% branches), lint:deps, and test:bun all green.

Base

Stacked on #51 (durable sleep) — it reuses the suspend/wake_at/claim-wake machinery. Once #51 (and its base #52) merge to main, this retargets to main. StepStatus/RunStatus already union sleeping + waiting here.

danfry1 added 2 commits June 22, 2026 21:41
Add durable, event-driven suspension — the second half of durable
suspend alongside .sleep():

  .waitForEvent('payment.succeeded', { schema, timeoutMs })
  // ...
  await engine.sendEvent(runId, 'payment.succeeded', payload)

A run reaching waitForEvent is persisted as 'waiting' with its lease
released, and resumes on any engine instance when the event is
delivered; the (optionally schema-validated) payload becomes the next
step's prev. Delivery is durable and order-independent — an event sent
before the run reaches the wait is buffered (workflow_events table) and
consumed when it gets there. With timeoutMs, the run fails with the new
WaitTimeoutError if no event arrives in time.

A deliver-during-suspend race is closed by having waitRun check, in its
own transaction, for an already-buffered event and stay reclaimable if
one exists.

Implementation:
- New waitForEvent execution unit; RunStatus/StepStatus gain 'waiting'.
- StorageAdapter gains waitRun/deliverEvent/takeEvent; claimNextRun wakes
  due 'waiting' runs. Implemented across all four adapters (SQLite adds a
  workflow_events table). engine.sendEvent validates + delivers.
- New exported WaitTimeoutError.
Three correctness fixes found in an adversarial review:

- Lease loss after consuming an event no longer loses it: if
  saveStepResult fails (the run was reclaimed), the event is re-delivered
  before bailing, so the reclaiming engine consumes it. Previously the
  event was deleted by takeEvent and gone, leaving the run waiting forever.
- The payload is no longer re-validated on consume. It is validated once
  in sendEvent; re-validating the stored value broke non-idempotent
  transform schemas (e.g. z.string().transform(s => s.length)) by failing
  the run during execution.
- sendEvent returns false (and buffers nothing) for a run that has already
  finished, instead of leaking an orphan event and reporting success.

Also fixes the waitForEvent schema typing to infer the payload from the
schema's OUTPUT (StandardSchemaV1<unknown, TPayload>), so transform and
coerce schemas type-check.

Adds regression tests for all three.
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