feat: wait for external events via .waitForEvent + engine.sendEvent#55
Open
danfry1 wants to merge 2 commits into
Open
feat: wait for external events via .waitForEvent + engine.sendEvent#55danfry1 wants to merge 2 commits into
danfry1 wants to merge 2 commits into
Conversation
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.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
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):A run reaching the wait is persisted as
waitingwith its lease released, and resumes — on any engine instance — when the event is delivered. The (optionally schema-validated) payload becomes the next step'sprev.Design
waitForEventexecution unit (mirrors.sleep), so it fits the replay model with no determinism constraints.workflow_eventstable / in-memory buffer). An event sent before the run reaches the wait is buffered and consumed when it gets there.waitRuncheck, in its own transaction, for an already-buffered event and stay reclaimable if one exists.wake_atmachinery; on expiry the run fails with the newWaitTimeoutError.engine.sendEventvalidates the payload against the wait's schema before delivering.Surface
RunStatus/StepStatusgainwaiting. New exportedWaitTimeoutError.StorageAdaptergainswaitRun/deliverEvent/takeEvent;claimNextRunnow also wakes duewaitingruns. Implemented across all four adapters (SQLite adds aworkflow_eventstable, created viaCREATE 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.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, andtest:bunall green.Base
Stacked on #51 (durable sleep) — it reuses the suspend/
wake_at/claim-wake machinery. Once #51 (and its base #52) merge tomain, this retargets tomain.StepStatus/RunStatusalready unionsleeping+waitinghere.