Docs / Guides

contract run · 30-run.md · 31-pipeline.md · 32-derive.md · guide · card

Runs, pipelines and derive

The problem a run solves

A pipeline copies records from a source, such as a vendor API, into store tables. When a process dies halfway through a pull, the restart has to write each row once, keep the source position in step with those rows, and send no second paid request. A run is one execution of a pipeline over one scope.

Declaring and compiling a pipeline

The example pipeline, record-sync, declares a source, one table named records and incremental = "updated_time" (pipeline-specrun.declare.pipeline-specA pipeline specification takes PipelineSpec, its source and destination each SourceBlock, and each tables entry a bare name or a table block, the two forms mixing in one array.). That field is the stream clock: the source reports a cursor that stores the field name beside its value (incremental-fieldrun.declare.incremental-fieldincremental = "<field>" names the stream clock, an RFC 6901 pointer into each fetched row when it opens with /; with it the source reports a monotonic cursor committing watermark-shape, and without it the source refetches in full.). An edit that restates a default leaves the content hash unchanged (content-hashrun.declare.content-hashcontent_hash is the sha256 of the specification's RFC 8785 canonical JSON with every optional field at its default elided, so an explicit default hashes as its absence.).

The compiler turns the declaration into a plan, a flat node list holding no script (plan-noderun.compile.plan-nodeA plan is a flat node list with predecessor edges, each node a NodeKind.). Branches and fan-out come from declared nodes, not from data, so a replay walks the same graph (control-flowrun.compile.control-flowA data-dependent conditional or loop in a workflow body raises PipelineUndeclaredControlFlow; branching travels as branch and fan-out as parallel.).

Each deployment supplies a durable orchestrator implementing one execution interface (rolerun.orchestrate.roleA target's orchestrator implements every operation in substrate-port. The engine runner over storage-ports is the reference implementation.).

A run that dies and resumes

The engine writes the run row before the first pull, so a reader tells a zero-row success from no run (row-at-openrun.record.row-at-openThe engine writes a run row at open, before the first pull. Readers distinguish zero-row success, failed with an error kind, and no run row.). An execution owner binds the pending work to its scope, connector build and plan hash (execution-ownerrun.own.execution-ownerOne durable execution owner exists per scope; a table or chunk owner holds the connector identity with its component world, the pipeline content_hash and the input hash.), and the owner renews a lease while it runs (owner-leaserun.record.owner-leaseA running or waiting row carries an owner of process id, boot id and a lease expiry, renewed every 10 s with a 30 s time-to-live.).

The pull admits rows at or after the stored position, so each poll re-reads the boundary instant. records declares a key, which makes those repeats idempotent; a keyless table accumulates them (inclusive-boundaryrun.advance.inclusive-boundaryA polled load admits a row whose clock value is at or after the stored position, re-landing the boundary instant's rows on every poll, idempotently on a keyed table and accumulating in a keyless table's union view.).

The journal records each pulled page exactly once, although the request behind it runs at least once (step-outputrun.journal.step-outputThe journal records a step's value exactly once. Its effect runs at least once: a crash between effect and write repeats the effect. Every later call under the key returns the recorded value.). Each outbound request carries an idempotency key derived from the journal entry, so a repeated request names itself (idempotency-keyrun.journal.idempotency-keyEvery outbound request a step makes carries an idempotency key derived from its entry key, identical on every re-entry of that effect.). A value over 1 MiB moves to a content-addressed blob (inline-cutoffrun.journal.inline-cutoffThe journal stores values of 1 MiB or smaller inline as row bytes. For larger values, the journal stores a content-addressed blob named by its sha256 and a reference in the row.).

Now the process dies after its first page. Startup marks the row partial_failure because its owner lease has expired (orphan-reaprun.record.orphan-reapStartup marks only non-terminal rows with expired owner leases as partial_failure. Startup leaves rows held by live processes unchanged.). The next fire finds no commit marker newer than the cached position, so the pending owner stays (marker-reconcilesrun.own.marker-reconcilesAt run open, a commit marker newer than the catalog's cached position retires the pending owner that produced it before any replay.).

The plan hash matches, so the run replays. The journal hands the recorded page to the land path byte for byte (replay-landsrun.journal.replay-landsA journal hit hands the recorded batch to the land path, so a resumed run lands bytes identical to an uninterrupted one, batch ordinal included.), and the replay sends no request to the vendor (no-double-spendrun.retry.no-double-spendA resumed attempt replays every recorded step, so it reserves no meter permit and issues no call an earlier attempt already recorded.).

The remaining pages pull live. The commit publishes the rows and the new position together (commit-with-rowsrun.advance.commit-with-rowsA table's run commit marker, run-manifest, publishes its rows and new position together. The catalog cursor caches the newest committed marker.), and success releases the owner (pin-releaserun.own.pin-releasesuccess, a failure that wrote no batch, and replay-stable failures under staged-bytes, late-type, or source-stage-schema release the owner; every other status holds it.).

Two changes stop the replay instead. A declaration whose hash changed while the owner was pending fails the open, naming both identity sets (pinned-plan-changedrun.own.pinned-plan-changedWhile an owner is pending, a changed pin — connector identity, component world, pipeline content_hash, or a host's plan reference or identities — raises ExecutionPinMismatch before replay, terminal and non-retryable, naming the scope and both pin sets.). A renamed updated_time fails the open before any request leaves the host (field-renamerun.advance.field-renameOpening a position whose stored field differs from the declared incremental field raises CursorFieldMismatch before any request leaves the host.).

On D1, a callback guard holds only its own request’s owners and leases (scoperun.callback-guard.scopeA guard protects each owner scope its callback admits, every ancestor scope and every lease key, entity fences included; another job's admission, an unrelated export and transfer of any unprotected row proceed.), and releases unattended only when a replay is safe (replayrun.callback-guard.replayWithout an operator, a guard releases only when its holder is gone, its owner lease has lapsed and window still covers its key; release and takeover commit in one D1 batch, and replay sends that key.).

Inside one batch

A batch passes a fixed stage order from pull to commit (stage-orderrun.land.stage-orderA batch passes pull, the secret guard, the recorded pull, normalize, the transform chain, write-path redaction, shredding and batch write; the run then commits rows and position together through commit-with-rows.). The secret guard runs before the recorded pull, so a replay reintroduces no credential (placementrun.guard-secrets.placementThe secret guard runs at the one pull path streaming and backfill share, ahead of the recorded pull and the land path, so a replay reintroduces no credential.). The guard masks credential-shaped text and logs a count per column; it blocks no run (mask-onlyrun.guard-secrets.mask-onlyThe guard is on by default and blocks no run; each pull logs the count of masked cells per column.).

Normalize keeps the nesting a sink supports, or flattens it into child tables joined by a foreign key (moderun.normalize.modenative keeps nesting the sink supports and expands unsupported nesting with schema-diff-home. relational flattens structs into parent-child column names and converts lists into child tables joined by a foreign key.). The declared transform chain then selects, renames, casts, extracts and filters columns of the root table (chainrun.transform.chainThe chain is an ordered list of TransformOp, declared once per pipeline and bound to the root table; a filter takes exactly one column.). A commit makes one table’s rows visible in one step, and a crash before it leaves a recoverable partial run (commit-visibilityrun.land.commit-visibilityA commit makes a run's rows visible for one table in one step; a crash before it leaves a recoverable partial run.).

Failures, stops and waits

One tagged failure type crosses every port, and a caller branches on the tag, not on a message (failure-taxonomyrun.retry.failure-taxonomyOne tagged failure type, FailureTag, crosses every port. A caller branches on the tag, never on a message.). Only Transient and RateLimited retry, under the step’s schedule (retryable-classesrun.retry.retryable-classesTransient and RateLimited retry; every other tag stands as the step's answer whatever budget remains. retry_on narrows that pair.). In the example, a vendor rate limit retries the page, and an expired credential ends the step.

A stop marks the run, and every await observes it through one cancellation token (one-tokenrun.cancel.one-tokenEach run holds one cancellation token that every await selects on: the pull, a retry sleep, an awakeable wait and a subprocess wait. A meter reservation holds no wait of its own, under acquire-call.). A stop leaves landing uncut: received bytes finish landing first (land-path-uncutrun.cancel.land-path-uncutA table run's own stop does not interrupt landing. The engine finishes landing bytes already received and records the unfulfilled stop. Prepared body outputs additionally require body-parent.).

A run waiting on an outside party suspends on an awakeable, a single-use token the party posts back (awakeablerun.suspend.awakeableAn awakeable suspends a run durably: the engine mints an opaque single-use token and persists a pending row through the awakeable store; an external party resumes by posting the token back.). The caller supplies the deadline, and the core reads no wall clock (deadlinerun.suspend.deadlineA deadline is the creation instant plus a time-to-live, both caller-supplied RFC3339 Zulu strings; the core reads no wall clock, and a suspension past its deadline at an injected instant stays timed_out under any later one.).

A store-driven job that declares isolated items gives each item a completion receipt, so a later fire skips finished items (completionrun.items.completionEach successful item lands independently with a durable completion receipt that a later fire under its execution identity skips; once every item holds a terminal receipt, the aggregate releases its pinned input owner, so later fires read fresh input.).

Loading history

A backfill splits history into chunks that workers lease. The chunk commits only while that worker’s attempt holds the row, so a stale worker’s files stay uncommitted (fenced-commitrun.backfill.fenced-commitA chunk commits in one catalog update conditioned on attempt = :mine AND status != 'done'; an update matching no row leaves that worker's files uncommitted.).

A seed bulk-loads consumer-held history through the same stage order as live pulls (one-land-pathrun.seed.one-land-pathSeeded rows travel stage-order unchanged, landing availability and _ingested_at as available-instant does; attribution survives only in the load id and a run recorded under the seeding phase.). A seeded row at or past the declared ceiling fails its chunk, which writes nothing (ceiling-breachedrun.seed.ceiling-breachedA seeded row whose ordering stamp reaches below raises PipelineSeedCeilingBreached naming the value; its chunk lands nothing, and the stamp is neither clamped nor dropped.).

A sweep runs one job per origin across a span, every origin under one frozen transaction-time bound (frozen-as-ofrun.sweep.frozen-as-ofEvery origin of one sweep reads at one as_of: the job's declared as_of, else the instant the sweep opens, recorded once as open-as-of records it for resumes.). Under carried state, each origin also reads the output the preceding origin committed (carried-staterun.sweep.carried-stateUnder state = "carried", origins run in ascending order one at a time, each reading the job's output tables at the commit of the preceding origin and every other table at the sweep's as_of.).

Models

A model builds a table from SQL over store tables (statement-sourcerun.model.statement-sourceA model declares exactly one of inline sql and sql_file; the latter reads UTF-8 text relative to the manifest containing the model.). build reads each input at its committed frontier and fires no pipeline (build-frontierrun.model.build-frontierbuild reads each input at its committed frontier and fires no pipeline; a build ordered after a landing is a build job naming that pipeline in after, under build-dependency.). A model whose SQL reaches its own id fails validation before any read (model-cyclerun.model.model-cycleIf a model's sql reaches its own id through one or more other models, pipeline validate, import, apply and build raise ModelCycle. The engine names every model on the cycle and reads no store.).

A build materializes into staging, so a refused build leaves the last published state serving readers (stagingrun.publish.stagingA build materializes into staging and publishes through manifest-commit; a refused build leaves the last published state serving.).

Derive

Derive runs a task over parent rows. Each tick selects parent rows holding no settled output under the current derivation key (anti-joinrun.select.anti-joinEach tick recomputes the outstanding set: every parent row holding neither a passage nor a settled marker under its current derivation-key in the pipeline's own output table, or, for a host task, its marker table.). A binding names a registered engine; a command in the manifest fails (command-in-manifestrun.bind.command-in-manifestA Binding key preprocess, env or allow_hosts, or a step's command, inside [pipeline.source.config] raises DeriveCommandInManifest.).

An exec engine runs a pinned binary as an argument array, with no shell and a cleared environment (no-shellrun.exec.no-shellA step runs as its argument array with no shell, in a process group of its own, under a cleared environment holding only the binding's env table.). A vendor step’s loopback proxy attaches its credential (exec-proxyconnector.attach.exec-proxyCONTEXTFUL_EGRESS_URL names a host loopback proxy, http://127.0.0.1:<port>, open for one pull. The proxy sends each request's method, path, query, headers and body to https://<endpoint_host> through the mediated client, following no redirect.). A fetch engine follows third-party links, and a host written as an address literal fails (address-literalrun.fetch.address-literalA host written as an address literal raises DeriveAddressLiteral.).

The caption parser reads SubRip and WebVTT through one grammar (grammarrun.parse-cues.grammarSubRip and WebVTT read through one grammar: a block's timing line, then its text lines joined by a space.). Each unit records one status: ok, empty, unavailable or failed (unit-statusrun.emit.unit-statusunit_status is ok, empty where the engine established nothing to derive, unavailable where it returned nothing and stated no reason, or failed on a typed error; any other value raises DeriveUnitStatusUnknown.). The command in test-enginerun.test-engineThe test-engine command builds one derive engine outside a pipeline and runs that engine against one file. runs one engine against one file, outside any pipeline.

Delivering and watching

An export delivers committed rows to a declared target. Its cursor commits after the target acknowledges a batch, so a run ending between the two resends that batch (at-least-oncerun.export.at-least-onceA run ending between a target's acknowledgement and the cursor commit resends that batch on the next run; a target deduplicates on contextful.commit_seq and contextful.row_seq.).

An Iceberg export writes one credential’s copy of a table for a query engine outside the store (audiencerun.export-iceberg.audienceEach Iceberg export publishes for exactly one admitted credential, which maps to exactly one namespace; the gateway derives a reader's namespace from its audience key, and the reader never names one.).

A live progress view projects runner events, and no execution step reads that view (best-effortrun.project.best-effortLive run state is a best-effort in-memory projection of events the runner emits after durable state changes; execution never reads it, and a restart discards it, so a subscriber recovers from the durable record.).