121 · T106 — the exec-node pod-side verb and attempt shard writer - #131
Merged
Merged
Conversation
…d writer
The ticket's Test plan translated into one suite, authored BEFORE any
implementation. Every test launches the pipeline binary as a SUBPROCESS against
a local-filesystem blob store — a subprocess invocation is indistinguishable
from a pod invocation from the verb's point of view, which is the whole reason
this half of ADR 115 lands before the executor that submits it.
* One attempt, faithfully — the golden path stores its output and records
`succeeded`; the shard's records are the LOCAL in-process records (same
kinds, same order, same classification, compared against a real engine run
rather than a hand-written list); each `TaskError` class records its
matching outcome on a distinguishing exit code; a panic is contained and
attributed without aborting; a node whose policy grants retries still
performs exactly one attempt and emits no backoff.
* Shard integrity — the shard names the build that wrote it (run/node/
attempt, both fingerprints, tool version, image digest); an interrupted
write leaves no shard and no temp debris, and a truncated one is refused as
incomplete; a foreign structural fingerprint is refused naming BOTH values,
at the reader and up front at the pod.
* Inputs — missing (absent) and corrupt inputs fail on a code distinct from a
task failure and name the reference; a multi-input node rehydrates in
declared order (asserted by swapping the references and watching the answer
change); wrong arity is refused before any work; the references the pod was
actually given are recorded positionally with their content hashes; a
consume-nothing node rehydrates nothing and refuses a spurious reference.
* Re-entrancy — the task obtains its resource from the registry the binary's
own flow-building path rebuilt, constructed exactly once per invocation.
* Cancellation — SIGTERM mid-attempt (synchronised on an on-disk marker, never
a sleep) yields a truthful `cancelled` shard inside the shutdown budget.
* No new exit numbers, and no cluster — asserted structurally over this file's
own source so the claim cannot rot.
They fail: `dagr_cli::shard`, `dagr_cli::exec_node_demo`, and the demo binary
do not exist.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…-triggered signal wiring
Three pieces the verb is built out of, each useful on its own.
The **attempt shard** (`crate::shard`) is a fragment of the existing event stream,
not a second format: its body is the same records `AttemptEventSink` produces,
stamped through the SAME translation the driver uses (`records_for` drives
`driver::write_attempt_event`), so "the pod's records are the local records" is true
by construction rather than by review. Around that body sit a header — run, node,
attempt, both fingerprints, tool version, image digest, and the ordered input
references the attempt was given — and a trailer carrying the terminal state, the
durable output reference with its `DurableReferenceMeta` fields, diagnostics, and the
record count. Both carry `dagr.attempt-shard@1`, never the event stream's version, so
they cannot be mistaken for event-stream records.
* It lives in the blob store at a deterministic address derived from the identity
triple, because there is no callback: the orchestrator must be able to COMPUTE
where the shard is. The node name is addressed by its digest, for the same reason
ADR 115 §4 puts identity in annotations rather than labels — a node name is
author-chosen and need not be a legal path segment.
* A partial shard is undetectable-as-complete twice over: the write is temp +
fsync + rename (so an interrupted write leaves no shard and no debris), and the
trailer is last and declares the record count (so bytes truncated by any other
route are refused as `Incomplete`).
* `verify_build` refuses a shard from a different program and names BOTH values.
`RunnableFlow::prepare_attempt` is the **single-attempt seam**: it assembles the flow
exactly as `run` does and builds only the requested node's runner, under a new
`AttemptDiscipline`. `ExactlyOnce` collapses the retry budget to one attempt and arms
no timeout, because two retry loops duplicate an attempt and both decisions are the
orchestrator's (ADR 115 §2). Remote eligibility is a captured capability, not a
runtime check: the `Payload`-bounded registrars monomorphize a decode-into-slot and
an encode-from-slot `fn` pair where the concrete type is still known, and a node
registered through an ordinary registrar carries none and is refused by name.
Filling an upstream's slot is what rehydration does, so the node's own runner reads
its inputs through the ordinary deferred read — the attempt path has no special case
for having run elsewhere. `register_payload_with` / `register_source_payload_with`
close the gap the bound opened: a payload-bounded node with a stated policy is
exactly what a placed node is.
`install_signal_handlers_with` generalizes the OS-signal wiring to any trigger
closure, reusing the isolated listener runtime and the same re-entry hardening. A
single pod-side attempt has no run loop to request cancellation OF — only a
`CancellationSource` the attempt observes.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
The pod side of ADR 115, which does not know it is in a pod. It takes an identity
triple and a list of durable references on argv, rehydrates them, runs exactly one
attempt through the same caught-attempt path a local run uses, stores the produced
value, and writes a shard — submitting nothing, watching nothing, speaking to no
cluster. That is why its whole suite runs as a plain subprocess.
* `exec-node` joins the closed verb table UNCONDITIONALLY: an operator (or an
orchestrator) must get the same answer from every dagr binary about what a verb
IS. Its body needs the blob store the default-off `blob` feature provides, so a
build without it answers with a recognized stub naming the missing feature —
the shape `resume` and the metastore's reserved open modes already use. It
routes through the same flow-selection rules every other flow-selecting verb
uses, so a `#[dag]` binary gets it for free.
* `single-node` is untouched. The two are siblings, not one verb: `single-node` is
operator-facing and replays node N from run R out of the run store;
`exec-node` is machine-facing, needs no prior run, and reports through a shard.
A merged verb would have to guess its mode from which flags were present.
* Inputs arrive as a repeated `--input`, preserving positional order. Trailing
positionals would collide with the flow-name positional C26 already reserves,
and a single-flow binary may omit that name — so the first reference would be
eaten as a flow name.
* Missing, corrupt, wrong-arity, and undecodable inputs are refused BEFORE the
task runs and on a code distinct from a task failure: an orchestrator must be
able to tell "the storage lost the input" from "the task said no", because only
one of them is the pipeline's fault. `get` verifies bytes against their key, so
an out-of-band overwrite is corrupt rather than a wrong value decoded.
* Cancellation: SIGTERM fires the `CancellationSource` the attempt observes. The
verb computes the EFFECTIVE terminal state — cancelled when the signal fired and
the attempt did not succeed, the runner's answer otherwise — and stamps it onto
the single `node-terminal` record, leaving every other record byte-for-byte what
a local attempt emits. A success that arrives after the signal stands, because
that is the truthful record. A task that does not observe cancellation cannot be
killed, so a watchdog writes a truthful `abandoned` shard after grace and exits,
exactly as C16 describes.
* Exit codes reuse the existing table with no new numbers, documented as a table in
the module docs.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…ubprocess The honest way to test a verb whose whole premise is re-entrancy is to run the real binary: a subprocess invocation is indistinguishable from a pod invocation from the verb's point of view. `dagr-exec-node-demo` is one line of dispatch over a single-flow registry — exactly what a real pipeline binary is — and the flow behind it has a node for every shape the verb must get right: a source, a doubler, an order-sensitive two-input combiner (so swapping the references CHANGES the answer, which is how positional order is proved rather than asserted), a task that fails in each `TaskError` class, a panicking task, a cooperatively-cancelling task, a resource-reading source, and a retrying node that really does attempt several times locally — otherwise "the pod attempts it once" would prove nothing. The resource registry is built INSIDE the factory, which is the whole point of the per-pod lifetime claim: because the pod re-enters this code, the resource is constructed once per invocation, and the test measures exactly that (one line after one invocation, two after two). Test fixes that fell out of running it: the shard's output/identity accessors are `recorded_*` (the builders own the short names, mirroring `DurableReferenceMeta`); `libc::kill` carries its safety justification at the call site; and the no-cluster assertion assembles its needles at run time, because a literal would be the very occurrence it forbids. Every one of the 21 tests passes. The suite reaches no cluster, and says so structurally over its own source so the claim cannot rot. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Twelve decisions recorded in the ticket file per ticket-conventions §5 — the ticket's own two plus the ten the implementation had to make that the ticket did not name. `docs/tasks.md` carries no T106 entry (it enumerates M0–M4 only), so there were no `Q:` items beyond this file's own. The two the ticket asked: `exec-node` and `single-node` stay SIBLINGS (operator- facing vs machine-facing; a merged verb would guess its mode from which flags were present, and `single-node`'s behaviour is byte-for-byte unchanged), and the shard lives in the BLOB STORE at a deterministic attempt-keyed path — with the consequence the ticket does not spell out written down, that with no callback the orchestrator must be able to COMPUTE the address, which content addressing cannot supply. The ten the implementation answered: the shard's shape and its two independent partial-shard defences; a repeated `--input` rather than trailing positionals (they would collide with the flow-name positional C26 reserves); `--blob-store` for the output and shard destination; `contract::TOOL_VERSION` rather than the package version, because the shard's tool version answers the comparability question resume already asks; `--image-digest` as an argument, since no image digest exists in the engine and Kubernetes does not expose one through the downward API; remote eligibility as a captured `fn` pair from the `Payload`-bounded registrars; the effective-terminal-state override that makes a cancelled attempt truthful in exactly one record; the verb in the table unconditionally with its body behind `blob`; `with_resources` reaching the pod path only, with the driver's own resource injection deliberately left alone; and the exit-code mapping onto the existing table. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
The verb reads its inputs from a blob store and writes its shard to one, and the demo binary it launches carries `required-features = ["test-kit", "blob"]` — so a default `cargo test --workspace` has neither the module nor the binary and the target failed to compile. `#![cfg(feature = "blob")]` is the same gating `blob_bridge.rs` already uses for the same reason. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…pipeline identity
Two corrections found by reading the gate's own assertions rather than by a test
failing at random:
* `SPAWN_INVENTORY` (the C16 "nothing orphaned" inventory) gains the pod-side
shutdown watchdog and the second OS-signal listener, each with the reason it
cannot outlive the run. The watchdog's reason is the interesting one: a task
that does not observe cancellation cannot be killed, so the thread enforces the
grace budget and exits the PROCESS — which cannot outlive the run because for
`exec-node` the run IS the process, and the thread is disarmed the instant the
attempt completes.
* The attempt context's pipeline identity is the selected flow's name, not the
node's. Using the node name would have made the attempt's log span read
differently in a pod than it does locally, for no reason but an oversight.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
C26 enumerates the verb set every pipeline binary shares, and a verb that ships without appearing there leaves the spec quietly contradicting the surface — the exact drift ADR 115's truth pass was written to stop. The new paragraph states what the verb is (the machine-facing sibling of *run a single node*, changing nothing about that verb), why the wire format can be tiny (the remote side re-enters the same binary, so skew is impossible by construction), what it deliberately does not do (retry, backoff, timeout — the orchestrator's), and what it reports through (the attempt shard, its address, its atomic-and-last write discipline, and its build-identity check). It also records the consequence rather than hiding it: a resource registry rebuilt this way exists once per invocation, not once per run. `ShardOutput` additionally documents why it carries no `produced_at_offset_ns`: an offset is measured from a run's start and the pod's clock starts at its own attempt, so carrying one across would be a number that looks comparable and is not. `scripts/check-remote-execution-scope-adr.sh` still passes — the carve-out is described, not widened. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…ences, and apply rustfmt The pod cannot call `Blob::<T>::rehydrate` itself — its codec is erased at the node boundary, which is precisely what lets one verb serve every node — so "encode and store the output through T104" has to mean that the bytes and the reference grammar are identical on both sides. That is now asserted rather than assumed, in both directions: an input published with `Blob::put` is what the pod reads, and the reference the pod's shard names rehydrates back through `Blob::<Counted>::rehydrate` with a content hash matching the one the pod wrote down. T108 depends on this entirely. `cargo fmt --all` across the ticket's files. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
The `blob` cli feature is default-off, so the plain `cargo test --workspace` leg compiles the suite to nothing — it would have shipped un-run. It joins the existing `--features blob` step rather than adding a third cargo invocation: one invocation links both targets in parallel and pays cargo's startup once, which is the grouping the metastore steps already use, and a failure still names its own test and target. Both tiers, deliberately. The suite launches the pipeline binary as a subprocess in place of a pod and sends it a real SIGTERM, so process spawning, signal delivery, the shard's temp-file + fsync + rename, and the shutdown budget are all exercised per platform — exactly the class of thing that differs between ubuntu and macOS, and what the ticket's "tests pass on both" line is asking for. It reaches no cluster: a local-filesystem blob store is the whole of its infrastructure. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
The table enumerates what `run_registry` routes and what each verb's exit code is; `exec-node` now routes there and would have left the doc quietly incomplete. The added prose says WHY it routes there rather than being dispatched directly: the remote side re-enters the same binary and rebuilds the flow through the same factory, so it needs exactly the selection rules every other flow-selecting verb uses — which is also what gives a `#[dag]`-declared DAG the verb with no registry edit. The "what the registry does not route" section is unchanged and still correct. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…ame the shard fault seam
`cargo clippy --workspace --all-targets` — what the gate and CI both run — compiles
the `blob`-gated modules to nothing, so none of this code was linted by either. Run
deliberately with the feature on, it had real findings, all fixed:
* `args` next to `argv` was a similar-binding trap; the parsed value is now
`request`, which is what it is.
* `exit_code_for`'s two run-failure arms are merged, with the reasoning kept: the
three propagated/carried-forward states are decided over a whole graph and a
single attempt can never reach one, so if one appeared the engine would be wrong
about something — the conservative answer is the same number, which is why they
share the arm.
* `AttemptShard::parse` split: `parse_header` answers the identity question, and
what remains answers the structural one ("is this a complete shard?").
* `# Panics` on the demo flow factory; a redundant closure in the suite.
Separately, the shard's fault-injection seam is `DAGR_SHARD_WRITE_FAULT` rather than
a `DAGR_DEMO_*` name — it lives in production code and should not pretend otherwise
— and it now documents why it is an environment variable at all: the thing under
test is a separate process, so a test cannot reach a parameter inside it. That is
the one difference from the blob backend's in-process `stop_before_rename` seam.
`shard.rs` also records why the shard carries no `produced_at_offset_ns` and why its
`offset_ns` values are restamped on replay — the abstract attempt-event port has no
clock, so a local attempt's records are stamped at drain time too, and the shard
loses nothing by being stamped at write time.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
The table is pinned by name and order on purpose — verb parity is structural, and a verb appearing or moving is meant to be a review-visible diff. `exec-node` sits immediately after `single-node`, next to the verb it is the machine-facing sibling of, so every existing verb keeps its index. The doc comment records why it is pinned unconditionally rather than behind a `cfg`: an operator or an orchestrator must get the same answer from every dagr binary about what a verb IS, including from a build whose default-off `blob` feature leaves the verb's body unavailable — that build answers with a recognized stub naming the feature, which is a different thing from not recognizing the verb. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
The workspace runs two deliberate philosophies — recover where a poisoned lock must not escalate, panic where poisoning signals an invariant already violated — and `error_chain_and_panic_hardening` enforces that every production lock site says which one it is and why. Three new sites did not. All three are the same call: **recover**, for a reason specific to this verb. The shard is the only report a pod-side attempt will ever make, and a panic during the attempt is precisely one of the records the shard exists to carry. Refusing to read or append records because that panic poisoned the buffer would destroy the report of the failure it is reporting — strictly less truthful than the panic itself. The `LineSink` buffer additionally cannot be poisoned in practice: it is written and read by one thread inside `records_for`. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…te a byte-exact trybuild snapshot `feature-matrix` is the only leg that runs `cargo test --workspace --all-features`, and it red-lined on `crates/cli/tests/flow_builder_compile_fail.rs` — T78's harness, not anything this ticket wrote. That harness matches `.stderr` byte-exactly, and the `source_without_stable_name` sample's diagnostic reproduces rustc's "the following other types implement trait `StableName`" list. That list is the set of `StableName` impls reachable from `dagr_cli`, so it is a function of the crate's FEATURE RESOLUTION rather than of the boundary being pinned: rustc prints it in full up to nine candidates and truncates to eight plus "and N others" past that, the default resolution sits at exactly nine, and `exec_node_demo`'s four pipeline tasks (`Counted`, `Double`, `Combine`, `Boom`) take it to thirteen whenever `test-kit` and `blob` are both on. `trybuild` reads the enabled features out of the test binary's own fingerprint and passes them to the project it generates, so one snapshot cannot be blessed for two resolutions, and it offers no per-sample opt-out. Giving the demo tasks no `StableName` is not available (they are graph-emitted and fingerprinted) and neither is moving them out of the library (the suite and the demo binary both import them). What the harness pins — `FlowBuilder`'s bounds — is gated by no feature, so the `--all-features` repetition of those samples asserted nothing about the boundary and only re-asserted rustc's candidate-list formatting. The harness therefore carries `#![cfg(not(feature = "blob"))]`, the negated form of the file-level gate `blob_bridge.rs`, the `metastore_*` suites and the `schema-validation` suites already use, with the reasoning written at the gate. The `test` job still runs it on both platform tiers under the default resolution, and T78's samples, snapshots, and bounds are untouched. Recorded as resolution 13 in the ticket's open-questions section, including the standing consequence: the default resolution is one impl below rustc's truncation cliff, so the next `StableName` impl on `dagr_cli`'s default surface rewrites this snapshot and must be blessed deliberately. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Ticket: T106 — docs/implementation/121-T106-exec-node-verb-and-shard-writer.md
Summary
Adds the pod-side
exec-nodeverb: it runs exactly one attempt of one named nodethrough the existing attempt path — no retry, no backoff — rehydrating its inputs
through the T103/T104 codec and blob seams and writing an attempt shard back to
the blob store. The shard is event-stream-shaped JSON Lines (header + records +
trailer), written atomically and last, so a partial shard is detectable.
single-nodeis untouched; the two verbs are siblings. No Kubernetes code — the whole suite drives
the pipeline binary as a subprocess over a local blob store, with no cluster
anywhere.
Tests-first
Confirmed — failing tests committed first in 0b7a075.
Definition of done
exec-node --run --node --attempt <input refs>runs exactly one attempt of the named node through the existing attempt path, with no retry and no backoff.DurableReferenceMeta.cancelledshard inside the shutdown budget (delivered as a real signal vialibc::killto the spawned subprocess, not a simulation).ResourceRegistryis rebuilt in-pod; a test documents its once-per-pod lifetime.ubuntu-latestandmacos-latest— pending, confirmed by this PR's CI. A new CI step runs the suite on both tiers, folded into the existing--features blobinvocation alongside the T104 bridge test.GATE=PASS(35 checks PASS, 1 benign SKIP: a duplicate of an already-run check), but CI on the PR is the authoritative verdict.Open questions resolved
All 12 are recorded in an
## Open questions — resolvedsection appended to the ticketfile (
docs/tasks.mdcarries no T106 entry, so there were noQ:items beyond thefile's own two):
exec-nodevssingle-node→ siblings;single-nodeis byte-for-byte unchanged.--input, not trailing positionals.--blob-store <container>.contract::TOOL_VERSION(dagr@1), not the package version.--image-digest, optional, absent when not supplied.Payload-bounded registrars, as capturedfnpointers.cancelledwhen the attempt runner only reports the task's own result → an effective-state override, stamped in exactly one record.blob.RunnableFlow::with_resources, built inside the flow factory.Deviations
None.
Notes
Clippy under
--features blobsurfaces real lints that the gate's and CI's default--all-targetsleg cannot see, because the defaultcargo test --workspacecompilesthis suite to nothing. Those lints were fixed in
28f7de6, andcargo clippy -p dagr-cli --features blob --all-targets -- -D warningsis clean. Worthknowing that visibility gap exists for T107–T110.
docs/arch.mdrecords the verb at C26;docs/flow-registry.mdrecords it in therouting table.