Good regression test. Two small extensions worth considering: (1) exercise the iterate_with_conditions (plural) path too — that variant folds an .iter().all(...) across a Vec<Condition> inside the same Consensus, and its per-worker eagerness is a separate code path; (2) the current test loops 5×; on my mental model that is enough to shake most races out, but if you have a #[cfg(feature = "stress")] or similar knob elsewhere in the crate, cranking it to a few hundred for the CI stress job would be cheap.

For now, I leave it as is and let others decide if that's compatible with the CI pipeline here or not.

I thought that rust comes with some very nice benchmarking tools like Criterion. Why not use these?

At some point I'd love to this but I'd do that in a separate PR. So far I think it is educational to see how to manually construct an equivalent circuit to the ones created by the recursive API through using the lower-level iterative API. If you don't want to see any stats as an indication of what's going on behind the scenes, I'm fine with deleting them. However, I think especially for tutorial 10 the performance gain from dropping the distinct is part of the tutorial.

I don't understand why you are taking a functional tutorial and turning it into a broken example.
Maybe create a separate "broken-example" fille and leave this one alone

No, the tutorial works just as before! Nothing is broken and in fact, not much has changed from before from a logical point of view. The computation does not terminate only if people include the data from the third transaction/step (not sure about the best terminology here; a year ago when I first contributed these tutorials the concept of a transaction didn't exist and these were usually referred to as steps).

Tutorial 10 and 11 always operated on top of the same data to make the termination issue very clear. The only difference now is that instead of having had the third transaction commented out in tutorial 10 (for people to play around with), the data for both tutorial 10 and 11 is now shared in the transitive_closure module to not have it duplicated anymore. And instead of commenting stuff, you can now control the inclusion of the third transaction with the STEPS constant.

Why not put this into a separate file?

The iterate_variants and recursive_variants of tutorial 10, 11, and 13 show two approaches to construct circuits for the exact same problems: transitive closure with acyclic data (tutorial 10), transitive closure with cyclic data (tutorial 11), as well as mutual recursion and graph coloring (tutorial 13). Now, for each tutorial we show how to construct such a circuit using the higher-level recursive API and the lower-level iterative API. I'm fond of having each tutorial deal with one problem and possibly show multiple variants of how to approach that problem with DBSP circuits. Hence, both variants still live in the same file.

Of course we can structure things following a different paradigm but I tend to write the tutorials in a way I had loved to read them when I started figuring out how to compute these problems with DBSP.

This is a pure alias for the already-public check_fixedpoint (declared on this trait at line 1864, plumbed through everywhere else). Adding a second name for the same operation isn't free — users hit two docs.rs entries doing the same thing and have to guess which is canonical. If the goal is a friendlier is_* predicate name for the iterate-based API, consider renaming check_fixedpoint → is_fixedpoint outright (Rust convention for -> bool predicates) and moving this docstring onto that method. If you want to keep the existing name, at least drop this alias and add the extra prose to check_fixedpoint — one canonical entry point beats two. (Also called out in the PR description as an open question for @ryzhyk.)

I'll leave that up to @ryzhyk to decide! I have no preference here.

Nit: flake.nix and flake.lock are project-level artifacts, not personal ones. If the team ever adds a Nix flake for the repo, ignoring them by default will silently hide someone's legitimate check-in. Personal Nix bits usually live under .envrc/.direnv (already covered here) or a locally-git-excluded flake.local.nix per convention. Consider dropping the two flake.* lines and letting anyone who wants a private flake add it to .git/info/exclude locally.

I've deleted the two flake* lines and excluded them locally.

Is there interest in a flake.nix file for this project? I have one on my machine which sets up everything to have a dev environment for DBSP but not for any Java stuff required for the compiler so far. If you want, I can include this as a starting point.. Let me know!

Is this the wrong place to mark the stream distinct?

I recommend using the profiler in the webui to inspect the generated circuit, you will see exactly all the lowest-level operators that are being used.

@mihaibudiu , there is no webui for DBSP circuits.

You can use DBSPHandle::dump_profile to dump the profile in the graphviz format. Please feel free to share it here. My guess is the mark_distinct annotation is in the right place, but the annotation itself may not be accurate, i.e., the output is not distinct, and the circuit iterates more without the distinct call.

And I was hoping to retire this API...

This is the profile of the run with 500 nodes on the recursive circuit. It looks like the distinct is still applied.

Profile of worker 0:

Full profile of all four workers profile.zip.

Claude found the issue (see below). I am not 100% sure about the output of join being sharded, but the rest of it is correct. My takeaway is that mark_distinct is a low-level API and probably shouldn't be used in user code. I like the idea of a distinct-less variant of recursive. We should probably add that instead of hacking around it.

The layer mismatch

recursive is a thin typed wrapper. The user closure returns a typed stream, and recursive immediately erases it with .inner() before handing it to dyn_recursive:

benchmark closure (TYPED)                     recursive.rs:235          dyn_recursive
──────────────────────────                    ───────────────           ─────────────
len_1.plus(..).mark_distinct()   ──►  .map(|s| s.inner())   ──►   S::distinct(stream)   (recursive.rs:252)
   writes key                          erases TYPED→DYN            = Stream::dyn_distinct  (dyn/recursive.rs:77)
   DistinctIncrementalId<                (same StreamId,             reads key
     _, OrdIndexedZSet<Path,..>>          payload type erased)       DistinctIncrementalId<_, DynInner>
        ▲                                                                     ▲
        └──────────────── different key type → cache MISS ───────────────────┘

- Accumulator = Stream<NestedCircuit, OrdIndexedZSet<Path, TransClosurePath>> (typed; Path/TransClosurePath are concrete Tup2/Tup4, not DynData). .mark_distinct() records DistinctIncrementalId<_, OrdIndexedZSet<Path,TransClosurePath>>.
- recursive (recursive.rs:235) calls streams.inner(), which is transmute_payload (circuit_builder.rs:781). It keeps the StreamId (line 784) but changes the payload type D to the erased dynamic batch.
- dyn_recursive then runs S::distinct (dyn/recursive.rs:252 → 77 → Stream::dyn_distinct), which looks up DistinctIncrementalId<_, DynInner>.

Same StreamId, different type parameter, different bucket. The mark is never consulted. This is also why the typed .distinct() itself is written as self.inner().dyn_distinct().typed() (operator/distinct.rs:42): the whole distinct/mark machinery lives at the dynamic layer, and a mark placed on a typed stream is invisible to it.

A second, independent trap (also live here)

Even if the mark were placed on the dynamic stream, dyn_distinct keys on self.try_sharded_version().stream_id() (distinct.rs:353), i.e. the sharded sibling, not the stream you marked. With WORKERS = 4 the join output is sharded, so the distinct's input carries a different StreamId than the unsharded stream. The mark would have to land on the sharded version (mirroring dyn_distinct_inner, which does .mark_sharded().mark_distinct() at distinct.rs:339-340).

So: bug or misuse?

Both, arguably. It is misuse in that mark_distinct() only works within a single type layer, and recursive crosses that layer internally, so no placement in user code reaches the exact key. It is a footgun in that mark_distinct() is public on typed streams and silently no-ops there.

Fixes worth proposing

1. Explicit opt-out in the recursion API (cleanest, matches ryzhyk's "add a bounded_recursion operator" and issue #6666): let the caller tell dyn_recursive to skip the S::distinct step (dyn/recursive.rs:252) instead of threading a mark through two boundaries. This is exactly what the low-level iterate variant already achieves by not calling distinct at all, which is why it benchmarks 1.5-1.6x faster.
2. Make the typed API honor the mark: have typed recursive forward the mark across .inner() and onto the post-shard stream, or have RecursiveStreams::distinct short-circuit when the dynamic, sharded stream is_distinct(). dyn_distinct already skips a correctly-placed dynamic mark, so this only needs the mark on the right key.

Want me to prototype option 1 (a skip_distinct/recursive_multiset variant) or write a failing regression test that pins the current no-op behavior?

Still two public names for the same operation (is_fixedpoint here, check_fixedpoint at line 1963). I know this is punted to #6666 — just flagging that the alias survived the reset. If ryzhyk is fine with it, so am I; if not, one name should win before this lands.

Nice benchmarks overall — deterministic seeding, correctness assertions before measuring, throughput in closure-paths/sec. The forest-only restriction is well-justified in the generate_forest docstring. One optional nit: consider adding a brief ## Running section to the module doc (or a top-level benches/README.md) noting the --bench invocations for all three benchmarks, since new contributors may not know they exist.