Skip to content

Batch: pre-claim inline steps as born-running pairs in the suspension fold #3568

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
pranaygp wants to merge 2 commits into from
Open

Batch: pre-claim inline steps as born-running pairs in the suspension fold #3568

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
0ddc5ff
Pre-claim inline steps inside the suspension batch (born-running pairs)
pranaygp Aug 15, 2026
e49dd9b
Parallel chunk commits, per-chunk continuation, batch span attributes
pranaygp Aug 15, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 7 additions & 1 deletion docs/content/docs/v5/changelog/batched-event-writes.mdx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,8 @@ interface BatchEventRequest {
event: CreateEventRequest;
/** Client event time; under slot identity, the source of the durable createdAt. */
occurredAt?: Date;
/** Per-event compute attribution, same as the single create's CreateEventParams. */
computeInstanceId?: string;
}

type BatchEventItemResult =
Expand All @@ -60,7 +62,11 @@ The contract:

## The runtime integration (suspension fan-out fold)

**On by default.** The suspension handler folds a **clean fan-out** — the suspension's eager `step_created` and `wait_created` writes — into `createBatch` calls of at most 32 events (mirroring the server's transaction budgets). The fold only engages when the World implements `createBatch`, the run is on slot identity, and the suspension carries no attribute writes, no hook writes, and no resilient step dispatch; everything else keeps the single-event path byte-for-byte. Lazy-inline steps keep deferring their `step_created` to the lazy start exactly as before.
**On by default.** The suspension handler folds a **clean fan-out** — the suspension's eager `step_created` and `wait_created` writes — into `createBatch` calls of at most 32 events (mirroring the server's transaction budgets). Chunks of a larger fan-out commit **concurrently**: slot assignment is the World's, so parallel chunks race for slot ranges exactly like the pre-fold path's parallel single writes did, and per-entity conditions — not commit order — carry correctness. The fold only engages when the World implements `createBatch`, the run is on slot identity, and the suspension carries no attribute writes, no hook writes, and no resilient step dispatch; everything else keeps the single-event path byte-for-byte.

**Per-chunk continuation.** Each chunk's follow-on work starts the moment **that chunk** commits, not when the whole fold does: a chunk's step-execution queue messages publish right off its own commit (publish-after-create holds per step), and only the chunk carrying the inline pairs gates the replay's continuation — trailing chunks' commits and publishes are joined before the invocation can acknowledge its message, so the durability contract ("every create durable before ack") is unchanged.

**Pre-claimed inline pairs.** When the fold engages and has company for them (at least two inline steps, or one plus other batchable events), the steps the runtime is about to execute inline join the batch as adjacent `[step_created, step_started]` pairs — the created row carrying the input, the started row a bare ownership-stamped claim the World folds into a born-running create. The inline bodies start straight off the pair chunk's commit (in parallel with the queue publishes and any trailing chunks) with no per-step claim POST at all, and a pair that loses its atomic create-claim to a concurrent delivery skips its body exactly as a lost lazy claim does. A lone inline step with nothing else to batch keeps the optimistic lazy-start path, whose claim overlaps the body.

Per-event `409`s are tolerated the same way the single path tolerates `EntityConflictError` (a concurrent delivery already created the entity); any other per-event failure fails the suspension write the way a single-path rejection would.

Expand Down
130 changes: 115 additions & 15 deletions packages/core/src/runtime.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -100,6 +100,7 @@ import { runIdCreatedAt } from './runtime/run-id-time.js';
import {
DEFAULT_STEP_MAX_RETRIES,
executeStep,
type PreclaimedInlineStart,
} from './runtime/step-executor.js';
import { computeStepLatencyTracking } from './runtime/step-latency.js';
import {
Expand Down Expand Up @@ -3126,6 +3127,21 @@ export function workflowEntrypoint(
),
getTraceCarrier: nextTraceCarrier,
},
// Inline pre-claims: lets the batched fan-out
// fold the lazy-inline steps' step_created +
// step_started pairs — stamped with this
// message's ownership — into the one commit, so
// the bodies below start straight off it. See
// SuspensionHandlerResult.inlineClaims.
ownerMessageId: metadata.messageId,
// Let the fold return once the pair chunk has
// committed: trailing chunks and the per-chunk
// step-message publishes ride
// `deferredBatchWork`, which this invocation
// joins before it can ack (below, next to the
// dispatch join) — so the durability contract
// is unchanged while the bodies start earlier.
allowDeferredBatchWork: true,

