dbsp: Give every stream in a nested circuit an id of its own - #6853

Stream ids are documented as globally unique, and Edges bookkeeping, the
replay machinery and several circuit caches key on them. They are not unique.

A child circuit receives a copy of the parent's counter rather than the counter:

fn last_stream_id(&self) -> RefCell<StreamId> {
    self.inner().last_stream_id.clone()   // clones the StreamId inside the cell
}

so the child hands out ids that the parent then hands out again. One of the ids
a child allocates belongs to a stream in the parent:
FeedbackOutputNode::with_export allocates the id of the stream a scope exports
from inside the child.

What it broke

CircuitHandle::complete_replay ends a bootstrap by deleting the replay edges
by stream id:

self.circuit.edges_mut().delete_stream(*stream_id);

On 0.291.1 (the 6765 fixes backported, so operators inside recursive scopes
checkpoint), a recursive pipeline resumed from a checkpoint panicked at
circuit_builder.rs:144, Option::unwrap() on None in StreamValue::peek,
at 2, 4 and 8 workers, immediately after "Bootstrap complete".

Instrumenting the schedule showed why. Each of the program's 12 recursive scopes
exported a stream whose id had also gone to a root level stream created later,
in every case the replay stream of the accumulate trace that consumes the
export:

StreamId(568) produced by ['579', '621']       # 579 Subcircuit, 621 Z1 (trace)
  n579 -> n580 stream=Some(StreamId(568))      # the live export edge
  n621 -> n620 stream=Some(StreamId(568))      # the replay edge

Deleting the replay edges took n579 -> n580 with them, so the scheduler
prepared Consolidate [580] with no predecessors:

 before:  n579 preds=[…] succs=["n580"]     n580 preds=["n579"] succs=[…]
 after:   n579 preds=[…] succs=[]           n580 preds=[]       succs=[…]

From then on the Consolidate ran in the first batch of every step, ahead of the
scope, and read an export stream that nothing had written:

MISSING STREAM VALUE: consumer="[580] Consolidate" tokens=0 consumers=1
 || COMPLETE_REPLAY
 || TRANSACTION
   [40] Z1 (trace)  … 6 more Z1 (trace)
   [580] Consolidate            <- no [579] Subcircuit ahead of it

Whether a program collides depends on how its ids fall, which is why main has
not tripped over it: the same program bootstraps cleanly here. The defect is the
same.

The fix

Share the counter, as the comment on last_stream_id always claimed. Stream ids
exist only at runtime; nothing on disk changes, and no fingerprint or persistent
id derives from them.

Testing

• New unit test asserting the invariant that broke: no two producers may share a
  stream id. It fails without the fix, on this branch and on main:
  stream s5 is produced by both n1 and n5.
• cargo test -p dbsp --lib, 435 tests including the replay tests: pass (run in
  the 0.291.1 tree where the fix originated).
• On the customer's program at 4 workers, the upgrade that panicked now
  completes: bootstrap finishes and result matches a reference run row for row
  after each of the following input batches.

🤖 Generated with Claude Code