Fix checkpointing recursive circuits. - #6779
Conversation
mythical-fred
left a comment
There was a problem hiding this comment.
LGTM. Excellent root cause analysis in the commit messages — four distinct failures that all had to be fixed together for correct checkpoint/restore of recursive circuits. The test coverage is thorough: unit-level checkpoint+restart for reachability (single, dual, tuple and dynamic APIs), star join, and a well-designed end-to-end Python test that catches exactly the bug by keeping the program unchanged across the restart.
| /// The clock is wrapped in a validated envelope, like `CommittedZ1`, because | ||
| /// a `Timestamp` is generic and cannot carry a `CheckBytes` derive of its own. | ||
| #[derive(rkyv::Serialize, rkyv::Deserialize, rkyv::Archive)] | ||
| #[archive_attr(derive(rkyv::CheckBytes))] |
There was a problem hiding this comment.
Nit: missing — would help when inspecting checkpoint data during debugging. Not a blocker.
| // `map_local_nodes` walks nodes in node id order, and every worker | ||
| // builds the same graph, so all workers derive the same name. | ||
| let _ = circuit.map_local_nodes(&mut |node| { | ||
| if let Some(label) = node.get_label(LABEL_PERSISTENT_OPERATOR_ID) { |
There was a problem hiding this comment.
This will ignore nodes without a persistent ID, I hope that's fine. Probably for checkpointing it is, but for semantic equivalence it is not.
There was a problem hiding this comment.
That's true. It's possible that all nodes needed for checkpointing have identical pids, but some others don't. Cannot happen if the compiler labels all nodes, but we don't currently enforce this.
| (named > 0).then(|| format!("scope-{:016x}", fingerprint.finish())) | ||
| } | ||
|
|
||
| /// Absolute path of the file holding this subcircuit's clock. |
There was a problem hiding this comment.
Is this the right place for this function?
Is this the "checkpointed value of the clock"?
| // the circuit would fail. | ||
| for (int i = 0; i < operator.outputCount(); i++) { | ||
| ProgramIdentifier view = operator.outputViews.get(i); | ||
| DBSPViewDeclarationOperator decl = operator.declarationByName.get(view); |
There was a problem hiding this comment.
The code for generateOperator() actually skips ViewDeclarationOperators, but I cannot guess whether not skipping them and implementing a preorder function would produce the same result. I would have to try. Perhaps this is fine.
I wonder what their hashes look like...
`CircuitHandle::checkpoint` and `analyze_checkpoint` walk the circuit with `map_nodes_recursive_mut`, but `ChildNode` overrode only the immutable `map_nodes_recursive`, so the mutable walk fell through to the no-op default in `Node` and stopped at the subcircuit node. Operators inside a `recursive()` scope were therefore never asked to save their state, and, because `restore` never ran on them either, none of them could report a missing state file: with `need_backfill` empty the engine saw nothing to replay, so restarting an unchanged circuit from a checkpoint silently dropped the recursive scope's state and computed wrong output. Forward the mutable walk into the subcircuit. `Node::map_nodes_recursive_mut` takes `&mut self` to reach the child circuit. Part of the fix for #6765. Signed-off-by: Leonid Ryzhyk <ryzhyk@gmail.com>
`Z1Trace` and `AccumulateZ1Trace` allocate their trace lazily, on `clock_start(0)`. A nested circuit's clock starts only when the circuit first runs, so at checkpoint or restore time every trace operator inside a `recursive()` scope can still hold `None`, and the old code treated that as "nothing to do": `restore` returned `Ok(())` without reading the state file it had just been asked to load, and `checkpoint` wrote nothing. The state on disk was therefore discarded and the operator came back empty. Allocate the trace on demand in both directions, so a `None` trace restores from its checkpoint instead of silently ignoring it. Part of the fix for #6765. Signed-off-by: Leonid Ryzhyk <ryzhyk@gmail.com>
Signed-off-by: Leonid Ryzhyk <ryzhyk@gmail.com>
The compiler didn't generate persistent ids for streams that represent recursive variables inside a recursive scope. This could cause checkpointing to fail if one of these streams was passed as input to an integrator. Part of the fix for #6765. Signed-off-by: Leonid Ryzhyk <ryzhyk@gmail.com>
The state of the operators inside a subcircuit is timestamped with the subcircuit's clock, which counts the parent steps that evaluated the scope. Restoring that state without its clock left it in the future, where the operators never emit it: a recursive view came back looking empty even though its traces were on disk. Store the clock in the checkpoint. One caveat here is that we needed to generate a persistent id for the nested circuit operator. We could ask the user to provide it, but we instead derive it from the persistent ids of all streams inside the nested circuit. Fixes #6765. Signed-off-by: Leonid Ryzhyk <ryzhyk@gmail.com>
Enhance existing recursion tests to exercise checkpoint and restart functionality. Signed-off-by: Leonid Ryzhyk <ryzhyk@gmail.com>
…oint Signed-off-by: Leonid Ryzhyk <ryzhyk@gmail.com>
Signed-off-by: Leonid Ryzhyk <ryzhyk@gmail.com>
f474ba2 to
8de52bc
Compare
mythical-fred
left a comment
There was a problem hiding this comment.
One new commit since prior APPROVE @ f474ba2 (07-31 10:02). 8de52bc "Address review comments." is comment-and-log-only, zero semantic change:
circuit_builder.rs: drops the "likeCommittedZ1" comparison fromCommittedClock's doc (a follow-up comment I noted was misleading since the two envelopes aren't structurally analogous), trims the "node id would not do" digression onclock_id, rewords thecheckpoint()docstring around what the clock is used for and why restoring it to 0 breaks emission, and moves the debug! line so it can include the resolvedfilename(previous ordering serialized-and-wrote-then-logged, new ordering namesfilenameup front and logs the path). Behavior identical.replay_tests.rs: unusedDBSPHandleimport drop.python/tests/runtime/test_recursive_view_checkpoint.py: removes the docstring block that spelled out why the test catches the bug (unchanged program + non-materializededges+ edge-through-old-nodes assertion). The properties still hold in the code; only the prose explanation is gone. I'd have kept it — that's exactly the kind of note that saves the next person who touches the test — but that's author preference, not a blocker.ToRustVisitor.java: single-sentence clarification on the recursive-stream naming comment (adds "after the subcircuit is constructed" as a parenthetical). No code change.
Both commits Signed-off-by, no AI trailers. Fresh APPROVE.
Mix a state format version into the persistent id of every stream that leaves a
recursive circuit, so that the ids of the recursive views and of everything
downstream of them differ from the ids the previous release computed.
A checkpoint written by an older runtime stores no state for the operators
inside a recursive scope, so the new runtime puts the whole scope in
`need_backfill`. The pipeline diff, however, compared unchanged persistent ids
and reported no change, and the controller rejected the start with
`UnexpectedBootstrap` ("the pipeline hasn't changed, yet the circuit is
bootstrapping"), which is how `test_runtime_upgrade_round_trip` failed. With
the version mixed in, the diff reports the recursive views as modified and the
pipeline bootstraps them instead.
Only the global hash carries the version: it is the one that becomes a
persistent id, while the local hash just names the generated code. The hash of
a nested operator is built from the hashes of its members, so the change
reaches every operator downstream of the recursion, in the dataflow graph the
manager diffs as well as in the ids the runtime uses.
Signed-off-by: Leonid Ryzhyk <ryzhyk@gmail.com>
|
@mihaibudiu , can you review the last commit in the series only (added after your last review)? |
mythical-fred
left a comment
There was a problem hiding this comment.
Re-approving after the new commit 7302dbf9 ("Force new persistent ids for the outputs of a recursive circuit").
Rationale for the change is airtight and lines up with the UnexpectedBootstrap failure it addresses: a checkpoint written before this series stores no state for operators inside a recursive scope, so on restart the whole scope legitimately needs backfill — but the manager diffs persistent ids and, without a version mixed in, sees no change, then rejects the start. Mixing RECURSIVE_STATE_VERSION only into the global hash (not the local one) is the right split: local hash just names generated code and shouldn't churn, global hash becomes the persistent id and must.
Doing it at the boundary streams (nested.internalOutputs) is the right place too — the Merkle-hash-of-inputs recursion then propagates the version to every downstream operator automatically, which is exactly what you want (and exactly what the manager needs to see for closure_size, not just closure).
The test (recursiveStateVersionTest) locks in the three invariants that matter: streams leaving the recursion change, sinks downstream of them change, unrelated sinks (unrelated) do not. Parameterizing MerkleOuter on the version string just for the test is clean.
Nit, not blocking: the constant is "recursive-state-v1". When a future format change happens, whoever bumps it needs to know that any string change works — the value is just a hash-input opaque tag, not something with parsing semantics. A one-line comment on the constant to that effect ("any change to this string invalidates recursive checkpoints; format is opaque") would save someone a minute of thinking. Fine to skip.
|
|
||
| public final Map<Long, HashString> operatorHash; | ||
| public final boolean includeInputs; | ||
| /** Value mixed into the hash of the streams that leave a recursive circuit; |
There was a problem hiding this comment.
I would add an explanation why this is necessary.
Fixes #6765
Describe Manual Test Plan
Checklist
Breaking Changes?
Mark if you think the answer is yes for any of these components:
Describe Incompatible Changes