@karthikscale3 karthikscale3 Aug 15, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

AI [question]: This opt-in changes an ordering property that held before it, and I'd like to confirm nothing downstream depends on the old one.

Bodies start off the pair chunk's commit while trailing chunks ride deferredBatchWork, so a fast inline body can write step_completed before a trailing chunk commits its step_createds. Previously — including #3025 — await Promise.all(dispatches) gated the bodies, so every create in the fold was durable before any body ran. The new contract is only "every create durable before ack", which is strictly weaker: the log can now hold a step's terminal event at a lower slot than a sibling's created event.

The replay path looks safe: matching is by correlation id, slots stay dense, and creates are idempotent, so a redelivery after a trailing-chunk failure re-creates the missing steps correctly.

What I can't rule out is consumers outside the replay path — the ClickHouse analytics ingest and the run-details UI reconstruct run shape from the event stream, and either could reasonably assume created-precedes-terminal globally rather than per-step. Is that assumption made anywhere? Asking for confirmation rather than a change.

});
} catch (suspensionError) {
// A suspension create was rejected as stale: re-derive
Expand Down Expand Up @@ -3586,7 +3602,21 @@ export function workflowEntrypoint(
)
);
}
await Promise.all(dispatches);
// The dispatch publishes and the inline bodies below
// run CONCURRENTLY: the suspension commit already made
// every dispatched step durable (and, when the fold
// engaged, settled the inline pairs' claims), which is
// the only ordering both sides need — so neither waits
// for the other. The joins below (before step results
// are read, and on the no-inline early returns) keep
// the failure contract: a dispatch rejection still
// fails this delivery, after in-flight bodies settle.
const dispatchesSettled = Promise.all(dispatches);
// A rejection must not surface as an unhandled
// rejection while the bodies run (or if setup between
// here and the join throws first); awaiting the
// original promise below still observes it.
dispatchesSettled.catch(() => {});

