skills/vercel__workflow
1
0
Fork 0
mirror of https://github.com/vercel/workflow.git synced 2026-09-14 19:59:43 +08:00
Code Issues Packages Projects Releases Wiki Activity

[core] Stop sending a slot snapshot on step executor writes (#4096)

Browse Source 
The only thing a World does with `eventCount` is bump-and-report: when the
write lands above the position named, it reads the events in between and
returns them so the writer can merge them without a second round-trip. The
replay loop and the suspension handler merge that page into their loaded
log. The step executor has no log to merge into, so it took the page's
highest position and discarded the rest.

In production that discarded read fell on a third of all `step_started`
writes (10.8M of 13.4M skipped-slot report reads per day were on executor
event types), each a strongly consistent DynamoDB query on the run
partition with resolved refs, on the response path. This removes the
executor's `knownSlot` / `observeSlot` machinery, the `slotSnapshot`
executor param, and the `batchCommittedSlotCeiling` the suspension handler
computed only to seed it. The loop's and the suspension handler's own
snapshots are unchanged; they consume their reports.

The World contract already describes omitting the count for a caller with
no loaded log to be stale against; the executor now matches it.

Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Pranay Prakash
2026-09-11 12:57:25 -07:00
committed by GitHub
parent e00b1a57ee
commit 6cc851c342
8 changed files with 48 additions and 186 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.changeset/drop-executor-slot-snapshot.md
+6
View File
@@ -0,0 +1,6 @@
---
'@workflow/core': patch
'@workflow/world': patch
---
Stop sending a slot snapshot (`eventCount`) on step executor writes, so a World no longer reads and returns a skipped-slot event page that the executor only discards.
packages/core/src/runtime.test.ts
+6 -5
View File
@@ -2802,17 +2802,18 @@ describe('workflowEntrypoint inline-delta gate with open hooks', () => {
// The handler responds normally: the rejection restarts the replay inside
// this delivery, never a run_failed.
expect(res.status).toBe(204);
// Step B's claim was issued from a loaded (non-empty) log, so it named the
// position it was decided against. (The very first batch of a run loads an
// empty log and has no position to name; reporting is best-effort there,
// matching the suspension creates.)
// Step B's claim names no log position: the executor has no loaded log
// to merge a skipped-slot report into, so it sends no `eventCount` and
// the World reads no page for it. The fence this test exercises does not
// depend on one either; a World that rejects a claim does so from its own
// view of the log, as the scripted rejection above does.
const rejectedClaim = eventsCreate.mock.calls.find(
(c) =>
(c[1] as any).eventType === 'step_started' &&
((c[1] as any).eventData as { stepName?: string })?.stepName ===
'deltaGateStepB'
);
expect(typeof (rejectedClaim?.[2] as any)?.eventCount).toBe('number');
expect((rejectedClaim?.[2] as any)?.eventCount).toBeUndefined();
// The fenced claim's body never ran: step B executes exactly once, on the
// restarted replay whose claim the backend accepted.
expect(deltaGateBodyRuns).toEqual(['B']);
packages/core/src/runtime.ts
-31
View File
@@ -4334,36 +4334,6 @@ export function workflowEntrypoint(
retained: servedByRetainedSession,
});
// Slot snapshot for the inline step_started claims: the
// lazy claim is the first durable write of a hot-path
// step (its step_created is deferred), so without a
// snapshot it would name no position at all and a stale
// replay could claim (and commit) a step scheduled off
// a view that misses an event it never loaded.
//
// Taken here rather than inside the executor because
// this is the view the scheduling decision was made
// against. The executor advances from it as its own
// writes land; see `slotSnapshot` in step-executor.
const loadedSlotSnapshot = slotSnapshotParams(
eventLog.events
);
// The batched fan-out's own events are not in the
// loaded log yet (the next iteration reloads), but
// this invocation wrote them, so fold the batch's
// ceiling in, or every inline terminal write would
// name a pre-batch position and be answered with a
// skipped-slot report echoing the events this
// suspension just committed.
const batchSlotCeiling =
suspensionResult.batchCommittedSlotCeiling;
const inlineClaimSnapshot =
batchSlotCeiling !== undefined &&
batchSlotCeiling >
(loadedSlotSnapshot.eventCount ?? 0)
? { eventCount: batchSlotCeiling }
: loadedSlotSnapshot;
// TTR: consumed by this batch. Every step is handed
// the SAME tracking object and its one-shot
// `reported` latch picks the single step that
@@ -4464,7 +4434,6 @@ export function workflowEntrypoint(
// See suppressOptimisticStart above.
suppressOptimisticStart,
runReadyBarrier,
slotSnapshot: inlineClaimSnapshot,
...(stepIndex === 0 &&
(s.lazyStepInput !== undefined ||
s.preclaimedStart !== undefined) &&
packages/core/src/runtime/step-executor.test.ts
+10 -24
View File
@@ -523,8 +523,6 @@ describe('executeStep — compute instance stamping', () => {
// persist — so observe the call itself rather than the stored event.
const createSpy = vi.spyOn(world.events, 'create');
const slotSnapshot = { eventCount: 7 };
await executeStep({
world,
workflowRunId: runId,
@@ -533,7 +531,6 @@ describe('executeStep — compute instance stamping', () => {
requestId: 'req_step_executor',
stepId,
stepName,
slotSnapshot,
});
const started = createSpy.mock.calls.filter(
@@ -544,8 +541,9 @@ describe('executeStep — compute instance stamping', () => {
requestId: 'req_step_executor',
computeInstanceId: COMPUTE_INSTANCE_ID,
});
// All dimensions ride the same params object and neither may clobber another.
expect(started[0]?.[2]?.eventCount).toBe(slotSnapshot.eventCount);
// An executor write names no log position: it has no log to merge a
// skipped-slot report into, so it must not ask the World to read one.
expect(started[0]?.[2]?.eventCount).toBeUndefined();
});
it('stamps provenance when a lazy unregistered step is materialized', async () => {
@@ -636,10 +634,12 @@ describe('executeStep — compute instance stamping', () => {
expect(started?.[2]?.requestId).toBeUndefined();
});
it('advances the snapshot it sends as its own writes land', async () => {
// The executor writes twice for one step. If the second write still named
// the position its caller scheduled against, the World would report the
// first one back to it on every step, forever.
it('sends no slot snapshot on any of its writes', async () => {
// The only thing a World does with `eventCount` is bump-and-report: read
// the events between the named position and the committed one and hand
// them back. The executor has no loaded log to merge that page into, so
// naming a position would make the World read a page nobody consumes, on
// every contended step_started.
const world = makeWorld();
const stepName = uniqueStepName();
const { runId, stepId } = await setupRunningStep({
@@ -648,13 +648,6 @@ describe('executeStep — compute instance stamping', () => {
onBody: () => {},
});
// The position the caller would have scheduled against, taken from the log
// rather than written down, so the seed stays below the slots the executor
// is about to commit at. Seeding it above them would leave `observeSlot`
// with nothing to raise and the test would pass without exercising it.
const { data: seeded } = await world.events.list({ runId });
const scheduledAt = seeded.length;
const createSpy = vi.spyOn(world.events, 'create');
await executeStep({
@@ -664,18 +657,11 @@ describe('executeStep — compute instance stamping', () => {
workflowStartedAt: Date.now(),
stepId,
stepName,
slotSnapshot: { eventCount: scheduledAt },
});
// world-local mints slots for a run created on this scheme, so every write
// reads back as a position and each one has to name the position its
// predecessor landed on.
const counts = createSpy.mock.calls.map((call) => call[2]?.eventCount);
expect(counts.length).toBeGreaterThan(1);
expect(counts[0]).toBe(scheduledAt);
for (let i = 1; i < counts.length; i++) {
expect(counts[i]).toBeGreaterThan(counts[i - 1] as number);
}
expect(counts.every((count) => count === undefined)).toBe(true);
});
});
packages/core/src/runtime/step-executor.ts
+20 -86
View File
@@ -25,7 +25,6 @@ import type {
World,
} from '@workflow/world';
import {
requireEventSlot,
SPEC_VERSION_CURRENT,
SPEC_VERSION_SUPPORTS_COMPRESSION,
} from '@workflow/world';
@@ -57,11 +56,7 @@ import {
isOptimisticInlineStartExplicitlyDisabled,
} from './constants.js';
import { getPortLazy } from './get-port-lazy.js';
import {
maxEventSlot,
memoizeEncryptionKey,
type SlotSnapshotParams,
} from './helpers.js';
import { memoizeEncryptionKey } from './helpers.js';
import { ReplayRecoveryReporter } from './replay-recovery-reporter.js';
import {
computeResumeTtrAttributes,
@@ -220,26 +215,6 @@ export interface StepExecutorParams {
* handler is the sole inline writer for the run on this iteration.
*/
inlineDeltaSinceCursor?: string;
/**
* How much of the run's log the caller's replay had loaded when it scheduled
* this step, as the highest slot that log occupies. Seeds the snapshot every
* write this executor makes carries; each committed event advances it to its
* own slot, so a later write never names a position that predates an earlier
* one from the same step.
*
* It matters most on the lazy inline path, where the `step_started` claim is
* the step's FIRST durable write (its `step_created` is deferred): without a
* seed the claim would name no position at all, and a replay working from a
* stale view could claim (and then commit) a step scheduled without
* observing an event it never loaded.
*
* A World that fences rejects a stale claim with `PreconditionFailedError`
* (412); executeStep does NOT translate that rejection (re-claiming in place
* would still commit the stale schedule), so it propagates for the caller to
* abandon the batch and restart its replay. Undefined for a caller with
* nothing loaded.
*/
slotSnapshot?: SlotSnapshotParams;
/**
* Suppress optimistic inline start for this step regardless of
* `WORKFLOW_OPTIMISTIC_INLINE_START` / `forceOptimisticStart`: take the
@@ -405,59 +380,22 @@ export async function executeStep(
(params.runSpecVersion ?? 0) >= SPEC_VERSION_SUPPORTS_COMPRESSION;
const replayRecoveryReporter =
params.replayRecoveryReporter ?? ReplayRecoveryReporter.inert();
/**
* The highest log slot this executor knows about, seeded from the view its
* caller scheduled the step against and advanced by every event it commits.
*
* Advancing is what keeps the snapshot honest across a step's own writes. A
* step commits `step_started` and then `step_completed`; if the second still
* named the caller's original position, the World would report the first one
* back as an event this writer had not seen, on every step, forever.
*
* This reads a report for its highest position and then discards it, where
* the replay loop and the suspension handler merge theirs with
* `absorbSkippedSlotReport`. That is the difference between the callers, not
* an oversight: an executor holds no loaded log to merge into. It runs from a
* queued delivery whose only view of the log is the integer its caller passed
* in, so the position is the entire value the report has to it. Whoever
* replays next loads the log and gets the events themselves.
*/
let knownSlot = params.slotSnapshot?.eventCount;
const observeSlot = (result: { event?: Event; events?: Event[] }): void => {
if (knownSlot === undefined) {
// The caller scheduled this step without naming a position, so there is
// no snapshot to advance and the writes below send none. Not the same as
// a run without positions: every run has them, this executor was not
// told which one it started from.
return;
}
const observed: number[] = [];
if (result.event) {
observed.push(requireEventSlot(result.event.eventId));
}
const reported = maxEventSlot(result.events ?? []);
if (reported !== undefined) {
observed.push(reported);
}
for (const slot of observed) {
if (slot > knownSlot) {
knownSlot = slot;
}
}
};
// Executor writes carry no slot snapshot (`CreateEventParams.eventCount`).
// The only thing a World does with one is bump-and-report: when the write
// lands above the position named, it reads the events in between and hands
// them back. The replay loop and the suspension handler merge that page into
// their loaded log; this executor has no log to merge into, so the page was
// read only to be discarded, and in production that read fell on a third of
// all `step_started` writes. Omitting the count is the documented shape for
// a caller with no loaded log to be stale against, and the conditional
// write on (runId, correlationId) remains the ownership fence.
const createEvent = async <T extends CreateEventRequest>(
data: T,
eventParams?: CreateEventParams
) => {
const result = await replayRecoveryReporter.withEventCreate(
knownSlot === undefined
? eventParams
: { eventCount: knownSlot, ...eventParams },
(p) => world.events.create(workflowRunId, data, p)
) =>
replayRecoveryReporter.withEventCreate(eventParams, (p) =>
world.events.create(workflowRunId, data, p)
);
observeSlot(result);
return result;
};
// `step_started` identifies the invocation that performed this attempt.
// Keep request and compute provenance independent: world-vercel serializes
@@ -766,9 +704,7 @@ export async function executeStep(
}
let step: StartedStep;
// Params for the `step_started` create on either path below. The slot
// snapshot is not spread here: `createEvent` attaches it to every write,
// this one included.
// Params for the `step_started` create on either path below.
const startEventParams = stepStartedEventParams;
// `Date.now()` taken immediately before the `step_started` create is
// issued (either path below); anchors RSFS's end point. See
@@ -842,10 +778,9 @@ export async function executeStep(
: {}),
},
},
// Guard the claim; see StepExecutorParams.slotSnapshot. A
// stale (412) rejection surfaces via reconcileOptimisticStart as a
// non-translatable error: the body result is discarded and the
// rejection propagates to the caller.
// A 412 rejection from a fencing World surfaces via
// reconcileOptimisticStart as a non-translatable error: the body
// result is discarded and the rejection propagates to the caller.
startEventParams
);
}
@@ -902,10 +837,9 @@ export async function executeStep(
}
: { stepName, ...ownershipStamp },
},
// Guard the claim; see StepExecutorParams.slotSnapshot. A
// stale (412) rejection is intentionally NOT translated by
// startErrorToResult below, so it propagates to the caller for a
// fresh replay.
// A 412 rejection from a fencing World is intentionally NOT
// translated by startErrorToResult below, so it propagates to the
// caller for a fresh replay.
startEventParams
);
stepClaimCompletedAtMs = Date.now();
packages/core/src/runtime/suspension-handler.test.ts
-3
View File
@@ -1584,7 +1584,6 @@ describe('handleSuspension batched fan-out', () => {
expect(result.inlineClaims.size).toBe(0);
expect(result.deferredBatchWork).toBeUndefined();
expect(result.batchCommittedSlotCeiling).toBeUndefined();
// The deferred inline step still carries its input for the lazy start.
expect(result.lazyInlineSteps).toHaveLength(1);
expect(result.lazyInlineSteps[0].correlationId).toBe('s1');
@@ -1788,8 +1787,6 @@ describe('handleSuspension batched fan-out', () => {
's1',
's2',
]);
// 6 events at slots 10..15.
expect(result.batchCommittedSlotCeiling).toBe(15);
expect(eventsCreate).not.toHaveBeenCalled();
});
packages/core/src/runtime/suspension-handler.ts
+2 -35
View File
@@ -232,28 +232,6 @@ export interface SuspensionHandlerResult {
* path, the exact machinery the lazy claim's crash window already uses.
*/
inlineClaims: Map<string, PreclaimedInlineStart>;
/**
* The highest slot the batched fan-out committed, when it ran. The batch's
* own events are not in the caller's loaded log (the next reload picks
* them up), so the caller folds this ceiling into the slot snapshot it
* hands the inline executions; otherwise every inline terminal write
* would name a pre-batch position and be answered with a skipped-slot
* report echoing the events this suspension just wrote. Under
* {@link SuspensionHandlerParams.allowDeferredBatchWork} this covers the
* chunks that had committed by the handler's return (always the pair
* chunk); a trailing chunk that commits later is echoed back on the
* terminal writes like any foreign event: reports the executor reads for
* position and discards.
*
* So the echo is only fully suppressed for a SINGLE-chunk fold. On a
* multi-chunk fan-out the bodies start off the pair chunk while trailing
* chunks are still in flight, and an inline terminal write issued in that
* window still names a position below them and still draws a report for
* their events. Bounded (trailing chunks only, large fan-outs only) and
* self-correcting on the next reload, and recorded so a report seen there
* reads as expected rather than as a bug.
*/
batchCommittedSlotCeiling?: number;
/**
* The batched fan-out's deferred work, present only when the caller opted
* in via {@link SuspensionHandlerParams.allowDeferredBatchWork} and
@@ -1133,7 +1111,6 @@ export async function handleSuspension({
uncreatedWaitCount >=
1);
const inlineClaims: SuspensionHandlerResult['inlineClaims'] = new Map();
let batchCommittedSlotCeiling: number | undefined;
// The trace carrier for resilient step dispatches, resolved at most once per
// suspension (the per-step ops run concurrently and share it).
@@ -1686,22 +1663,13 @@ export async function handleSuspension({
{ status: item.status }
);
}
// Highest slot this chunk committed: the ceiling the caller folds
// into the inline executions' slot snapshot (see
// SuspensionHandlerResult.batchCommittedSlotCeiling) and one input
// of the interleaving diagnostic.
// Highest slot this chunk committed: one input of the interleaving
// diagnostic.
const chunkMaxSlot = maxEventSlot(
results.flatMap((item) =>
item.error === undefined && item.event ? [item.event] : []
)
);
if (
chunkMaxSlot !== undefined &&
(batchCommittedSlotCeiling === undefined ||
chunkMaxSlot > batchCommittedSlotCeiling)
) {
batchCommittedSlotCeiling = chunkMaxSlot;
}
committedCount += results.filter(
(item) => item.error === undefined
).length;
@@ -1963,7 +1931,6 @@ export async function handleSuspension({
queuedStepCorrelationIds,
lazyInlineSteps,
inlineClaims,
batchCommittedSlotCeiling,
deferredBatchWork,
// On hook conflict the caller advances the workflow over the conflict
// before scheduling anything and never reads the wait timeout, so don't
packages/world/src/events.ts
+4 -2
View File
@@ -799,8 +799,10 @@ export interface CreateEventParams {
/**
* How many events the writer held in its loaded log when it decided to write
* this one: equivalently, the slot it expects to land on minus one. Sent by
* every replay-context create; omitted by callers with no loaded log to be
* stale against.
* the replay loop and the suspension handler, which merge the report below
* back into their loaded log. Omitted by callers with no loaded log to be
* stale against, the step executor included: for those the report would be
* a read the World does for no one.
*
* A World's slots are dense and 1-based (see `Storage.events`), so a count
* and a position are the same number. An id that is not a position does not