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).

Catching up to timely #817, #818, #819 (folded in from #874).
Since #817, InputCapability::time() requires TotalOrder and returns Option, retain is retain_least, and inspect is inspect_core.

• Collection::inspect_batch hands the closure the batch's Stamp<T>; Collection::inspect_container is inspect_core.
• Collection::delay no longer uses timely's delay_batch (removed in dogsdogsdogs: one lookup primitive under the delta-join operators #818). It delays each stamp element and each update's time through func and sends at once under the delayed capabilities, which works for partial orders; the T: Hash bound is gone.
• capture's CDC progress operator holds a CapabilitySet from retain_stamp; upsert takes retain_least if present; the diagnostics loggers take least() if present.
• dogsdogsdogs half_join inserts each element of retain_stamp.
• dogsdogsdogs: one lookup primitive under the delta-join operators #818 removed timely's count and delay: the dogsdogsdogs examples count with DD's count, and graphs-static no longer delays records to their times (the reduce acts on record times).
• Comments no longer describe the removed unary_notify pattern (Cursors migration #819).

Workspace tests (all targets), doctests, clippy, and the explain release test pass against timely master at 4eb3ee09.

🤖 Generated with Claude Code

https://claude.ai/code/session_01SFY2AqDJTLX9TRGRuGR8dQ