Stamps at the batch-to-records seams, leave_dynamic under multi-element stamps, and unconsolidated feedback

[frankmcsherry]

Two bugs found while running DDIR programs in steady state, the seam that let the first one through, and the driver edit mode and corgi pin that surfaced them.

Where multi-element stamps come from, and where they were leaking. Since timely #813 a message's stamp is a multiset, and DD's batch-shipping operators make multi-element ones: arrange ships each batch under the set of capabilities it retires, reduce likewise, join forwards its input batch's set. In an iterative scope with two epochs in flight, an arrange holding (1, [18]) (epoch 1, round 18) and (2, []) (epoch 2, just entered) retires both in one batch when its input frontier passes both at once. Instrumenting the operators on the run that panicked (a DDIR program at 20k nodes, four workers, epochs fed without waiting) shows the chain: an arrange minted the first multi-element stamp, then a join, then a reduce, then an arrange whose batch reached the scope's exit as (1, [19]), (2, [1]).

The leak was the seam from batches to records: as_collection forwarded a batch's records under the batch's set, so a collection carried messages whose capability spanned two epochs though no record did, and anything reading a message's time as one, cap.time(), panicked with "expected a singleton stamp". StampRouter (collection.rs) now gives one capability per element of a message's stamp and routes each record to the first element at or before its time; it is applied at every seam that makes records from batches: the vec as_collection, count, threshold, the columnar as_recorded_updates, and DDIR's corgi as_collection. With that, collection streams carry one timestamp per message and only batch streams carry sets. Not yet routed: join_with_tactic's output, whose tactics build containers the operator cannot split (the fix belongs in the tactics, a builder per element), and as_container (opaque containers, no users found). tests/dynamic.rs checks the seam with a hand-built batch under two epochs' capabilities.

leave_dynamic itself. All three (DD's row and columnar collections, DDIR's corgi backend) read cap.time() to truncate it. Each now holds a capability per stamp element, truncated exactly as the records are, and opens its session on the set, so it is safe whatever reaches it; a test makes the two-epoch message directly, panics before the fix and passes with it. An audit of DD, DDIR, and dogsdogsdogs found no other InputCapability::time() reads.

Unconsolidated feedback. A DDIR var was set to its step unconsolidated. An iteration quiesces when its feedback carries no records, not when it carries nothing net, so a step whose additions and retractions cancel only logically (a negate under a concat) never quiesced: the vec backend ran forever on a program whose fixpoint was reached in one round. Every program in the suite fed its vars through a reduce, which consolidates by nature, so nothing had seen it. Each var is now set to its consolidated step (Backend::consolidate, through the arrangement by default); the suite's timings are unchanged. The regression test is DD's identifiers algorithm written as a DDIR program (interactive/tests/identifiers.rs): hash, min, negate, and a var give 32 records unique ids over 64 slots.

The ddir driver's rounds edit file-fed rows. With EDGES_FILE, the round loop used to churn synthetic rows the dataflow never saw. Each update now retracts one of the file's rows and re-inserts it with its last field rewired, so a round is an edit to data the dataflow holds.

corgi pinned at WIP 3c4b0d6f, where compare_idx and gather read sum discriminants in place (43s -> 23s on a 200k-node saturation).

🤖 Generated with Claude Code

https://claude.ai/code/session_012k2GSwxmvD2LvckkoXi6GK