From ff3b8a16a4cb40bbe9914321b196d43de0ca8e44 Mon Sep 17 00:00:00 2001 From: Peter Wielander Date: Fri, 11 Sep 2026 09:11:05 -0700 Subject: [PATCH 1/3] [core] Give the QuickJS engine log-position parity and document who names one Every write the QuickJS engine makes from its view of the log now carries `eventCount`, a single inline step's terminal write asks for the inline delta (`sinceCursor`), and the page a World hands back on either is queued for the live VM in position order ahead of the next `events.list`, which now reads forward from a cursor instead of the top of the log. Run-terminal writes and the terminal drain send nothing, as nothing replays the log after them. A World whose ids are not slots turns all of this off and gets the engine's previous behavior. Documents the rule for which writes name a position (the writer holds a loaded log it keeps deciding from) as a per-event table in the event-sourcing docs and above `slotSnapshotParams`, with a World-author pointer in building-a-world. Co-Authored-By: Claude Fable 5 --- .changeset/quickjs-slot-report-parity.md | 5 + .../docs/v5/how-it-works/event-sourcing.mdx | 27 ++ docs/content/worlds/v5/building-a-world.mdx | 2 + packages/core/src/runtime.ts | 7 + packages/core/src/runtime/helpers.ts | 47 +++ .../core/src/runtime/quickjs-entrypoint.ts | 213 +++++++++++-- .../core/src/runtime/quickjs-log-view.test.ts | 137 ++++++++ packages/core/src/runtime/quickjs-log-view.ts | 226 ++++++++++++++ .../src/runtime/quickjs-slot-report.test.ts | 293 ++++++++++++++++++ 9 files changed, 930 insertions(+), 27 deletions(-) create mode 100644 .changeset/quickjs-slot-report-parity.md create mode 100644 packages/core/src/runtime/quickjs-log-view.test.ts create mode 100644 packages/core/src/runtime/quickjs-log-view.ts create mode 100644 packages/core/src/runtime/quickjs-slot-report.test.ts diff --git a/.changeset/quickjs-slot-report-parity.md b/.changeset/quickjs-slot-report-parity.md new file mode 100644 index 0000000000..fad6ac8db6 --- /dev/null +++ b/.changeset/quickjs-slot-report-parity.md @@ -0,0 +1,5 @@ +--- +'@workflow/core': patch +--- + +The QuickJS engine now reports its log position on writes and feeds its VM from the events a World returns on the response, matching the node:vm engine. diff --git a/docs/content/docs/v5/how-it-works/event-sourcing.mdx b/docs/content/docs/v5/how-it-works/event-sourcing.mdx index cb6091010d..5539627c57 100644 --- a/docs/content/docs/v5/how-it-works/event-sourcing.mdx +++ b/docs/content/docs/v5/how-it-works/event-sourcing.mdx @@ -267,6 +267,33 @@ Both kinds of skip are logged at `debug`, so neither reaches the console unless The observability UI grays out events it can identify this way and shows the reason on hover. Its set is narrower than the runtime's because it reads the log without consumer state. A consumer for an entity that is still open can legitimately claim a repeat because each step retry writes another `step_started`. The UI marks a repeat only when no consumer can remain for it: after a terminal event for the same entity or at a second `run_started`, of which the log records one per run. The UI marks nothing on a partial log view, such as one page of a paginated list or search results, because identifying the first copy requires the entire log. +## Events returned on a write + +Concurrent invocations of one run write to a shared log, and a write is decided against the log the writer had loaded. When the write lands above other events the writer had not seen, the World can hand those events back on the write's own success response, so the writer merges them and continues rather than discovering the gap on its next read. Two parameters of `events.create()` ask for this: + +- **`eventCount`** says how many events the writer held. When the write commits at a higher position than `eventCount + 1`, the World returns the events on the positions in between (the *skipped-slot report*). +- **`sinceCursor`** is the `events.list()` cursor the writer had read to. The World returns everything after it, the write itself included (the *inline delta*), in the same `events` / `cursor` / `hasMore` shape as a list page. + +Both save a reload, so the rule for which writes send them is not about the event type. It is about whether the process that writes **holds a loaded log it will keep deciding from**: its own next writes, or the replay it resumes after the write. A writer with no log has nothing to merge a page into, and whoever decides next is a fresh replay that loads the log anyway. Three things must hold for a page to come back: the writer named a position, the World reads a page for that event type (a backend may decline for types whose writer never holds a log), and there is something to return (the report only when positions were skipped; the delta always). + +| Event | Writer | Sends | Why | +|-------|--------|-------|-----| +| `run_created` | `start()` | Nothing | No log exists yet. | +| `run_started` | Replay, first write of a delivery | `eventCount` when a log is loaded; no cursor | The `run_started` response already carries the full log as a preload. | +| `step_created`, `wait_created`, `hook_disposed`, `attr_set` | Suspension handler | `eventCount` | The handler keeps writing from this log and the replay resumes from it. | +| `hook_created` | Suspension handler | `eventCount` and `sinceCursor` (one hook write per suspension) | The delta is how a hook whose payload already arrived resolves without another invocation. | +| `hook_received` | Replay resuming a hook | `eventCount` | The replay continues from the log right after. | +| `hook_received` | Webhook or `resumeHook()` | Nothing | Out-of-band. It enqueues a delivery that loads the log. | +| `wait_completed` | Replay completing an elapsed wait | `eventCount` | The replay continues from the log right after. | +| `step_started`, `step_retrying` | Step executor | Nothing | The executor holds no log. The next decision is a replay that reloads, or receives the delta on the step's terminal write. | +| `step_completed`, `step_failed` | Step executor | `sinceCursor` when a single step runs inline; nothing from a queued delivery | Inline, this is where the invocation gets its log back without a list. From a queue, a fresh replay loads the log anyway. | +| `attr_set` | `setAttributes()` outside a workflow | Nothing | Out-of-band. | +| `run_completed`, `run_failed`, `run_cancelled` | Replay, or `cancel()` | Never `sinceCursor` | Terminal. Nothing replays the log afterwards, so a page would be read by no one. | + +Batched writes (`events.createBatch()`) carry no position and get no page. Both runtime engines follow this table: the `node:vm` engine merges a returned page into the log it replays from, and the QuickJS engine queues it for delivery to its live VM, in position order, ahead of its next `events.list()`. + +A World that supports this returns the page in `events`, `cursor`, and `hasMore` on the `EventResult`. A World that does not simply returns the created event, and the runtime lists. See [Building a World](/worlds/building-a-world#event-id-allocation) for the World-side contract. + ## Sealed positions (noop events) Runs at `specVersion` 7 and above use a *sealed log*. The backend gives each write its position from a per-run sequencer **before** the write commits, so concurrent writers hold distinct positions and never race for a slot. New runs use this behavior by default. [`WORKFLOW_SEALED_LOG=0`](/docs/configuration/runtime-tuning#workflow_sealed_log) returns a deployment to the previous scheme. Every runtime reads a sealed log regardless, and a run's version is fixed at creation, so changing the setting never affects an in-flight run. However, a writer can claim a position and then stop because of a crashed process or canceled transaction. This leaves a hole that no writer will fill, and a hole looks like an event the reader failed to load. diff --git a/docs/content/worlds/v5/building-a-world.mdx b/docs/content/worlds/v5/building-a-world.mdx index e71cdb66ee..3b7c2ff2c9 100644 --- a/docs/content/worlds/v5/building-a-world.mdx +++ b/docs/content/worlds/v5/building-a-world.mdx @@ -197,6 +197,8 @@ The version a World stamps comes from `mintedSpecVersion()`: 7 by default or the `events.create()` params carry `eventCount`: how many events the writer held in the log it replayed from, which is the position it expects to land on minus one. Attempt `eventCount + 1`. When that position is taken, **do not reject the write**. Advance to the next free position, commit there, and return the events occupying the positions you skipped over on the success response, in `events` with a matching `cursor` and `hasMore`. The writer merges them into its own log and replays once, rather than paying a second round trip to discover it was behind. A caller with a stale count is the normal case for a fan-out, and rejecting it would serialize writes the runtime issues in parallel. +Not every write carries `eventCount`, and a World is free to skip the report read for event types whose writer never holds a log to merge it into (the step executor's `step_started`, `step_retrying`, `step_completed`, and `step_failed`, and the run-terminal `run_completed`, `run_failed`, and `run_cancelled`). Which writes send `eventCount` or `sinceCursor`, and why, is tabulated in [Events returned on a write](/docs/how-it-works/event-sourcing#events-returned-on-a-write). When a write carries both, the `sinceCursor` delta is a superset of the report, so return the delta and skip the report. + ### Optional: rejecting a stale write A replay writes events derived from the event log it loaded, so a write made from an event log that no longer matches the store can commit events no correct replay would produce. Rejecting one means answering `events.create()` with a `PreconditionFailedError` when the run's log already holds more events than the caller's `eventCount` says it had loaded. diff --git a/packages/core/src/runtime.ts b/packages/core/src/runtime.ts index 51f013bd8a..6f329968e3 100644 --- a/packages/core/src/runtime.ts +++ b/packages/core/src/runtime.ts @@ -2885,6 +2885,13 @@ export function workflowEntrypoint( ? eventLog.events : undefined, preloadedEventsComplete: eventLog.type === 'ready', + // Where that log was read to, so the engine reads + // forward from it and can ask for an inline delta + // against it. + preloadedCursor: + eventLog.type === 'ready' + ? eventLog.cursor + : undefined, runInput, parentSpan: span, maxEventsLimit, diff --git a/packages/core/src/runtime/helpers.ts b/packages/core/src/runtime/helpers.ts index 82d9e44016..c9c4cf90e6 100644 --- a/packages/core/src/runtime/helpers.ts +++ b/packages/core/src/runtime/helpers.ts @@ -1032,6 +1032,53 @@ export interface SlotSnapshotParams { * Empty for an empty log, which is the state a `run_created` write is issued * from: there is no position held yet to name. * + * ## Who names a position + * + * A World can hand events back on a write's success response through two + * params (`CreateEventParams` in `@workflow/world`): `eventCount`, which is + * answered with the events on the positions the write skipped over (the + * skipped-slot report), and `sinceCursor`, which is answered with everything + * after that cursor (the inline delta). Both exist to save the writer a + * reload. So the rule for which writes send them is not about the event type; + * it is about whether **the process that writes holds a loaded log it will + * keep deciding from**, its own next writes or the replay it resumes. A writer + * with no log has nothing to merge a page into, and whoever decides next is a + * fresh replay that loads the log anyway. + * + * Three gates, all of which must pass for a page to come back: + * + * 1. The writer names a position. Only the replay loop's `createEvent` seam + * (runtime.ts) and the suspension handler's `createGuarded` do; the QuickJS + * engine's `createEvent` (quickjs-entrypoint.ts) follows the same rule. Raw + * `world.events.create` calls send nothing. + * 2. The World reads a page for that event type. A World may decline for types + * whose writer never holds a log (step executor writes, run-terminal + * writes), whatever the client sent. + * 3. There is something to hand back: the report only when the write landed + * more than one position above `eventCount`; the delta always. + * + * | Event | Writer | Sends | Why | + * |---|---|---|---| + * | `run_created` | `start()` | nothing | no log exists yet | + * | `run_started` | replay loop, first write of a delivery | `eventCount` when a log is loaded; no cursor, the `run_started` preload owns the response fields | the preload already returns the full log | + * | `step_created`, `wait_created`, `hook_disposed`, `attr_set` | suspension handler (`createGuarded`) | `eventCount` | the handler keeps writing from, and resumes replay from, this log | + * | `step_created` | queue-consumer re-ensure (raw) | nothing | runs from a queue message, no log | + * | `hook_created` | suspension handler | `eventCount` + `sinceCursor` (at most one hook write per suspension) | the delta is how an already-received hook resolves without a re-invocation | + * | `hook_received` | replay loop lazy resume; handler in-suspension resume | `eventCount` (preload owns the cursor fields) | both replay from the log right after | + * | `hook_received` | webhook / `resumeHook` (raw) | nothing | out-of-band, enqueues a delivery that loads the log | + * | `wait_completed` | replay loop timer path | `eventCount` | the loop keeps replaying from the log | + * | `wait_completed` | `runs.*` out-of-band (raw) | nothing | no log | + * | `step_started`, `step_retrying` | step executor | nothing | no log; the next decision is a replay that reloads or receives the delta below | + * | `step_completed`, `step_failed` | step executor | `sinceCursor` only, outside turbo, single inline step | this is where the inline execution loop gets its log back; in turbo or from a queued delivery a fresh replay loads it anyway | + * | `step_failed` | suspension handler, unserializable input | `eventCount` | the handler holds a log; rare, and follows a `step_created` that drew the page | + * | `attr_set` | `setAttributes()` API (raw) | nothing | out-of-band | + | `run_completed`, `run_failed`, `run_cancelled` | replay loop, replay budget, `runs.cancel` | `eventCount` where the node loop's seam stamps it (the QuickJS engine sends nothing); never `sinceCursor` (`deltaRequestCursor` excludes terminal types) | terminal: nothing replays the log afterwards, so a World reads no page for them | + * + * Batched writes (`events.createBatch`) are outside all of this: a batch carries + * no per-event position and gets no page. The user-facing version of this table + * is in `docs/content/docs/v5/how-it-works/event-sourcing.mdx`; keep the two in + * step. + * * The maximum rather than the length, for the reason {@link maxEventSlot} * gives: a partially-read log holds fewer events than its highest position, and * counting those would make the write claim to have seen less than it has, so diff --git a/packages/core/src/runtime/quickjs-entrypoint.ts b/packages/core/src/runtime/quickjs-entrypoint.ts index d23ef01331..a8fa112036 100644 --- a/packages/core/src/runtime/quickjs-entrypoint.ts +++ b/packages/core/src/runtime/quickjs-entrypoint.ts @@ -23,7 +23,10 @@ import { } from '@workflow/errors'; import { parseWorkflowName } from '@workflow/utils/parse-name'; import { + type CreateEventParams, + type CreateEventRequest, type Event, + type EventResult, ROOT_RUN_ID_ATTRIBUTE, type RunInput, SPEC_VERSION_CURRENT, @@ -60,6 +63,7 @@ import { queueMessage, stepDispatchIdempotencyKey, } from './helpers.js'; +import { QuickJSLogView } from './quickjs-log-view.js'; import { BASELINE_BUNDLE_FILENAME, type PendingAttribute, @@ -77,6 +81,12 @@ import { unserializableStepInputPlaceholder } from './unserializable-step.js'; import { getWaitContinuationDispatch } from './wait-continuation.js'; import { getWorld } from './world.js'; +/** An `events.create` bound to the run; see `dispatchPendingOps.createEvent`. */ +type EventCreator = ( + data: CreateEventRequest, + params?: CreateEventParams +) => Promise; + /** Tiny ms timer using performance.now(), already monotonic on Node. */ function tick(): number { return performance.now(); @@ -220,6 +230,14 @@ async function queueStepMessage(params: { async function dispatchPendingOps(params: { world: Awaited>; runId: string; + /** + * The seam every event write in this pass goes through. The inline loop + * passes a create that names the log position this invocation holds + * (`eventCount`) and queues what the World hands back for the live VM; see + * {@link QuickJSLogView}. The terminal drain passes a plain create, since + * the run is ending and nothing reads its log afterwards. + */ + createEvent: EventCreator; workflowRun: WorkflowRun; encryptionKey: RunPayloadKeys | undefined; pendingOperations: PendingOperation[]; @@ -281,6 +299,7 @@ async function dispatchPendingOps(params: { pendingOperations, namespace, nextTraceCarrier, + createEvent, } = params; const skipStepCreation = params.skipStepCreation; const queueStepCids = params.queueStepCids; @@ -349,7 +368,7 @@ async function dispatchPendingOps(params: { typeof hook.metadata === 'undefined' ? undefined : await encryptSerializedData(hook.metadata, encryptionKey); - const result = await world.events.create(runId, { + const result = await createEvent({ eventType: 'hook_created', specVersion: SPEC_VERSION_CURRENT, correlationId: hook.correlationId, @@ -409,7 +428,7 @@ async function dispatchPendingOps(params: { )) as Uint8Array) : undefined; try { - await world.events.create(runId, { + await createEvent({ eventType: 'hook_received', specVersion: SPEC_VERSION_CURRENT, correlationId: hook.correlationId, @@ -451,7 +470,7 @@ async function dispatchPendingOps(params: { op: PendingHookDispose ): Promise => { try { - await world.events.create(runId, { + await createEvent({ eventType: 'hook_disposed', specVersion: SPEC_VERSION_CURRENT, correlationId: op.correlationId, @@ -555,7 +574,7 @@ async function dispatchPendingOps(params: { } ); try { - await world.events.create(runId, { + await createEvent({ eventType: 'step_created', specVersion: SPEC_VERSION_CURRENT, correlationId: step.correlationId, @@ -579,7 +598,7 @@ async function dispatchPendingOps(params: { if (!EntityConflictError.is(err)) throw err; } try { - await world.events.create(runId, { + await createEvent({ eventType: 'step_failed', specVersion: SPEC_VERSION_CURRENT, correlationId: step.correlationId, @@ -637,7 +656,7 @@ async function dispatchPendingOps(params: { encryptedInput.byteLength <= MAX_RESILIENT_STEP_INPUT_BYTES ) { const [createResult, queueResult] = await Promise.allSettled([ - world.events.create(runId, { + createEvent({ eventType: 'step_created', specVersion: SPEC_VERSION_CURRENT, correlationId: step.correlationId, @@ -705,7 +724,7 @@ async function dispatchPendingOps(params: { } try { - await world.events.create(runId, { + await createEvent({ eventType: 'step_created', specVersion: SPEC_VERSION_CURRENT, correlationId: step.correlationId, @@ -730,7 +749,7 @@ async function dispatchPendingOps(params: { opsPromises.push( (async () => { try { - await world.events.create(runId, { + await createEvent({ eventType: 'attr_set', specVersion: SPEC_VERSION_CURRENT, correlationId: attr.correlationId, @@ -759,7 +778,7 @@ async function dispatchPendingOps(params: { opsPromises.push( (async () => { try { - await world.events.create(runId, { + await createEvent({ eventType: 'wait_created', specVersion: SPEC_VERSION_CURRENT, correlationId: wait.correlationId, @@ -793,14 +812,28 @@ async function dispatchPendingOps(params: { * This replaces the `node:vm` replay path (runWorkflow + EventsConsumer) * with a QuickJS VM invocation that performs the same full event replay. * - * KNOWN GAP (slot snapshot): unlike the node:vm path, no event write in - * this file carries {@link CreateEventParams.eventCount}, so a World never - * learns which events the writer had not seen and never reports them back. - * The engine currently relies on per-(runId, correlationId) event - * uniqueness (EntityConflictError dedup) alone. This is a deliberate - * simplification while the engine is experimental: wiring the snapshot is - * tracked follow-up work; anyone adding new write paths here should not - * assume parity with the node engine on this axis. + * Log position on writes. This engine follows the same rule as the node:vm + * replay loop for which writes tell the World where the writer stood (see the + * "Who names a position" table above `slotSnapshotParams` in `helpers.ts`): + * + * - Writes made from this invocation's view of the log (`step_created`, + * `wait_created`, `hook_created`, `hook_disposed`, `attr_set`, the abort + * `hook_received`, `wait_completed`, and the `step_created` + `step_failed` + * pair for an unserializable input) carry `eventCount`, and the page a + * World hands back is queued for the live VM through {@link QuickJSLogView}. + * - A single inline step's terminal write asks for the inline delta + * (`sinceCursor`) through `executeStep`, and the delta is queued the same + * way. The step executor's other writes carry nothing: it holds no log. + * - Run-terminal writes (`run_completed`, `run_failed`) and the terminal drain + * of pending ops carry nothing: nothing reads the log afterwards. + * + * What differs from the node engine is what "merge into the log" means. The + * node engine merges a returned page into the array it replays from. This + * engine holds a live VM that consumes events exactly once, in position + * order, so a returned page is queued and delivered ahead of the next + * `events.list`, which then only runs when the queue cannot account for the + * next position. Both engines fall back to a list for anything a page did + * not carry. */ export async function runWorkflowWithQuickJS(params: { workflowCode: string; @@ -822,6 +855,13 @@ export async function runWorkflowWithQuickJS(params: { * hook-resume preload would be discarded and refetched. */ preloadedEventsComplete?: boolean; + /** + * The `events.list` cursor positioned after the last of `preloadedEvents`, + * when the caller has one. Lets this invocation read incrementally from + * where the preload ended and ask for an inline delta against it. Without + * it the first read after the preload starts from the top of the log. + */ + preloadedCursor?: string | null; /** * Run input carried through the queue message on first delivery. Used * as a last-resort fallback for `run_created.eventData.input` when @@ -889,6 +929,7 @@ export async function runWorkflowWithQuickJS(params: { workflowRun, preloadedEvents, preloadedEventsComplete, + preloadedCursor, runInput, parentSpan, maxEventsLimit, @@ -979,6 +1020,10 @@ export async function runWorkflowWithQuickJS(params: { // preload (lazy hook fast path) is trusted the same way. let events: Event[]; let eventsFetchedPages = 0; + // Where the log was read to: the cursor after the last page below, or the + // one the caller read the preload to. Seeds the incremental reads and the + // inline-delta requests that follow. + let loadedCursor: string | null = null; const usePreloaded = (preloadedEventsComplete === true && Array.isArray(preloadedEvents) && @@ -986,6 +1031,7 @@ export async function runWorkflowWithQuickJS(params: { isFirstInvocation(preloadedEvents); if (usePreloaded && preloadedEvents) { events = preloadedEvents; + loadedCursor = preloadedCursor ?? null; } else { const allEvents: Event[] = []; let cursor: string | null = null; @@ -1012,8 +1058,42 @@ export async function runWorkflowWithQuickJS(params: { } events = allEvents; + loadedCursor = cursor; } + // This invocation's view of the log, and the queue of events a World has + // handed back on a write that the VM has not been given yet. Every write + // made from this view goes through `createEvent` below so it names the + // position it was decided against and its response is queued here. + const logView = new QuickJSLogView(events, loadedCursor); + const createEvent: EventCreator = async (data, eventParams) => { + const result = await world.events.create(runId, data, { + ...eventParams, + ...logView.snapshotParams(), + }); + const absorbed = logView.absorb(result); + if (absorbed.truncated) { + runtimeLogger.debug( + 'QuickJS runtime: dropped a truncated skipped-slot report', + { + workflowRunId: runId, + eventType: data.eventType, + eventId: result.event?.eventId, + offered: result.events?.length ?? 0, + } + ); + } + return result; + }; + /** + * The create for writes that end the run (`run_completed`, `run_failed`, + * and the terminal drain of leftover ops): no position named and no page + * asked for, because nothing replays a finished run's log. The node + * engine's `deltaRequestCursor` makes the same exclusion for `sinceCursor`. + */ + const terminalCreateEvent: EventCreator = (data, eventParams) => + world.events.create(runId, data, eventParams); + // Event-limit guard: fail a runaway run once its log reaches the // server-supplied ceiling, the same enforcement point as the node:vm // engine's replay loop. @@ -1057,12 +1137,16 @@ export async function runWorkflowWithQuickJS(params: { const resumeAt = eventData?.resumeAt; if (resumeAt && now >= new Date(resumeAt as string).getTime()) { try { - const result = await world.events.create(runId, { + const result = await createEvent({ eventType: 'wait_completed', specVersion: SPEC_VERSION_CURRENT, correlationId: event.correlationId, }); - if (result.event) events.push(result.event); + if (!logView.tracking && result.event) { + // No positions to order by (see QuickJSLogView), so the event + // joins the initial log directly, as it always has. + events.push(result.event); + } } catch (err) { if (EntityConflictError.is(err)) continue; throw err; @@ -1070,6 +1154,10 @@ export async function runWorkflowWithQuickJS(params: { } } } + // The VM has not started, so whatever those writes handed back (each + // wait_completed, plus anything another writer appended that they skipped + // over) joins the initial log instead of waiting for a feed. + events.push(...logView.takeContiguous()); // Resolve the workflow server port so `getWorkflowMetadata().url` inside // the VM matches what the step-side handler reports. Skipped on Vercel: @@ -1228,10 +1316,17 @@ export async function runWorkflowWithQuickJS(params: { // and nothing scheduled to read it. let pendingRequeueSignal = false; - /** Fetch all events not yet processed by the live VM (log order). */ + /** + * Fetch all events not yet processed by the live VM (log order), reading + * from where the log was last read to. A World's cursor never passes a + * position whose writer is still in flight, so reading forward from it + * cannot skip an event; the id set is what makes a re-read of the same + * span (after a queued page or a delta moved the view ahead of the cursor) + * harmless. + */ const fetchUnseenEvents = async (): Promise => { const unseen: Event[] = []; - let cursor: string | null = null; + let cursor: string | null = logView.logCursor; let hasMore = true; while (hasMore) { const response = await world.events.list({ @@ -1250,10 +1345,30 @@ export async function runWorkflowWithQuickJS(params: { if (response.cursor) cursor = response.cursor; hasMore = response.data.length > 0 && response.cursor != null; } + logView.advanceCursor(cursor); + logView.markFed(unseen); observeEventsForOwnership(unseen); return unseen; }; + /** + * Events a World handed back on this invocation's writes that the VM can + * take now: the contiguous run above what it has (see + * `QuickJSLogView.takeContiguous`). Delivered ahead of `fetchUnseenEvents` + * so a write's response, not a listing, is what usually carries the log + * forward, which is the round-trip the page exists to save. Empty when + * nothing is queued or the next position is not in hand, and the caller + * lists. + */ + const takeQueuedEvents = (): Event[] => { + const queued = logView.takeContiguous(); + for (const e of queued) { + if (e.eventId) seenEventIds.add(e.eventId); + } + observeEventsForOwnership(queued); + return queued; + }; + try { let iteration = 0; while (result.suspended && !runGone && !budget.isExhausted()) { @@ -1334,6 +1449,7 @@ export async function runWorkflowWithQuickJS(params: { encryptionKey, namespace, nextTraceCarrier, + createEvent, pendingOperations: opsToDispatch, skipStepCreation: inlineClaimCids, queueStepCids: new Set(overflowSteps.map((s) => s.correlationId)), @@ -1393,7 +1509,7 @@ export async function runWorkflowWithQuickJS(params: { waitCompletePromises.push( (async () => { try { - await world.events.create(runId, { + await createEvent({ eventType: 'wait_completed', specVersion: SPEC_VERSION_CURRENT, correlationId: wait.correlationId, @@ -1410,9 +1526,13 @@ export async function runWorkflowWithQuickJS(params: { } // 2. Cheap progress first: feed newly recorded events into the live - // VM before blocking on step bodies. + // VM before blocking on step bodies. What the writes above handed back + // is delivered first; a listing runs only when that does not reach the + // next position. { - const newEvents = await fetchUnseenEvents(); + const queued = takeQueuedEvents(); + const newEvents = + queued.length > 0 ? queued : await fetchUnseenEvents(); if (newEvents.length > 0) { // The listing caught up with this invocation's writes, so any // attr_set / getConflict hook_created has been (or is being) @@ -1423,6 +1543,7 @@ export async function runWorkflowWithQuickJS(params: { iteration, phase: 'feed', fedEvents: newEvents.length, + fedFrom: queued.length > 0 ? 'write-response' : 'list', outcome: result.completed ? 'completed' : result.failed @@ -1596,6 +1717,19 @@ export async function runWorkflowWithQuickJS(params: { // own, matching the node:vm engine, where a long sequential // workflow likewise runs step-by-step until the platform reclaims // the invocation and a redelivery resumes from the log. + // Inline delta: a single inline step's terminal write asks the World + // for everything after the cursor this view holds, so the step's own + // events (and anything interleaved) arrive on the write's response and + // the feed below needs no listing. Only for a batch of one, as in the + // node engine: several steps would each get a delta from the same + // cursor, and only one could be taken. Not requested when the log has + // no cursor to name (tracking off, or nothing read yet). + const inlineDeltaSinceCursor = + inlineCandidates.length === 1 && + logView.tracking && + typeof logView.logCursor === 'string' + ? logView.logCursor + : undefined; budget.pause(); let outcomes: StepExecutionResult[]; try { @@ -1632,6 +1766,9 @@ export async function runWorkflowWithQuickJS(params: { // A lazy step is brand-new by construction: first // attempt. authoritativeAttempt: 1, + ...(inlineDeltaSinceCursor !== undefined + ? { inlineDeltaSinceCursor } + : {}), }))() ) ) @@ -1645,6 +1782,23 @@ export async function runWorkflowWithQuickJS(params: { const step = inlineCandidates[i]; const outcome = outcomes[i]; executedStepIds.add(step.correlationId); + if ( + outcome.type === 'completed' && + outcome.inlineDelta !== undefined && + inlineDeltaSinceCursor !== undefined + ) { + const advanced = logView.absorbDelta( + inlineDeltaSinceCursor, + outcome.inlineDelta + ); + wfdiag('inline_delta_absorbed', { + iteration, + correlationId: step.correlationId, + events: outcome.inlineDelta.events.length, + hasMore: outcome.inlineDelta.hasMore, + cursorAdvanced: advanced, + }); + } if (outcome.type === 'retry' || outcome.type === 'throttled') { // Hand the step to the queue with the requested backoff: // background delivery drives the retry from here. @@ -1690,7 +1844,8 @@ export async function runWorkflowWithQuickJS(params: { // body; 'gone', retry/throttled: a queue message exists) don't // need it, but signaling on them too only costs a no-op invocation // in an already-rare lag window. - const newEvents = await fetchUnseenEvents(); + const queued = takeQueuedEvents(); + const newEvents = queued.length > 0 ? queued : await fetchUnseenEvents(); if (newEvents.length === 0) { pendingRequeueSignal = true; break; @@ -1740,6 +1895,9 @@ export async function runWorkflowWithQuickJS(params: { encryptionKey, namespace, nextTraceCarrier, + // Plain create: the run is ending, nothing replays its log, so a + // page handed back here would be read by no one. + createEvent: terminalCreateEvent, pendingOperations: result.completed.drainOperations, wfdiag, }); @@ -1758,7 +1916,7 @@ export async function runWorkflowWithQuickJS(params: { // events have the same `encr`-prefixed payload shape that the node:vm // engine's `dehydrateWorkflowReturnValue` produces. try { - await world.events.create(runId, { + await terminalCreateEvent({ eventType: 'run_completed', specVersion: SPEC_VERSION_CURRENT, eventData: { @@ -1987,6 +2145,7 @@ export async function runWorkflowWithQuickJS(params: { encryptionKey, namespace, nextTraceCarrier, + createEvent: terminalCreateEvent, pendingOperations: result.failed.drainOperations, wfdiag, }); @@ -2112,7 +2271,7 @@ export async function runWorkflowWithQuickJS(params: { } } try { - await world.events.create(runId, { + await terminalCreateEvent({ eventType: 'run_failed', specVersion: SPEC_VERSION_CURRENT, eventData: { diff --git a/packages/core/src/runtime/quickjs-log-view.test.ts b/packages/core/src/runtime/quickjs-log-view.test.ts new file mode 100644 index 0000000000..2179d0b9cd --- /dev/null +++ b/packages/core/src/runtime/quickjs-log-view.test.ts @@ -0,0 +1,137 @@ +import { type Event, eventIdToSlot, slotToEventId } from '@workflow/world'; +import { describe, expect, it } from 'vitest'; +import { QuickJSLogView } from './quickjs-log-view.js'; + +function slotEvent(slot: number): Event { + return { + eventId: slotToEventId(slot), + eventType: 'attr_set', + runId: 'wrun_view', + createdAt: new Date(0), + specVersion: 7, + correlationId: `attr_${slot}`, + } as unknown as Event; +} + +const slots = (events: readonly Event[]) => + events.map((e) => eventIdToSlot(e.eventId)); + +describe('QuickJSLogView', () => { + it('names the highest position it has seen on the next write', () => { + const view = new QuickJSLogView([slotEvent(1), slotEvent(2)], 'c2'); + expect(view.snapshotParams()).toEqual({ eventCount: 2 }); + // Its own write landing at 5 raises the position it will name next, even + // though the VM has not been given that event yet. + view.absorb({ event: slotEvent(5) }); + expect(view.snapshotParams()).toEqual({ eventCount: 5 }); + }); + + it('names no position for an empty log', () => { + expect(new QuickJSLogView([], null).snapshotParams()).toEqual({}); + }); + + it('delivers a write and its skipped span in position order', () => { + const view = new QuickJSLogView([slotEvent(1), slotEvent(2)], 'c2'); + // The write landed at 5; the World hands back 3 and 4, which another + // writer appended while this one was deciding. + view.absorb({ + event: slotEvent(5), + events: [slotEvent(3), slotEvent(4), slotEvent(5)], + hasMore: false, + }); + expect(slots(view.takeContiguous())).toEqual([3, 4, 5]); + expect(view.bufferedCount).toBe(0); + expect(view.takeContiguous()).toEqual([]); + }); + + it('holds back a queued event above a position it does not have', () => { + const view = new QuickJSLogView([slotEvent(1)], 'c1'); + view.absorb({ event: slotEvent(3) }); + // Position 2 is not in hand, so 3 cannot be fed: the VM must not see it + // before whatever lands at 2. + expect(view.takeContiguous()).toEqual([]); + expect(view.bufferedCount).toBe(1); + // A listing delivers 2 to the VM; 3 is now next. + view.markFed([slotEvent(2)]); + expect(slots(view.takeContiguous())).toEqual([3]); + }); + + it('drops a truncated report whole but keeps the write itself', () => { + const view = new QuickJSLogView([slotEvent(1)], 'c1'); + const absorbed = view.absorb({ + event: slotEvent(10), + events: [slotEvent(2), slotEvent(3)], + hasMore: true, + }); + expect(absorbed).toEqual({ queued: 1, truncated: true }); + // 10 is queued and waits for a listing to fill 2..9. + expect(view.bufferedCount).toBe(1); + expect(view.takeContiguous()).toEqual([]); + }); + + it('drops an event a listing already delivered', () => { + const view = new QuickJSLogView([slotEvent(1)], 'c1'); + view.absorb({ event: slotEvent(3) }); + // The listing raced ahead and delivered 2 and 3 itself. + view.markFed([slotEvent(2), slotEvent(3)]); + expect(view.bufferedCount).toBe(0); + expect(view.takeContiguous()).toEqual([]); + }); + + it('advances the cursor on a complete delta that leaves no hole', () => { + const view = new QuickJSLogView([slotEvent(1), slotEvent(2)], 'c2'); + const advanced = view.absorbDelta('c2', { + events: [slotEvent(3), slotEvent(4)], + cursor: 'c4', + hasMore: false, + }); + expect(advanced).toBe(true); + expect(view.logCursor).toBe('c4'); + expect(slots(view.takeContiguous())).toEqual([3, 4]); + }); + + it('keeps the cursor when the delta was computed from an older position', () => { + const view = new QuickJSLogView([slotEvent(1)], 'c1'); + view.advanceCursor('c3'); + const advanced = view.absorbDelta('c1', { + events: [slotEvent(2)], + cursor: 'c2', + hasMore: false, + }); + expect(advanced).toBe(false); + expect(view.logCursor).toBe('c3'); + // Nothing was queued from it either: a listing from c3 covers the rest. + expect(view.bufferedCount).toBe(0); + }); + + it('keeps the cursor when the delta leaves a hole below its end', () => { + const view = new QuickJSLogView([slotEvent(1)], 'c1'); + // Position 4 is known (an own write) but 2 and 3 are not in hand. + view.absorb({ event: slotEvent(4) }); + const advanced = view.absorbDelta('c1', { + events: [slotEvent(5)], + cursor: 'c5', + hasMore: false, + }); + expect(advanced).toBe(false); + expect(view.logCursor).toBe('c1'); + // Moving the cursor to c5 would have made 2 and 3 unreachable by a + // listing, and the VM would never get past 1. + }); + + it('turns tracking off for good on an id that is not a position', () => { + const view = new QuickJSLogView([slotEvent(1)], 'c1'); + view.absorb({ event: slotEvent(2) }); + expect(view.bufferedCount).toBe(1); + view.absorb({ + event: { ...slotEvent(3), eventId: 'evnt_01ARZ3NDEKTSV4RRFFQ69G5FAV' }, + }); + expect(view.tracking).toBe(false); + expect(view.snapshotParams()).toEqual({}); + expect(view.bufferedCount).toBe(0); + expect(view.takeContiguous()).toEqual([]); + // Nothing queues again afterwards. + view.absorb({ event: slotEvent(4) }); + expect(view.bufferedCount).toBe(0); + }); +}); diff --git a/packages/core/src/runtime/quickjs-log-view.ts b/packages/core/src/runtime/quickjs-log-view.ts new file mode 100644 index 0000000000..5bbc3a8f5c --- /dev/null +++ b/packages/core/src/runtime/quickjs-log-view.ts @@ -0,0 +1,226 @@ +import { type Event, eventIdToSlot, FIRST_EVENT_SLOT } from '@workflow/world'; + +/** + * The QuickJS engine's bookkeeping for the run's event log, giving that engine + * the same relationship to a World's write responses that the node:vm replay + * loop has. + * + * The node engine holds the log as an array it replays from, so a page a World + * hands back on a write (a skipped-slot report against `eventCount`, or an + * inline delta against `sinceCursor`; see `CreateEventParams` in + * `@workflow/world`) is merged into that array and the next replay reads it + * there. The QuickJS engine holds a LIVE VM instead: events are delivered to it + * incrementally through `continueWithEvents`, in log order, exactly once. So + * for this engine a returned page is not merged into a log; it is queued to be + * delivered. This class owns that queue and the three numbers around it: + * + * - `knownMaxSlot`, the highest position this invocation has seen anywhere + * (fed, buffered, or its own write). Reported as `eventCount` on every write + * this invocation makes from its view of the log, so the World can hand + * back what landed above it. + * - `fedMaxSlot`, the highest position delivered to the VM. Delivery is + * strictly in position order with no gaps, because the VM consumes events + * as they arrive and a later replay will read the log in position order; + * feeding position 13 before 12 exists would let the two disagree. + * - `cursor`, the read position for `events.list`, advanced by list pages and + * by a complete inline delta. + * + * Positions come from slot-numbered event ids. A log whose ids are not slots + * (a World on the old id scheme, or a mocked World) turns tracking off for the + * rest of the invocation: no `eventCount` is sent, nothing is buffered, and the + * engine reads the log back the way it did before any of this existed. + */ +export class QuickJSLogView { + private slotTracking = true; + private knownMaxSlot: number | undefined; + private fedMaxSlot: number | undefined; + /** Events handed back by a World that the VM has not been given yet. */ + private readonly unfed = new Map(); + private cursor: string | null; + + constructor(fedEvents: readonly Event[], cursor: string | null) { + this.cursor = cursor; + this.markFed(fedEvents); + } + + /** Read position for the next `events.list`, or `null` for the start. */ + get logCursor(): string | null { + return this.cursor; + } + + /** How many events are queued for delivery to the VM. */ + get bufferedCount(): number { + return this.unfed.size; + } + + /** Whether positions are being tracked (see class doc). */ + get tracking(): boolean { + return this.slotTracking; + } + + /** + * The `eventCount` to attach to a write made from this view. Empty while no + * position is known (an empty log) or once tracking has been turned off. + */ + snapshotParams(): { eventCount?: number } { + return this.slotTracking && this.knownMaxSlot !== undefined + ? { eventCount: this.knownMaxSlot } + : {}; + } + + /** A page of `events.list` was read to `cursor`. */ + advanceCursor(cursor: string | null): void { + if (cursor !== null) { + this.cursor = cursor; + } + } + + /** + * Events delivered to the VM by the caller (the initial load, or an + * `events.list` page). Removes them from the delivery queue if a write + * response had already handed them back. + */ + markFed(events: readonly Event[]): void { + for (const event of events) { + const slot = this.slotOf(event); + if (slot === undefined) continue; + this.unfed.delete(slot); + if (this.fedMaxSlot === undefined || slot > this.fedMaxSlot) { + this.fedMaxSlot = slot; + } + } + } + + /** + * Absorb a write's response: the committed event, and the skipped-slot + * report or inline delta a World attached to it. + * + * A truncated page (`hasMore`) is dropped whole, the same policy as + * `absorbSkippedSlotReport` in the node engine: it covers a span of + * positions but carries only some of them, and queuing part of a span + * would have this view claim, on its next write, to have seen positions it + * never received. The committed event itself is always taken. + */ + absorb(result: { + event?: Event; + events?: readonly Event[]; + hasMore?: boolean; + }): { queued: number; truncated: boolean } { + let queued = 0; + if (result.event && this.queue(result.event)) { + queued++; + } + const page = result.events ?? []; + if (page.length === 0) { + return { queued, truncated: false }; + } + if (result.hasMore === true) { + return { queued, truncated: true }; + } + for (const event of page) { + if (this.queue(event)) queued++; + } + return { queued, truncated: false }; + } + + /** + * Absorb the inline delta a step-terminal write returned for + * `sinceCursor`, and advance the read cursor past it when that is safe. + * + * The delta is everything after `sentCursor`, so it may be taken only if + * the view still stands at that cursor (a list in between would have moved + * it, and appending a delta computed from an older position could deliver + * events behind ones already fed). The cursor advances only when the queue + * then holds every position between what the VM has and the delta's end: + * with a hole in between, a later `events.list` from the advanced cursor + * would never return the missing event and the VM would stall on it. + * + * @returns whether the cursor was advanced. + */ + absorbDelta( + sentCursor: string, + delta: { events: readonly Event[]; cursor: string | null; hasMore: boolean } + ): boolean { + if (!this.slotTracking || this.cursor !== sentCursor) { + return false; + } + if (delta.hasMore) { + return false; + } + for (const event of delta.events) { + this.queue(event); + } + if (!this.queueIsDense()) { + return false; + } + this.advanceCursor(delta.cursor); + return true; + } + + /** + * Drain the events that can be delivered to the VM now: the queued run of + * consecutive positions directly above the highest one already fed. A queued + * event above a position nothing has filled yet stays queued until a list + * fills the gap. + */ + takeContiguous(): Event[] { + if (!this.slotTracking) { + return []; + } + const out: Event[] = []; + // A VM started on an empty log is waiting for the first position. + let next = + this.fedMaxSlot === undefined ? FIRST_EVENT_SLOT : this.fedMaxSlot + 1; + for (;;) { + const event = this.unfed.get(next); + if (event === undefined) break; + this.unfed.delete(next); + out.push(event); + next++; + } + if (out.length > 0) { + this.fedMaxSlot = next - 1; + } + return out; + } + + private queueIsDense(): boolean { + if (this.knownMaxSlot === undefined) { + return this.unfed.size === 0; + } + const fed = this.fedMaxSlot ?? FIRST_EVENT_SLOT - 1; + return this.unfed.size === this.knownMaxSlot - fed; + } + + /** Queue one event for delivery unless the VM already has it. */ + private queue(event: Event): boolean { + const slot = this.slotOf(event); + if (slot === undefined) return false; + if (this.fedMaxSlot !== undefined && slot <= this.fedMaxSlot) { + return false; + } + if (this.unfed.has(slot)) return false; + this.unfed.set(slot, event); + return true; + } + + /** + * The event's position, bumping `knownMaxSlot`. `undefined` once an id + * turns out not to be a slot, which turns tracking off for good. + */ + private slotOf(event: Event): number | undefined { + if (!this.slotTracking) return undefined; + const slot = + typeof event.eventId === 'string' ? eventIdToSlot(event.eventId) : null; + if (slot === null) { + this.slotTracking = false; + this.knownMaxSlot = undefined; + this.unfed.clear(); + return undefined; + } + if (this.knownMaxSlot === undefined || slot > this.knownMaxSlot) { + this.knownMaxSlot = slot; + } + return slot; + } +} diff --git a/packages/core/src/runtime/quickjs-slot-report.test.ts b/packages/core/src/runtime/quickjs-slot-report.test.ts new file mode 100644 index 0000000000..6f0714a6e4 --- /dev/null +++ b/packages/core/src/runtime/quickjs-slot-report.test.ts @@ -0,0 +1,293 @@ +/** + * Pins that the QuickJS engine reports its log position on writes and feeds + * the live VM from what the World hands back, the way the node:vm engine + * merges the same pages into its replay log (see the "Who names a position" + * table above `slotSnapshotParams` in helpers.ts). + * + * The QuickJS VM is mocked; the World is a real `@workflow/world-local`, which + * numbers events by slot and implements both the skipped-slot report and the + * inline delta, so what is under test is the engine's side of the exchange: + * which writes name a position, and whether a page on the response reaches the + * VM without an `events.list`. + */ +import { mkdtempSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { + type CreateEventParams, + type CreateEventRequest, + type Event, + eventIdToSlot, + SPEC_VERSION_CURRENT, + type World, +} from '@workflow/world'; +import { createWorld } from '@workflow/world-local'; +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { registerStepFunction } from '../private.js'; +import { + dehydrateStepArguments, + dehydrateStepReturnValue, +} from '../serialization.js'; +import { setWorld } from './world.js'; + +vi.mock('@vercel/functions', () => ({ waitUntil: vi.fn() })); +vi.mock('./get-port-lazy.js', () => ({ + getPortLazy: vi.fn().mockResolvedValue(3000), +})); + +const startQuickJSWorkflow = vi.fn(); +vi.mock('./quickjs-runtime.js', () => ({ + startQuickJSWorkflow: (...args: unknown[]) => startQuickJSWorkflow(...args), +})); + +let counter = 0; + +async function setupRun() { + counter += 1; + const dataDir = mkdtempSync(join(tmpdir(), 'wf-quickjs-slot-report-')); + const world: World = createWorld({ dataDir, tag: `t${counter}` }); + setWorld(world); + + const runInput = await dehydrateStepArguments([], 'run', undefined); + const created = await world.events.create(null, { + eventType: 'run_created', + specVersion: SPEC_VERSION_CURRENT, + eventData: { + deploymentId: 'dpl_quickjs_slot_report', + workflowName: 'workflow', + input: runInput, + }, + }); + const workflowRun = created.run!; + await world.events.create(workflowRun.runId, { + eventType: 'run_started', + specVersion: SPEC_VERSION_CURRENT, + eventData: {}, + } as never); + const loaded = await world.events.list({ + runId: workflowRun.runId, + pagination: { sortOrder: 'asc', limit: 1000 }, + }); + + const writes: Array<{ + request: CreateEventRequest; + params: CreateEventParams | undefined; + }> = []; + /** World calls in order: `write:` and `list`. */ + const sequence: string[] = []; + const originalCreate = world.events.create.bind(world.events); + world.events.create = (async ( + runId: string | null, + request: CreateEventRequest, + params?: CreateEventParams + ) => { + writes.push({ request, params }); + sequence.push(`write:${request.eventType}`); + return originalCreate(runId, request, params); + }) as World['events']['create']; + const originalList = world.events.list.bind(world.events); + const listSpy = vi.fn((...args: Parameters) => { + sequence.push('list'); + return originalList(...args); + }); + world.events.list = listSpy as World['events']['list']; + + const fed: Event[][] = []; + const completedResult = await dehydrateStepReturnValue( + 'done', + workflowRun.runId, + undefined + ); + + return { + world, + workflowRun, + preload: { events: loaded.data, cursor: loaded.cursor }, + writes, + sequence, + listSpy, + fed, + completedResult, + }; +} + +async function runEngine(setup: Awaited>) { + const { runWorkflowWithQuickJS } = await import('./quickjs-entrypoint.js'); + await runWorkflowWithQuickJS({ + workflowCode: '// not evaluated: the VM is mocked', + workflowName: 'workflow', + workflowRun: setup.workflowRun, + preloadedEvents: setup.preload.events, + preloadedEventsComplete: true, + preloadedCursor: setup.preload.cursor, + }); +} + +const slotsOf = (events: readonly Event[]) => + events.map((e) => eventIdToSlot(e.eventId)); + +afterEach(() => { + startQuickJSWorkflow.mockReset(); +}); + +describe('QuickJS engine: log position on writes', () => { + it('names the loaded position on a suspension write and feeds the VM from the response', async () => { + const setup = await setupRun(); + const resumeAt = new Date(Date.now() + 60 * 60 * 1000).toISOString(); + startQuickJSWorkflow.mockResolvedValue({ + result: { + suspended: { + pendingOperations: [ + { + type: 'wait', + correlationId: 'wait_1', + resumeAt, + hasCreatedEvent: false, + }, + ], + }, + }, + continueWithEvents: vi.fn(async (events: Event[]) => { + setup.fed.push(events); + return { completed: { result: setup.completedResult } }; + }), + dispose: vi.fn(), + }); + + await runEngine(setup); + + const waitCreated = setup.writes.find( + (w) => w.request.eventType === 'wait_created' + ); + // The log held run_created and run_started, so the write says it stood + // at position 2. + expect(waitCreated?.params?.eventCount).toBe(2); + + // The VM got the wait_created (position 3) off the write's own response. + // No listing was needed: the preload covered the log and the response + // carried the write. + expect(setup.fed.map(slotsOf)).toEqual([[3]]); + expect(setup.listSpy).not.toHaveBeenCalled(); + + // The run-terminal write names nothing: nothing replays the log after it. + const runCompleted = setup.writes.find( + (w) => w.request.eventType === 'run_completed' + ); + expect(runCompleted).toBeDefined(); + expect(runCompleted?.params?.eventCount).toBeUndefined(); + expect(runCompleted?.params?.sinceCursor).toBeUndefined(); + }); + + it('delivers the span a write skipped over ahead of the write, without a listing', async () => { + const setup = await setupRun(); + const resumeAt = new Date(Date.now() + 60 * 60 * 1000).toISOString(); + startQuickJSWorkflow.mockImplementation(async () => { + // Another writer appends to the log after this invocation loaded it + // and before it writes: the shape of a concurrent replay or an + // out-of-band hook delivery. + await setup.world.events.create(setup.workflowRun.runId, { + eventType: 'wait_created', + specVersion: SPEC_VERSION_CURRENT, + correlationId: 'wait_foreign', + eventData: { resumeAt: new Date(resumeAt) }, + }); + return { + result: { + suspended: { + pendingOperations: [ + { + type: 'wait', + correlationId: 'wait_1', + resumeAt, + hasCreatedEvent: false, + }, + ], + }, + }, + continueWithEvents: vi.fn(async (events: Event[]) => { + setup.fed.push(events); + return { completed: { result: setup.completedResult } }; + }), + dispose: vi.fn(), + }; + }); + + await runEngine(setup); + + const waitCreated = setup.writes.find( + (w) => + w.request.eventType === 'wait_created' && + w.request.correlationId === 'wait_1' + ); + // Still names position 2: the foreign write at 3 was not in its view. + expect(waitCreated?.params?.eventCount).toBe(2); + // The write landed at 4 and the World reported 3 back; the VM receives + // both in position order, foreign event first, from the response alone. + expect(setup.fed.map(slotsOf)).toEqual([[3, 4]]); + expect(setup.listSpy).not.toHaveBeenCalled(); + }); + + it('asks a single inline step for the inline delta and feeds the VM from it', async () => { + const setup = await setupRun(); + const stepName = `step//./quickjs-slot-report//inline${counter}`; + registerStepFunction(stepName, async () => 'ok'); + const stepInput = await dehydrateStepArguments( + { args: [], closureVars: undefined, thisVal: undefined }, + setup.workflowRun.runId, + undefined + ); + startQuickJSWorkflow.mockResolvedValue({ + result: { + suspended: { + pendingOperations: [ + { + type: 'step', + correlationId: 'step_1', + stepId: stepName, + input: stepInput, + hasCreatedEvent: false, + }, + ], + }, + }, + continueWithEvents: vi.fn(async (events: Event[]) => { + setup.fed.push(events); + return { completed: { result: setup.completedResult } }; + }), + dispose: vi.fn(), + }); + + await runEngine(setup); + + const stepCompleted = setup.writes.find( + (w) => w.request.eventType === 'step_completed' + ); + // The terminal write asked for everything after the cursor the engine + // held (the preload's), and named no position: the executor holds no + // log, so the page it gets is the delta, not a report. + expect(stepCompleted?.params?.sinceCursor).toBe(setup.preload.cursor); + expect(stepCompleted?.params?.eventCount).toBeUndefined(); + const stepStarted = setup.writes.find( + (w) => w.request.eventType === 'step_started' + ); + expect(stepStarted?.params?.eventCount).toBeUndefined(); + + // The lazy claim (which this World materializes as step_created + + // step_started, positions 3 and 4) and the terminal at 5 all reached the + // VM off the terminal write's response; no listing ran. + expect(setup.fed.map(slotsOf)).toEqual([[3, 4, 5]]); + expect(setup.fed[0].map((e) => e.eventType)).toEqual([ + 'step_created', + 'step_started', + 'step_completed', + ]); + // The one listing is the loop's probe for cheap progress before it runs + // a step body, which found nothing. After the terminal write, the delta + // carried the log forward and nothing was listed. + const afterTerminal = setup.sequence.slice( + setup.sequence.indexOf('write:step_completed') + 1 + ); + expect(setup.listSpy).toHaveBeenCalledTimes(1); + expect(afterTerminal).not.toContain('list'); + }); +}); From 7cd9c3ea047397df550c7e35123d20a693da1195 Mon Sep 17 00:00:00 2001 From: Peter Wielander Date: Fri, 11 Sep 2026 09:28:09 -0700 Subject: [PATCH 2/3] [core] Deliver only pages to the QuickJS VM, never the created event off a write A create response may carry the created event with its payload still a ref descriptor and telemetry fields stripped (world-vercel does, for every type whose entity the runtime does not read off the response). Feeding that copy of a step_failed gave the VM a step that failed with no error in it; the e2e serialization-failure tests saw a FatalError where a SerializationError was expected. The created event now only advances the position the next write names, and is delivered by the page that covers it or by the next listing. wait_completed, which carries nothing a VM reads, opts in. Co-Authored-By: Claude Fable 5 --- .../core/src/runtime/quickjs-entrypoint.ts | 7 ++- .../core/src/runtime/quickjs-log-view.test.ts | 40 ++++++++++---- packages/core/src/runtime/quickjs-log-view.ts | 40 ++++++++++---- .../src/runtime/quickjs-slot-report.test.ts | 53 +++++++++++++++---- 4 files changed, 108 insertions(+), 32 deletions(-) diff --git a/packages/core/src/runtime/quickjs-entrypoint.ts b/packages/core/src/runtime/quickjs-entrypoint.ts index a8fa112036..884c7e6852 100644 --- a/packages/core/src/runtime/quickjs-entrypoint.ts +++ b/packages/core/src/runtime/quickjs-entrypoint.ts @@ -1071,7 +1071,12 @@ export async function runWorkflowWithQuickJS(params: { ...eventParams, ...logView.snapshotParams(), }); - const absorbed = logView.absorb(result); + // The created event is delivered off the response only when it carries + // no payload a VM reads; every other type waits for a page or a listing, + // which return it with its refs resolved. See QuickJSLogView. + const absorbed = logView.absorb(result, { + deliverEvent: data.eventType === 'wait_completed', + }); if (absorbed.truncated) { runtimeLogger.debug( 'QuickJS runtime: dropped a truncated skipped-slot report', diff --git a/packages/core/src/runtime/quickjs-log-view.test.ts b/packages/core/src/runtime/quickjs-log-view.test.ts index 2179d0b9cd..33c9181a27 100644 --- a/packages/core/src/runtime/quickjs-log-view.test.ts +++ b/packages/core/src/runtime/quickjs-log-view.test.ts @@ -44,9 +44,26 @@ describe('QuickJSLogView', () => { expect(view.takeContiguous()).toEqual([]); }); + it('does not deliver the created event itself, only its position', () => { + const view = new QuickJSLogView([slotEvent(1), slotEvent(2)], 'c2'); + // A create response may carry the event with its payload unresolved, so + // it is not what a listing would deliver. Its position still counts. + view.absorb({ event: slotEvent(3) }); + expect(view.bufferedCount).toBe(0); + expect(view.takeContiguous()).toEqual([]); + expect(view.snapshotParams()).toEqual({ eventCount: 3 }); + // A write whose event carries nothing a VM reads can opt in. + view.absorb({ event: slotEvent(4) }, { deliverEvent: true }); + expect(view.bufferedCount).toBe(1); + // 3 is not in hand, so 4 waits for the listing that delivers 3. + expect(view.takeContiguous()).toEqual([]); + view.markFed([slotEvent(3)]); + expect(slots(view.takeContiguous())).toEqual([4]); + }); + it('holds back a queued event above a position it does not have', () => { const view = new QuickJSLogView([slotEvent(1)], 'c1'); - view.absorb({ event: slotEvent(3) }); + view.absorb({ event: slotEvent(3) }, { deliverEvent: true }); // Position 2 is not in hand, so 3 cannot be fed: the VM must not see it // before whatever lands at 2. expect(view.takeContiguous()).toEqual([]); @@ -56,22 +73,23 @@ describe('QuickJSLogView', () => { expect(slots(view.takeContiguous())).toEqual([3]); }); - it('drops a truncated report whole but keeps the write itself', () => { + it('drops a truncated report whole', () => { const view = new QuickJSLogView([slotEvent(1)], 'c1'); const absorbed = view.absorb({ event: slotEvent(10), events: [slotEvent(2), slotEvent(3)], hasMore: true, }); - expect(absorbed).toEqual({ queued: 1, truncated: true }); - // 10 is queued and waits for a listing to fill 2..9. - expect(view.bufferedCount).toBe(1); - expect(view.takeContiguous()).toEqual([]); + expect(absorbed).toEqual({ queued: 0, truncated: true }); + expect(view.bufferedCount).toBe(0); + // The write's position is still known, so the next write names 10 and a + // listing fills 2..10. + expect(view.snapshotParams()).toEqual({ eventCount: 10 }); }); it('drops an event a listing already delivered', () => { const view = new QuickJSLogView([slotEvent(1)], 'c1'); - view.absorb({ event: slotEvent(3) }); + view.absorb({ event: slotEvent(3) }, { deliverEvent: true }); // The listing raced ahead and delivered 2 and 3 itself. view.markFed([slotEvent(2), slotEvent(3)]); expect(view.bufferedCount).toBe(0); @@ -106,8 +124,8 @@ describe('QuickJSLogView', () => { it('keeps the cursor when the delta leaves a hole below its end', () => { const view = new QuickJSLogView([slotEvent(1)], 'c1'); - // Position 4 is known (an own write) but 2 and 3 are not in hand. - view.absorb({ event: slotEvent(4) }); + // Position 4 is in hand (an own write) but 2 and 3 are not. + view.absorb({ event: slotEvent(4) }, { deliverEvent: true }); const advanced = view.absorbDelta('c1', { events: [slotEvent(5)], cursor: 'c5', @@ -121,7 +139,7 @@ describe('QuickJSLogView', () => { it('turns tracking off for good on an id that is not a position', () => { const view = new QuickJSLogView([slotEvent(1)], 'c1'); - view.absorb({ event: slotEvent(2) }); + view.absorb({ event: slotEvent(2) }, { deliverEvent: true }); expect(view.bufferedCount).toBe(1); view.absorb({ event: { ...slotEvent(3), eventId: 'evnt_01ARZ3NDEKTSV4RRFFQ69G5FAV' }, @@ -131,7 +149,7 @@ describe('QuickJSLogView', () => { expect(view.bufferedCount).toBe(0); expect(view.takeContiguous()).toEqual([]); // Nothing queues again afterwards. - view.absorb({ event: slotEvent(4) }); + view.absorb({ event: slotEvent(4) }, { deliverEvent: true }); expect(view.bufferedCount).toBe(0); }); }); diff --git a/packages/core/src/runtime/quickjs-log-view.ts b/packages/core/src/runtime/quickjs-log-view.ts index 5bbc3a8f5c..d5a530e5f3 100644 --- a/packages/core/src/runtime/quickjs-log-view.ts +++ b/packages/core/src/runtime/quickjs-log-view.ts @@ -25,6 +25,18 @@ import { type Event, eventIdToSlot, FIRST_EVENT_SLOT } from '@workflow/world'; * - `cursor`, the read position for `events.list`, advanced by list pages and * by a complete inline delta. * + * Only PAGES are queued for delivery, never the created event a write returns + * on its own. A World reads a report or a delta the way it reads a listing, + * with payload refs resolved, so those events are what a listing would have + * delivered. The created event on a create response is not: a World may hand + * it back with its payload still a ref descriptor and telemetry fields + * stripped (the Vercel World does, for every type whose entity the runtime + * does not read off the response), and a VM given that copy of a `step_failed` + * sees a step that failed with no error in it. The created event still counts + * toward the position the next write names; it is delivered by the page that + * covers it or by the next listing. `deliverEvent` opts a write in when its + * event carries nothing a VM reads (`wait_completed`). + * * Positions come from slot-numbered event ids. A log whose ids are not slots * (a World on the old id scheme, or a mocked World) turns tracking off for the * rest of the invocation: no `eventCount` is sent, nothing is buffered, and the @@ -92,23 +104,31 @@ export class QuickJSLogView { } /** - * Absorb a write's response: the committed event, and the skipped-slot - * report or inline delta a World attached to it. + * Absorb a write's response: note the committed event's position, and + * queue the skipped-slot report a World attached to it. * * A truncated page (`hasMore`) is dropped whole, the same policy as * `absorbSkippedSlotReport` in the node engine: it covers a span of * positions but carries only some of them, and queuing part of a span * would have this view claim, on its next write, to have seen positions it - * never received. The committed event itself is always taken. + * never received. The committed event is queued only under `deliverEvent` + * (see the class doc for why not by default); its position always counts. */ - absorb(result: { - event?: Event; - events?: readonly Event[]; - hasMore?: boolean; - }): { queued: number; truncated: boolean } { + absorb( + result: { + event?: Event; + events?: readonly Event[]; + hasMore?: boolean; + }, + options: { deliverEvent?: boolean } = {} + ): { queued: number; truncated: boolean } { let queued = 0; - if (result.event && this.queue(result.event)) { - queued++; + if (result.event) { + if (options.deliverEvent === true) { + if (this.queue(result.event)) queued++; + } else { + this.slotOf(result.event); + } } const page = result.events ?? []; if (page.length === 0) { diff --git a/packages/core/src/runtime/quickjs-slot-report.test.ts b/packages/core/src/runtime/quickjs-slot-report.test.ts index 6f0714a6e4..295920140c 100644 --- a/packages/core/src/runtime/quickjs-slot-report.test.ts +++ b/packages/core/src/runtime/quickjs-slot-report.test.ts @@ -131,7 +131,7 @@ afterEach(() => { }); describe('QuickJS engine: log position on writes', () => { - it('names the loaded position on a suspension write and feeds the VM from the response', async () => { + it('names the loaded position on a suspension write and lists for the event itself', async () => { const setup = await setupRun(); const resumeAt = new Date(Date.now() + 60 * 60 * 1000).toISOString(); startQuickJSWorkflow.mockResolvedValue({ @@ -163,11 +163,19 @@ describe('QuickJS engine: log position on writes', () => { // at position 2. expect(waitCreated?.params?.eventCount).toBe(2); - // The VM got the wait_created (position 3) off the write's own response. - // No listing was needed: the preload covered the log and the response - // carried the write. + // The VM got the wait_created (position 3) from a listing, not off the + // write's response: a create response may carry the event with its + // payload unresolved, so the created event is never fed from there. The + // listing read forward from the preload's cursor rather than from the + // top of the log. expect(setup.fed.map(slotsOf)).toEqual([[3]]); - expect(setup.listSpy).not.toHaveBeenCalled(); + expect(setup.listSpy.mock.calls[0][0].pagination?.cursor).toBe( + setup.preload.cursor + ); + // No listing before the write: the preload was the whole log. + expect(setup.sequence.indexOf('list')).toBeGreaterThan( + setup.sequence.indexOf('write:wait_created') + ); // The run-terminal write names nothing: nothing replays the log after it. const runCompleted = setup.writes.find( @@ -206,7 +214,24 @@ describe('QuickJS engine: log position on writes', () => { }, continueWithEvents: vi.fn(async (events: Event[]) => { setup.fed.push(events); - return { completed: { result: setup.completedResult } }; + // Suspend again until the VM has been given the wait it asked for. + const sawOwnWait = setup.fed + .flat() + .some((e) => e.correlationId === 'wait_1'); + return sawOwnWait + ? { completed: { result: setup.completedResult } } + : { + suspended: { + pendingOperations: [ + { + type: 'wait', + correlationId: 'wait_1', + resumeAt, + hasCreatedEvent: true, + }, + ], + }, + }; }), dispose: vi.fn(), }; @@ -221,10 +246,18 @@ describe('QuickJS engine: log position on writes', () => { ); // Still names position 2: the foreign write at 3 was not in its view. expect(waitCreated?.params?.eventCount).toBe(2); - // The write landed at 4 and the World reported 3 back; the VM receives - // both in position order, foreign event first, from the response alone. - expect(setup.fed.map(slotsOf)).toEqual([[3, 4]]); - expect(setup.listSpy).not.toHaveBeenCalled(); + // The write landed at 4 and the World reported 3 back. The foreign event + // is fed off the response, ahead of anything else; the write itself + // (which this World's report excludes) follows from a listing. The VM + // saw them in position order. + expect(setup.fed.map(slotsOf)).toEqual([[3], [4]]); + // The foreign event was fed before any listing ran. + expect(setup.sequence.indexOf('write:wait_created')).toBeLessThan( + setup.sequence.indexOf('list') + ); + expect(setup.listSpy.mock.calls[0][0].pagination?.cursor).toBe( + setup.preload.cursor + ); }); it('asks a single inline step for the inline delta and feeds the VM from it', async () => { From ebb53375e9ccad88864066982c4a2a817aa07056 Mon Sep 17 00:00:00 2001 From: Peter Wielander Date: Fri, 11 Sep 2026 13:30:59 -0700 Subject: [PATCH 3/3] [core] Address review: mirror the node inline-delta gate, hook_created delta, doc fixes - The QuickJS inline-delta gate now matches the node engine's requestInlineDelta: sole outstanding step (no overflow, unserializable, or earlier-invocation sibling) and no pending wait, with the reasoning for the wait half written down. - A lone hook_created asks for the inline delta (sinceCursor) with the same single-hook gate as the node suspension handler, and the createEvent seam takes a delta response through the delta path. The hook_created row now holds for both engines. - Feed comment no longer claims a report saves the pre-step listing; only the delta does, since a report does not advance the cursor. - Restore the JSDoc gutter on the run-terminal table row in helpers.ts. Co-Authored-By: Claude Fable 5 --- .../docs/v5/how-it-works/event-sourcing.mdx | 2 +- packages/core/src/runtime/helpers.ts | 4 +- .../core/src/runtime/quickjs-entrypoint.ts | 127 ++++++++++++++---- .../src/runtime/quickjs-slot-report.test.ts | 40 ++++++ 4 files changed, 143 insertions(+), 30 deletions(-) diff --git a/docs/content/docs/v5/how-it-works/event-sourcing.mdx b/docs/content/docs/v5/how-it-works/event-sourcing.mdx index 5539627c57..92b43b335b 100644 --- a/docs/content/docs/v5/how-it-works/event-sourcing.mdx +++ b/docs/content/docs/v5/how-it-works/event-sourcing.mdx @@ -281,7 +281,7 @@ Both save a reload, so the rule for which writes send them is not about the even | `run_created` | `start()` | Nothing | No log exists yet. | | `run_started` | Replay, first write of a delivery | `eventCount` when a log is loaded; no cursor | The `run_started` response already carries the full log as a preload. | | `step_created`, `wait_created`, `hook_disposed`, `attr_set` | Suspension handler | `eventCount` | The handler keeps writing from this log and the replay resumes from it. | -| `hook_created` | Suspension handler | `eventCount` and `sinceCursor` (one hook write per suspension) | The delta is how a hook whose payload already arrived resolves without another invocation. | +| `hook_created` | Suspension handler | `eventCount` and `sinceCursor` (only when the suspension creates exactly one hook) | The delta is how a hook whose payload already arrived resolves without another invocation. Two creates diffing against one cursor would give two deltas of which only the first could be taken. | | `hook_received` | Replay resuming a hook | `eventCount` | The replay continues from the log right after. | | `hook_received` | Webhook or `resumeHook()` | Nothing | Out-of-band. It enqueues a delivery that loads the log. | | `wait_completed` | Replay completing an elapsed wait | `eventCount` | The replay continues from the log right after. | diff --git a/packages/core/src/runtime/helpers.ts b/packages/core/src/runtime/helpers.ts index c9c4cf90e6..d5a1f54a4a 100644 --- a/packages/core/src/runtime/helpers.ts +++ b/packages/core/src/runtime/helpers.ts @@ -1063,7 +1063,7 @@ export interface SlotSnapshotParams { * | `run_started` | replay loop, first write of a delivery | `eventCount` when a log is loaded; no cursor, the `run_started` preload owns the response fields | the preload already returns the full log | * | `step_created`, `wait_created`, `hook_disposed`, `attr_set` | suspension handler (`createGuarded`) | `eventCount` | the handler keeps writing from, and resumes replay from, this log | * | `step_created` | queue-consumer re-ensure (raw) | nothing | runs from a queue message, no log | - * | `hook_created` | suspension handler | `eventCount` + `sinceCursor` (at most one hook write per suspension) | the delta is how an already-received hook resolves without a re-invocation | + * | `hook_created` | suspension handler (node), `dispatchPendingOps` (QuickJS) | `eventCount` + `sinceCursor` (only when the suspension creates exactly one hook) | the delta is how an already-received hook resolves without a re-invocation; two creates against one cursor would give two deltas of which only the first could be taken | * | `hook_received` | replay loop lazy resume; handler in-suspension resume | `eventCount` (preload owns the cursor fields) | both replay from the log right after | * | `hook_received` | webhook / `resumeHook` (raw) | nothing | out-of-band, enqueues a delivery that loads the log | * | `wait_completed` | replay loop timer path | `eventCount` | the loop keeps replaying from the log | @@ -1072,7 +1072,7 @@ export interface SlotSnapshotParams { * | `step_completed`, `step_failed` | step executor | `sinceCursor` only, outside turbo, single inline step | this is where the inline execution loop gets its log back; in turbo or from a queued delivery a fresh replay loads it anyway | * | `step_failed` | suspension handler, unserializable input | `eventCount` | the handler holds a log; rare, and follows a `step_created` that drew the page | * | `attr_set` | `setAttributes()` API (raw) | nothing | out-of-band | - | `run_completed`, `run_failed`, `run_cancelled` | replay loop, replay budget, `runs.cancel` | `eventCount` where the node loop's seam stamps it (the QuickJS engine sends nothing); never `sinceCursor` (`deltaRequestCursor` excludes terminal types) | terminal: nothing replays the log afterwards, so a World reads no page for them | + * | `run_completed`, `run_failed`, `run_cancelled` | replay loop, replay budget, `runs.cancel` | `eventCount` where the node loop's seam stamps it (the QuickJS engine sends nothing); never `sinceCursor` (`deltaRequestCursor` excludes terminal types) | terminal: nothing replays the log afterwards, so a World reads no page for them | * * Batched writes (`events.createBatch`) are outside all of this: a batch carries * no per-event position and gets no page. The user-facing version of this table diff --git a/packages/core/src/runtime/quickjs-entrypoint.ts b/packages/core/src/runtime/quickjs-entrypoint.ts index 884c7e6852..4a1c184179 100644 --- a/packages/core/src/runtime/quickjs-entrypoint.ts +++ b/packages/core/src/runtime/quickjs-entrypoint.ts @@ -238,6 +238,19 @@ async function dispatchPendingOps(params: { * the run is ending and nothing reads its log afterwards. */ createEvent: EventCreator; + /** + * The cursor this invocation's log was read to, when the caller wants a lone + * `hook_created` to ask for the inline delta against it (`sinceCursor`). + * The hook's awaiters are settled by the event that write commits and by + * nothing else (a `hook_created`, or the `hook_conflict` a claimed token + * commits instead), so the delta hands the VM that event, plus anything + * another writer landed meanwhile, without a listing. Same gate as the node + * engine's `hookDeltaCursor`: asked for only when exactly one hook needs + * creating, since two creates diffing against one cursor would produce two + * deltas of which only the first could be taken. Omitted by the terminal + * drain. + */ + deltaCursor?: string; workflowRun: WorkflowRun; encryptionKey: RunPayloadKeys | undefined; pendingOperations: PendingOperation[]; @@ -338,6 +351,11 @@ async function dispatchPendingOps(params: { // same pattern as an elapsed wait. let createdAttributeEvent = false; const opsPromises: Promise[] = []; + const hooksNeedingCreation = pendingOperations.filter( + (op) => op.type === 'hook' && !op.hasCreatedEvent + ).length; + const hookDeltaCursor = + hooksNeedingCreation === 1 ? params.deltaCursor : undefined; const processHookOp = async (hook: PendingHook): Promise => { runtimeLogger.debug('QuickJS runtime: processing hook op', { @@ -368,26 +386,31 @@ async function dispatchPendingOps(params: { typeof hook.metadata === 'undefined' ? undefined : await encryptSerializedData(hook.metadata, encryptionKey); - const result = await createEvent({ - eventType: 'hook_created', - specVersion: SPEC_VERSION_CURRENT, - correlationId: hook.correlationId, - eventData: { - token: hook.token, - tokenRetentionUntil: - hook.tokenRetentionUntil === undefined - ? undefined - : new Date(hook.tokenRetentionUntil), - metadata: encryptedMetadata, - // Always include isWebhook explicitly. Worlds default it to - // `true` when absent, which would break the public webhook - // endpoint's 404 guard for hooks created via createHook(). - isWebhook: hook.isWebhook, - // System hooks (AbortController) are exempt from user - // token namespace conflict checks. - ...(hook.isSystem ? { isSystem: true } : {}), - } as any, - }); + const result = await createEvent( + { + eventType: 'hook_created', + specVersion: SPEC_VERSION_CURRENT, + correlationId: hook.correlationId, + eventData: { + token: hook.token, + tokenRetentionUntil: + hook.tokenRetentionUntil === undefined + ? undefined + : new Date(hook.tokenRetentionUntil), + metadata: encryptedMetadata, + // Always include isWebhook explicitly. Worlds default it to + // `true` when absent, which would break the public webhook + // endpoint's 404 guard for hooks created via createHook(). + isWebhook: hook.isWebhook, + // System hooks (AbortController) are exempt from user + // token namespace conflict checks. + ...(hook.isSystem ? { isSystem: true } : {}), + } as any, + }, + hookDeltaCursor !== undefined + ? { sinceCursor: hookDeltaCursor } + : undefined + ); // If storage detected a real token conflict with another // workflow's hook, re-queue so the workflow handler can @@ -822,7 +845,8 @@ async function dispatchPendingOps(params: { * pair for an unserializable input) carry `eventCount`, and the page a * World hands back is queued for the live VM through {@link QuickJSLogView}. * - A single inline step's terminal write asks for the inline delta - * (`sinceCursor`) through `executeStep`, and the delta is queued the same + * (`sinceCursor`) through `executeStep`, and so does a lone `hook_created` + * (see `dispatchPendingOps.deltaCursor`); the delta is queued the same * way. The step executor's other writes carry nothing: it holds no log. * - Run-terminal writes (`run_completed`, `run_failed`) and the terminal drain * of pending ops carry nothing: nothing reads the log afterwards. @@ -1071,6 +1095,28 @@ export async function runWorkflowWithQuickJS(params: { ...eventParams, ...logView.snapshotParams(), }); + if ( + typeof eventParams?.sinceCursor === 'string' && + result.events !== undefined + ) { + // The write asked for the inline delta and got one: everything after + // the cursor, this write included, read with refs resolved. Taken + // through the delta path so the cursor moves with it when that is + // safe; the created event's position is noted either way. + logView.absorb({ event: result.event }); + const advanced = logView.absorbDelta(eventParams.sinceCursor, { + events: result.events, + cursor: result.cursor ?? null, + hasMore: result.hasMore ?? false, + }); + wfdiag('inline_delta_absorbed', { + eventType: data.eventType, + events: result.events.length, + hasMore: result.hasMore ?? false, + cursorAdvanced: advanced, + }); + return result; + } // The created event is delivered off the response only when it carries // no payload a VM reads; every other type waits for a page or a listing, // which return it with its refs resolved. See QuickJSLogView. @@ -1455,6 +1501,9 @@ export async function runWorkflowWithQuickJS(params: { namespace, nextTraceCarrier, createEvent, + ...(logView.tracking && typeof logView.logCursor === 'string' + ? { deltaCursor: logView.logCursor } + : {}), pendingOperations: opsToDispatch, skipStepCreation: inlineClaimCids, queueStepCids: new Set(overflowSteps.map((s) => s.correlationId)), @@ -1532,8 +1581,14 @@ export async function runWorkflowWithQuickJS(params: { // 2. Cheap progress first: feed newly recorded events into the live // VM before blocking on step bodies. What the writes above handed back - // is delivered first; a listing runs only when that does not reach the - // next position. + // is delivered first, and a listing runs when the queue does not reach + // the next position. A report changes what is fed first, not whether + // this listing happens: after a queued page is fed, this branch + // `continue`s, and the next iteration finds the queue empty and lists + // from a cursor a report does not advance (re-reading the reported + // span, deduped on `seenEventIds`). Only the inline delta below, which + // does advance the cursor, saves a listing outright. Same shape as the + // node engine. { const queued = takeQueuedEvents(); const newEvents = @@ -1725,12 +1780,30 @@ export async function runWorkflowWithQuickJS(params: { // Inline delta: a single inline step's terminal write asks the World // for everything after the cursor this view holds, so the step's own // events (and anything interleaved) arrive on the write's response and - // the feed below needs no listing. Only for a batch of one, as in the - // node engine: several steps would each get a delta from the same - // cursor, and only one could be taken. Not requested when the log has - // no cursor to name (tracking off, or nothing read yet). + // the feed below needs no listing. Same gate as the node engine's + // `requestInlineDelta` (runtime.ts), translated to this loop's terms: + // + // - This step is the only step outstanding: no overflow sibling queued + // this iteration, no unserializable sibling, no step from an earlier + // invocation handed to the queue above. Several writers each diffing + // against the same cursor would produce deltas of which only the + // first could be taken. + // - No wait is pending. A `wait_completed` is a resolution the + // workflow is waiting on rather than an event it can observe one + // iteration late, so a delta that predates it would settle the + // sleep from a view that does not hold its completion; the listing + // after the step is what reads it in order. + // - The log has a cursor to name (tracking on, something read). + const hasPendingWait = pendingOperations.some( + (op) => + op.type === 'wait' && + !completedWaitIds2.has((op as PendingWait).correlationId) + ); const inlineDeltaSinceCursor = + stepOps.length === 1 && + freshSteps.length === 1 && inlineCandidates.length === 1 && + !hasPendingWait && logView.tracking && typeof logView.logCursor === 'string' ? logView.logCursor diff --git a/packages/core/src/runtime/quickjs-slot-report.test.ts b/packages/core/src/runtime/quickjs-slot-report.test.ts index 295920140c..519b85a0a6 100644 --- a/packages/core/src/runtime/quickjs-slot-report.test.ts +++ b/packages/core/src/runtime/quickjs-slot-report.test.ts @@ -260,6 +260,46 @@ describe('QuickJS engine: log position on writes', () => { ); }); + it('asks a lone hook_created for the inline delta and feeds the VM from it', async () => { + const setup = await setupRun(); + startQuickJSWorkflow.mockResolvedValue({ + result: { + suspended: { + pendingOperations: [ + { + type: 'hook', + correlationId: 'hook_1', + token: 'tok-1', + isWebhook: false, + hasCreatedEvent: false, + }, + ], + }, + }, + continueWithEvents: vi.fn(async (events: Event[]) => { + setup.fed.push(events); + return { completed: { result: setup.completedResult } }; + }), + dispose: vi.fn(), + }); + + await runEngine(setup); + + const hookCreated = setup.writes.find( + (w) => w.request.eventType === 'hook_created' + ); + // The one hook create of the suspension names its position and asks for + // the delta against the cursor the engine held, as the node engine's + // suspension handler does for a single-hook suspension. + expect(hookCreated?.params?.eventCount).toBe(2); + expect(hookCreated?.params?.sinceCursor).toBe(setup.preload.cursor); + // The hook_created (position 3) reached the VM off the delta; no listing + // ran before it was fed. + expect(setup.fed.map(slotsOf)).toEqual([[3]]); + expect(setup.fed[0][0].eventType).toBe('hook_created'); + expect(setup.sequence.indexOf('list')).toBe(-1); + }); + it('asks a single inline step for the inline delta and feeds the VM from it', async () => { const setup = await setupRun(); const stepName = `step//./quickjs-slot-report//inline${counter}`;