// The set of steps THIS invocation executes: the
// deferred lazy-inline batch plus any owned-recovery
Expand All @@ -3602,12 +3632,29 @@ export function workflowEntrypoint(
correlationId: string;
stepName: string;
lazyStepInput?: (typeof lazyInlineSteps)[number]['dehydratedInput'];
preclaimedStart?: PreclaimedInlineStart;
}> = [
...lazyInlineSteps.map((s) => ({
correlationId: s.correlationId,
stepName: s.stepName,
lazyStepInput: s.dehydratedInput,
})),
...lazyInlineSteps.map((s) => {
// Pre-claimed by the suspension batch: the pair
// already settled this step's create + claim, so
// the executor runs (or skips) the body off that
// verdict instead of sending a lazy start with
// the input.
const claim = suspensionResult.inlineClaims.get(
s.correlationId
);
return claim
? {
correlationId: s.correlationId,
stepName: s.stepName,
preclaimedStart: claim,
}
: {
correlationId: s.correlationId,
stepName: s.stepName,
lazyStepInput: s.dehydratedInput,
};
}),
...ownedRecoverySteps.map((s) => ({
correlationId: s.correlationId,
stepName: s.stepName,
Expand Down Expand Up @@ -3652,6 +3699,14 @@ export function workflowEntrypoint(
// queued (or no work needs scheduling). Exit and let
// the queue drive subsequent replays.
if (inlineExecutions.length === 0) {
// Nothing runs concurrently with the dispatches on
// this path — join them (and the fold's deferred
// chunk commits/publishes) here so a failure fails
// the delivery exactly as it always has.
await Promise.all([
dispatchesSettled,
suspensionResult.deferredBatchWork,
]);
// A `hook.getConflict()` awaiter needs an immediate
// re-invocation: the replay consumes the
// just-committed hook_created and resolves the
Expand Down Expand Up @@ -3848,9 +3903,24 @@ export function workflowEntrypoint(
// 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 inlineClaimSnapshot = slotSnapshotParams(
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 — 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 =

@karthikscale3 karthikscale3 Aug 15, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

AI [note]: This fix is partial under the round-2 architecture, and the description reads as unconditional.

batchCommittedSlotCeiling only folds in slots from chunks that have committed, but the bodies start off the pair chunk while trailing chunks are still in flight. So on a multi-chunk fold, an inline terminal write issued before the trailing chunks land still names a position below them and still draws a skipped-slot report — the thing this change removes, partially reintroduced by the per-chunk deferral.

Bounded (trailing chunks only, big fan-outs only) and self-correcting, so not worth restructuring. Worth narrowing the claim to single-chunk folds so the next reader doesn't chase a report that is expected.

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
Expand Down Expand Up @@ -3909,7 +3979,8 @@ export function workflowEntrypoint(
// every inline step (which would be O(n²)
// across a long sequential workflow).
authoritativeAttempt:
s.lazyStepInput !== undefined
s.lazyStepInput !== undefined ||
s.preclaimedStart !== undefined
? 1
: countStepStartedEvents(
eventLog.events,
Expand All @@ -3923,8 +3994,15 @@ export function workflowEntrypoint(
// input on step_started so the world creates
// the step on the fly. Absent for
// owned-recovery steps, whose input hydrates
// from the existing step entity.
// from the existing step entity, and for
// pre-claimed steps, whose pair already
// carried it.
lazyStepInput: s.lazyStepInput,
// Pre-claimed inline start: the suspension
// batch settled this step's create + claim;
// the executor runs (or skips) the body off
// that verdict with no start write of its own.
preclaimedStart: s.preclaimedStart,
// Inline ownership: stamp (or re-stamp) this
// invocation's queue message ID on the
// step_started, so wake replays see the body
Expand All @@ -3945,7 +4023,8 @@ export function workflowEntrypoint(
runReadyBarrier,
slotSnapshot: inlineClaimSnapshot,
...(stepIndex === 0 &&
s.lazyStepInput !== undefined &&
(s.lazyStepInput !== undefined ||
s.preclaimedStart !== undefined) &&
latencyTracking
? { latencyTracking }
: {}),
Expand All @@ -3964,14 +4043,15 @@ export function workflowEntrypoint(
// these bodies until they settle — see
// assertNoInFlightOwnedSteps.
inFlightOwnedSteps.add(s.correlationId);
// Lazy steps are brand-new (their create-claim
// is the exactly-once gate), but an
// owned-recovery step already exists and its
// delayed backstop message may fire mid-body
// Lazy and pre-claimed steps are brand-new
// (their create-claim is the exactly-once gate),
// but an owned-recovery step already exists and
// its delayed backstop message may fire mid-body
// in this same process — route those through
// the in-process single-flight.
const executed =
s.lazyStepInput === undefined
s.lazyStepInput === undefined &&
s.preclaimedStart === undefined
? runStepSingleFlight(
runId,
s.correlationId,
Expand All @@ -3984,6 +4064,26 @@ export function workflowEntrypoint(
}
);
try {
// Join the dispatch publishes launched above and
// the fold's deferred batch work (trailing chunk
// commits + per-chunk step-message publishes) —
// the bodies are already running in parallel with
// both, and this invocation must not ack before
// every create and publish is durable. A failure
// keeps its old contract (fail this delivery so
// the message redelivers), but the in-flight
// bodies must settle first: an owned body left
// running past this handler would race its own
// redelivery.
try {
await Promise.all([
dispatchesSettled,
suspensionResult.deferredBatchWork,
]);
} catch (dispatchErr) {
await Promise.allSettled(stepExecutionPromises);
throw dispatchErr;
}
stepResults = await Promise.all(
stepExecutionPromises
);
Expand Down
Loading
Loading