diff --git a/quest/m1/route-stale-copies.md b/quest/m1/route-stale-copies.md deleted file mode 100644 index f035afce3a..0000000000 --- a/quest/m1/route-stale-copies.md +++ /dev/null @@ -1,169 +0,0 @@ -# [L] A prototype settles what keeps a route free of stale copies - -## Goal - -A test-only router beside `route::tests::Sim` runs lite-07 routing on paths -alone: a relay uses any route whose path does not name it, and a withdrawal -names the hop that died, so every route through that hop goes at once. The -same seeded meshes and scripts run on it and on today's router, which keeps -Babel's distance. The PR reports for both: - -- whether selected routes ever loop, and for how many steps; -- for how many steps a relay still shows a route whose path is gone, after a - link closes, a relay dies, and a publisher leaves; -- how many route messages each of those changes costs on 3, 6, 12, and 24 - relays; -- how many steps a relay with a longer way round takes to use it; -- whether a relay can pull a broadcast from a neighbour that already carries - it and is no closer to the publisher than the relay. - -The PR recommends one of the two rules, and the maintainer's pick is recorded -in the [epic](/quest/m2/cluster-routing/routes/README.md)'s Plan. Each bug the -reviews of [moq-dev/moq#8487](https://github.com/moq-dev/moq/pull/8487) found -is pinned as a `residual_` test on today's router. The production router, the -wire, and the draft do not change here. - -## Plan - -Decided 2026-11-11 (maintainer, /quest-plan after #8487's two adversarial -reviews, revised the same day once the first plan's rule was tried on the -router). Moved here from the routes epic the same day (maintainer): the epic -and m2 wait on m1 quests this needs nothing from, it changes tests only, and -the m1 model quests wait on its outcome. - -Why: today a relay believes a neighbour's copy of a publisher's claims only -while that neighbour is "feasible", which means strictly cheaper than the -lowest cost the relay ever advertised (Babel's distance). That set moves -whenever a cost or a seqno moves, and the relay's claims move with it. The -reviews found no loop and no stuck relay in route selection, and no remaining -bug on a mesh where every relay may see everything. Every finding needs -neighbours that know different parts of a publisher's claims, which happens -when a relay asks a neighbour for less than it wants because its grant on -that session is narrower. Call the prefix it asked for the **scope**. - -What was tried, each as a few lines behind a switch on `main` at `688312b9eb`: - -| | Today | Believe any neighbour whose route did not cross the relay | Believe a neighbour cheaper than the relay is now, ignoring the seqno, and keep what it sent | -| --- | --- | --- | --- | -| The seqno storm mesh settles | no | no | yes | -| A returning neighbour brings its claims back | yes | yes | yes | -| A seqno advance changes nothing shown | yes | yes | yes | -| Two relays behind a scoped relay let a dead claim go | yes | no, kept for good | yes | -| The blind relay shows a farther neighbour's claim | no | yes, as available, with no route to request it | no | -| The rest of `moq-net` | pass | 1 fails | 1 fails | - -On lite-06 (`release`), over real sessions, the blind relay sees and resolves -the far claim, and the two relays retract the dead one once. lite-06 carries -a path on every announcement, so believing by path is sound there. lite-07 -has one path per publisher, which cannot vouch for claims that reached the -sender another way. - -Settled: - -1. ✅ **A route is shared per publisher and per scope, and a relay believes a - route by that route's own path.** Every claim on a route is then known - completely for the route's prefix, and came along the path the route - names. Where every relay may see everything this is one route per - publisher, as today. - [The model](/quest/m2/cluster-routing/routes/scope-model.md) and - [the router](/quest/m2/cluster-routing/routes/scope.md) build it. Rejected: - believing by the publisher's one path, the first plan, which the table - shows is unsound. Rejected: believing by cost alone, the third column, - which fixed the storm in 25 lines with no wire change; the maintainer does - not want belief to rest on distance, and the blind relay stays blind. - Rejected: a path per broadcast as lite-06 has, which re-announces every - broadcast when a path changes. -2. ✅ **The blind relay is fixed by routing, not by showing more.** A relay - whose closest neighbour is scoped does not show what a farther neighbour - knows, and today that is consistent: it believes exactly the neighbours it - can forward a request to, and the farther one is not feasible. With a - route per scope the wider route is its own destination, so the relay can - both show the claim and reach it. -3. ✅ **Only a broadcast some relay carries gets its own route, and that - route propagates.** This reopens - [cache-aware routing](/quest/m3/cache-routing/README.md)'s one-hop warm - route (its Plan item 8). -4. ✅ **What keeps a route free of loops and stale copies is this quest's to - measure, not yet decided.** The hops stop loops, not stale copies. - Two candidates: - - **Distance, as today.** A relay uses a neighbour only when the - neighbour's cost is below the lowest cost the relay ever advertised. - Anything routed through the relay costs more than that, so such a - neighbour's route is no copy of the relay's own. It can still be stale - another way: a neighbour whose path closed stays usable until its - withdrawal arrives. It never loops, even for a moment. It refuses - good routes: a longer way round waits for the publisher to advance its - seqno, and a relay cannot join a copy carried by a neighbour that is no - closer to the publisher, which is the join cache-aware routing wants. - The seqno, ROUTE_REQUEST, ROUTE_HOLD, and the ROUTE_END echo all exist - to recover what this check refuses. - - **Paths alone, with a withdrawal that names what died.** A relay uses - any route whose path does not name it. Each hop in a path names the - relay and the session it learned the route on. When that session - closes, the relay's withdrawal names that hop, and every relay drops - every route through it at once, whichever neighbour offered it. This is - root cause notification from BGP research - ([Pei et al. 2005](https://www.cs.colostate.edu/~massey/pubs/journal/massey_comnet05.pdf)), - which bounds convergence by the mesh's diameter. Its known gap: when a - relay with several neighbours dies, each neighbour reports its own - session, so one stale switch per neighbour can remain. A route can also - loop for a moment while a path change is in flight. It needs no seqno, - request, hold, or echo. lite-06 instead holds stale paths back with a - 300 ms hold and a 2 s quarantine, and - [moq-dev/moq#5375](https://github.com/moq-dev/moq/pull/5375) still - measured stale revivals. - - The October measurement that chose Babel compared it with a path per - broadcast, not with shared routes, so its message counts do not answer - this. -5. ✅ **No routes to relays.** A relay keeps the cheapest path to each - publisher's scope, not to a relay that publisher is attached to. - -Work: - -- Write the paths-alone router as test-only code under `route/tests/`, as - small as it can be: routes shared per publisher, each with a path and a - cost; the withdrawal that names a hop; and claims that ride a route with - one floor, newest floor winning. Leave out what the comparison does not - need: scoped grants, qualified keys, interest changes, GOAWAY, and RTT. -- A relay remembers a dead hop only while one of its neighbours still offers - a route through it. Report whether a stale route can return after that, - and for how long. -- Drive both routers from one script generator, the one behind - `sweep_no_route_outgrows_a_loop_free_path`, and from the mesh shapes of - `rs/moq-net/tests/integration/mesh_sweep.rs`'s benchmark. Check at every - step and at every quiet point, and count as the Goal lists. -- Report what a relay's reader sees when its only route goes. Today it holds - the claims as unavailable while a seqno request is out - ([epic](/quest/m2/cluster-routing/routes/README.md) Plan item 19); on - paths alone there is nothing to wait for, so it lets go at once. -- Build the cached-copy case for both: a relay whose neighbour carries a - broadcast and is farther from the publisher. Two relays that each carry a - copy must not adopt each other, and a copy that goes must take every route - through it along. -- Pin each finding as a `residual_` test in `route/tests/`, asserting what - today's router does, with a comment naming - [the router quest](/quest/m2/cluster-routing/routes/scope.md): the seqno - storm ([reproducer](https://github.com/moq-dev/moq/pull/8487#issuecomment-6798722861)), - the forgotten claim and the two tests its failed fix broke - ([reproducers](https://github.com/moq-dev/moq/pull/8487#issuecomment-6802083874)), - and the blind relay: a publisher behind a relay that asked for `a` and - behind a two-relay chain that asked for everything, both joined to one - relay, which shows `a/x` and not `b/x`. -- Add the dead claim as a plain test, since today's router passes it and any - new rule must: a relay whose only way to the publisher asked for `a`, with - two relays joined to it and to each other, one of which also has a dear - link straight to the publisher. That link closes and the publisher - withdraws `b/x`; neither relay may still show it. -- Record the maintainer's pick in the epic's Plan item 26 and in - [Cluster routing](/quest/m2/cluster-routing/README.md)'s "Keep Babel" - bullet. If the pick is paths alone, plan removing the distance, the seqno, - ROUTE_REQUEST, ROUTE_HOLD, and the echo with `/quest-plan` before the - router quest starts. - -Public API: none. Wire: none. - -## Related - -- [The route model has a route per scope](/quest/m2/cluster-routing/routes/scope-model.md) - models the rule this picks -- [Cache-aware routing](/quest/m3/cache-routing/README.md) - its warm routes rest on the rule this picks diff --git a/quest/m2/cache-routing-proof.md b/quest/m2/cache-routing-proof.md index 985f83a9d4..9f2fe43472 100644 --- a/quest/m2/cache-routing-proof.md +++ b/quest/m2/cache-routing-proof.md @@ -29,13 +29,18 @@ second trip to the maintainer for choosing lite-07. Reopened 2026-11-11 (maintainer, /quest-plan on the route redesign, epic Plan item 8): a warm route propagates, and the rule that keeps it free of -loops and stale copies is -[the prototype](/quest/m1/route-stale-copies.md)'s to pick. -The eligibility rule and the fixtures below are for the one-hop warm route -first decided, which [moq-dev/moq#8457](https://github.com/moq-dev/moq/pull/8457) -proves. Once the prototype's pick is recorded, rewrite both for it: the -fixtures keep their meshes, and gain a relay that joins a copy at a neighbour -farther from the publisher than itself. +loops and stale copies is the route's path +([routes epic](/quest/m2/cluster-routing/routes/README.md) Plan item 27, +picked 2026-11-11 on +[moq-dev/moq#8533](https://github.com/moq-dev/moq/pull/8533)). The eligibility rule +and the fixtures below are for the one-hop warm route first decided, which +[moq-dev/moq#8457](https://github.com/moq-dev/moq/pull/8457) proves. Rewrite +both for the pick: the fixtures keep their meshes, and gain a relay that +joins a copy at a neighbour farther from the publisher than itself. Start +from `rs/moq-net/src/route/tests/copies.rs`, which runs that join, two +relays that start carrying together, and a copy that goes, on the prototype +router. It ranks carriers by a hash of the broadcast and the carrier; the +proof settles whether that rank is the rule. Build on the sans-IO `Router` in `rs/moq-net/src/route/`, which each origin's `route::Node` drives, and the harness behind its handoff sweeps, @@ -80,10 +85,6 @@ Route ID, and the instance, and say which stream carries it and how a peer that never sends one behaves. Size it against lite-07 and an additive SETUP extension and record the choice and its reason in the epic's Plan. -## Required - -- [A prototype settles what keeps a route free of stale copies](/quest/m1/route-stale-copies.md) - the rule a propagating warm route rests on, which replaces the one this proved - ## Related - [Routes and announces](/quest/m2/cluster-routing/routes/README.md) - the feasibility rules and handoff sweeps this extends diff --git a/quest/m2/cluster-routing/README.md b/quest/m2/cluster-routing/README.md index 220d37e7d7..cf4311089a 100644 --- a/quest/m2/cluster-routing/README.md +++ b/quest/m2/cluster-routing/README.md @@ -39,16 +39,19 @@ Decided with the maintainer, 2026-10-13: detection, additional routing recovery, and total interruption separately. Reopened 2026-11-11 (maintainer, /quest-plan after [moq-dev/moq#8487](https://github.com/moq-dev/moq/pull/8487)'s reviews): - whether Babel's distance stays is - [the prototype](/quest/m1/route-stale-copies.md)'s to - measure, against paths alone with a withdrawal that names the hop that - died. The distance refuses a longer way round until the publisher advances - its seqno, and refuses a cached copy at a neighbour no closer to the + the distance refuses a longer way round until the publisher advances its + seqno, and refuses a cached copy at a neighbour no closer to the publisher, which [cache-aware routing](/quest/m3/cache-routing/README.md) wants to join. Decided the same day: a route is shared per publisher and - per scope, and a relay believes a route by that route's own path, whichever - rule wins. Rejected: link-state among relays, and routes to relays in - place of publishers. + per scope, and a relay believes a route by that route's own path. Rejected: + link-state among relays, and routes to relays in place of publishers. + Decided 2026-11-11 (maintainer, on the measurements of + [moq-dev/moq#8533](https://github.com/moq-dev/moq/pull/8533)): Babel goes. A route is + kept free of loops by its path, a withdrawal that names the hop that died, + holds, and a timed doubt + ([routes epic](/quest/m2/cluster-routing/routes/README.md) Plan item 27). A + detour no longer waits on the publisher, and today's router keeps Babel + until [the router quest](/quest/m2/cluster-routing/routes/scope.md) lands. - An origin declares its own routing identity, fresh per process or tab restart and shared across its live sessions. It is unverified. Decided 2026-11-02 (maintainer, /quest-unblock): no key proof is planned; if a use case needs a provable identity, moq-lite-07's diff --git a/quest/m2/cluster-routing/routes/README.md b/quest/m2/cluster-routing/routes/README.md index 39e54ae09d..190e37f478 100644 --- a/quest/m2/cluster-routing/routes/README.md +++ b/quest/m2/cluster-routing/routes/README.md @@ -43,7 +43,15 @@ The items below were decided 2026-11-10 (maintainer, /quest-plan on lite-07 rout 25. ✅ Asked how a relay safely forgets a publisher that is gone, after the model showed a per-neighbour echo is not enough once the origin sits behind a relay: **each route names the nodes it crossed, and a node never uses a route that names itself** ([moq-dev/moq#8487](https://github.com/moq-dev/moq/pull/8487), which holds the counterexamples and the model results). Feasibility still selects routes; the list is one per origin route, not the per-claim Hop ID list item 6 removed, and the echo and the drain rule stay as they are. As built: Babel's feasibility condition is unchanged, and a node never selects or certifies through such a route. Where the condition refuses it, it still counts as infeasible, so a relay whose neighbours all route through it holds its claims while one of them finds another way. Under a qualified key it counts only while its hops extend those of the route the node last sent under that key: a copy that a scoped peer hands back under a key derived from the node's own route never moves, so nothing waits on it. The origin names itself too, so a rotated identity's route is refused where it began, and a node names itself by a Hop ID that a rotation leaves alone. The hop limit is the wire's 32 and no lower. A change of hops or RTT alone lets go of no region, unless it changes whether the node may use the route: by crossing it, or by reaching the hop limit. This supersedes the claim, in the handoff contract and in decision 6 below, that the echo alone makes forgetting safe. First decided as a relay echoing a ROUTE_END only once its own downstream had; the model ran that the same day and it fails when a session closes before answering. Rejected: that deferred echo; never dropping the distance; a timer; relays as the only origin nodes. -26. ✅ Decided 2026-11-11 (maintainer, /quest-plan after [moq-dev/moq#8487](https://github.com/moq-dev/moq/pull/8487)'s reviews), asked whose copy of a publisher's claims a relay believes: **a route is shared per publisher and per scope, and a relay believes a route by that route's own path** ([the prototype](/quest/m1/route-stale-copies.md), [the model](/quest/m2/cluster-routing/routes/scope-model.md), then [the router](/quest/m2/cluster-routing/routes/scope.md)). A scope is the prefix a relay asked a neighbour for when its grant there is narrower than its interest. Today only a feasible neighbour counts, one strictly cheaper than the distance the relay remembers, so a relay's claims move whenever a cost or seqno does: it can forget a claim for good, a mesh with two grants never settles after one seqno advance, and behind a scoped neighbour it never shows what a farther one knows. Only a broadcast some relay carries gets its own route, and that route propagates ([cache-aware routing](/quest/m3/cache-routing/README.md) Plan item 8). Open, for the prototype to measure and the maintainer to pick: whether Babel's distance or the path alone, with a withdrawal that names the hop that died, keeps a route free of loops and stale copies. The contract's bullet on which neighbours certify stands until the model quest rewrites it. First decided the same day as any neighbour whose path to the publisher does not pass through the relay; tried on the router, two relays behind a scoped relay then keep a dead claim alive through each other, and the relay behind a scoped neighbour shows a claim it has no route to request. Rejected: that rule; believing a neighbour that is cheaper than the relay is now, which fixed the storm and the forgotten claim with no wire change but rests belief on distance; a path per broadcast as lite-06 has; link-state among relays; routes to relays. +26. ✅ Decided 2026-11-11 (maintainer, /quest-plan after [moq-dev/moq#8487](https://github.com/moq-dev/moq/pull/8487)'s reviews), asked whose copy of a publisher's claims a relay believes: **a route is shared per publisher and per scope, and a relay believes a route by that route's own path** ([the prototype](https://github.com/moq-dev/moq/pull/8533), [the model](/quest/m2/cluster-routing/routes/scope-model.md), then [the router](/quest/m2/cluster-routing/routes/scope.md)). A scope is the prefix a relay asked a neighbour for when its grant there is narrower than its interest. Today only a feasible neighbour counts, one strictly cheaper than the distance the relay remembers, so a relay's claims move whenever a cost or seqno does: it can forget a claim for good, a mesh with two grants never settles after one seqno advance, and behind a scoped neighbour it never shows what a farther one knows. Only a broadcast some relay carries gets its own route, and that route propagates ([cache-aware routing](/quest/m3/cache-routing/README.md) Plan item 8). What keeps a route free of loops and stale copies is item 27. The contract's bullet on which neighbours certify stands until the model quest rewrites it. First decided the same day as any neighbour whose path to the publisher does not pass through the relay; tried on the router, two relays behind a scoped relay then keep a dead claim alive through each other, and the relay behind a scoped neighbour shows a claim it has no route to request. Rejected: that rule; believing a neighbour that is cheaper than the relay is now, which fixed the storm and the forgotten claim with no wire change but rests belief on distance; a path per broadcast as lite-06 has; link-state among relays; routes to relays. + +27. ✅ Decided 2026-11-11 (maintainer, on [moq-dev/moq#8533](https://github.com/moq-dev/moq/pull/8533), which measured it), asked what keeps a route free of loops and stale copies: **the route's path, a withdrawal that names the hop that died, holds, and a timed doubt. Babel's distance, the seqno, ROUTE_REQUEST, and the ROUTE_END echo go.** None of it is built: [the model](/quest/m2/cluster-routing/routes/scope-model.md) proves it, [the router](/quest/m2/cluster-routing/routes/scope.md) builds it, and the contract below describes today's router until they do. The rule, as the prototype in `rs/moq-net/src/route/tests/paths.rs` runs it: + - A hop names a relay and the session it learned the route on. A relay uses the cheapest route whose path does not name it, at once. + - A relay that loses a session names that hop, and every relay drops every route through it, whoever offered it. One message per session covers every publisher and lists the routes that moved to another path. + - A relay with no route left holds its claims, unavailable, until each neighbour that learned the route from it has answered: with another way, or with nothing once those behind it have answered. Then it lets go. + - A closed session casts doubt on the node at its far end, since a dead relay is reported one neighbour at a time. No relay moves to a route through that node until a timer lifts the doubt. The route a relay already uses is exempt. + + Measured on 24 relays (the mesh sweep's sparse shape, 16 seeds), against today's router, with the doubt lasting 4 hops of delay: a detour settles in 12 hops of delay where the distance takes 15.7, and in 6.2 where it does not run through the far end of the closed link; the publisher's relay dying costs 124 route messages where the distance costs 276, with no loop and 6 claims sent again; and a link closing in a clique costs 44 messages with one publisher behind it and 198 with eight, where the distance costs 621 and 5,024. Accepted: a detour through the far end of a closed link waits the timer out; selected routes can loop for a moment when two unrelated things fail at once (32 of 120,000 churn steps), which the distance never does; and holds and doubt have passed sweeps and no model. The timer's length is the model's and the router's to settle, from the maintainer's 1 s. Rejected: Babel's distance; paths with nothing else, which hunt through stale paths (65,219 messages for one publisher leaving a clique of 12); no doubt, which loops when a relay with several neighbours dies; and a doubt lifted by an epoch the node floods, which resolves in a couple of hops but costs a message per session on each relay-to-relay close ("Epoches are a bit scary in terms of messages"). The route layer's core landed in [moq-dev/moq#7676](https://github.com/moq-dev/moq/pull/7676): a diff --git a/quest/m2/cluster-routing/routes/scope-model.md b/quest/m2/cluster-routing/routes/scope-model.md index 8fe1a8d5d6..0348f642df 100644 --- a/quest/m2/cluster-routing/routes/scope-model.md +++ b/quest/m2/cluster-routing/routes/scope-model.md @@ -21,9 +21,11 @@ not cross. TLC shows on it: behind a scoped neighbour reaches what a farther, unscoped neighbour offers. -The model uses the rule -[the prototype](/quest/m1/route-stale-copies.md) picked for -keeping a route free of stale copies. The rule this settles is written into +The model keeps a route free of loops and stale copies by the rule the +[epic](/quest/m2/cluster-routing/routes/README.md)'s Plan item 27 picked: the +path, a withdrawal that names the hop that died, holds, and a timed doubt. +TLC also shows that every hold ends, and what bounds a loop of selected next +hops while messages are in flight. The rule this settles is written into the epic's contract for [the router](/quest/m2/cluster-routing/routes/scope.md) to implement. The router's behaviour, the wire, and the draft do not change here. @@ -31,9 +33,31 @@ router's behaviour, the wire, and the draft do not change here. ## Plan Decided 2026-11-11 (maintainer, /quest-plan). The decisions and what was -tried are in [the prototype](/quest/m1/route-stale-copies.md)'s -Plan and the [epic](/quest/m2/cluster-routing/routes/README.md)'s Plan item -26. +tried are in the [epic](/quest/m2/cluster-routing/routes/README.md)'s Plan +items 26 and 27, [moq-dev/moq#8514](https://github.com/moq-dev/moq/pull/8514), +and [moq-dev/moq#8533](https://github.com/moq-dev/moq/pull/8533). + +The rule for loops and stale copies has passed sweeps and no model. The +prototype in `rs/moq-net/src/route/tests/paths.rs` runs it (`Rules::waits`), +and `compare.rs` beside it holds the meshes. What the model must prove or +replace: + +- **Every hold ends.** A relay holds under the next hop it used, answers + that hop only once the neighbours that learned the route from it have + answered, and holds on its own account when that next hop's own route + names it. Two hangs the sweeps found are fixed in the prototype, and each + deserves a config: two relays that took each other's stale route each + waited on the other, and a relay answered "nothing" while a route only + doubt kept it off could still come back. +- **The doubt.** Model the timer as a step that may fire at any moment. Show + that firing late costs only the wait, and state what firing early costs: a + relay moves to a stale route, as with no doubt at all. +- **Loops.** Selected next hops form no loop once nothing is in flight. + While something is, find what bounds a loop, or the smallest change that + does: the churn sweep still loops on 32 of 120,000 steps, when two + unrelated things fail at once. +- **Forgetting.** A relay remembers a dead hop only while a neighbour's route + still names it. Show no stale route returns after that. Why a route per scope: a relay that asked a neighbour for prefix `a` knows a publisher's claims only inside `a`. Today that partial knowledge lives inside @@ -92,10 +116,6 @@ edits to the two prefixes this introduces. Public API: none. Wire: none. -## Required - -- [A prototype settles what keeps a route free of stale copies](/quest/m1/route-stale-copies.md) - the rule for loops and stale copies that this model is written on - ## Related - [The router keeps a route per scope](/quest/m2/cluster-routing/routes/scope.md) - implements the rule this settles diff --git a/quest/m2/cluster-routing/routes/scope.md b/quest/m2/cluster-routing/routes/scope.md index 5cdd4c56e6..6a64143a18 100644 --- a/quest/m2/cluster-routing/routes/scope.md +++ b/quest/m2/cluster-routing/routes/scope.md @@ -9,8 +9,8 @@ shows a claim exactly while such a route covers it and the newest revision a covering route gave it still has the claim, two relays never keep a dead route alive through each other, a change of path or seqno alone changes no claim a reader sees, and a relay behind a scoped neighbour reaches what a -farther one offers. The `residual_` tests -[the prototype](/quest/m1/route-stale-copies.md) pinned pass +farther one offers. The `residual_` tests in +`rs/moq-net/src/route/tests/residual.rs` pass as plain tests, a sweep over meshes with scoped relays holds the rules at every quiet point, and the draft and the epic's contract say what the router does. @@ -22,6 +22,18 @@ for stale copies, the model settles the route per scope, and the router comes last (rejected: the router first, patched case by case; recording the findings as known limits and waiting for federation). +Decided 2026-11-11 (maintainer, on +[moq-dev/moq#8533](https://github.com/moq-dev/moq/pull/8533)): the route rule +changes too. Babel's distance, the seqno, ROUTE_REQUEST, and the ROUTE_END +echo go, for the rule in the +[epic](/quest/m2/cluster-routing/routes/README.md)'s Plan item 27. That +removal is not planned yet. Plan it with `/quest-plan` before this quest +starts, as its own quest or as work here: it touches the wire (a session in +each hop, a message that names a dead hop, no ROUTE_REQUEST and no echo), +the draft, `js/net`, and most of `route/mod.rs`. The prototype +(`route/tests/paths.rs`, `compare.rs`, `copies.rs`) is then promoted or +deleted. + Work: - Key `rx`, the selection, `sent`, and the views in `route/mod.rs` by route. @@ -48,9 +60,8 @@ Work: - Draft: Routing says a route covers a prefix, and Floors says a floor certifies its route's prefix. Replace the Floors condition "every peer whose last advertised route to it is feasible here has answered the - current request" with the settled rule. The hold rule in Routing is not - about belief and stays unless the prototype's pick removes holds. Add a - changelog bullet. + current request" with the settled rule. The hold rule in Routing changes + with the route rule (epic Plan item 27). Add a changelog bullet. - If [js/net's route layer](/quest/m2/cluster-routing/routes/js.md) has landed, change it in the same PR. - Rewrite the epic's bullet on who certifies as built, and update the diff --git a/quest/m3/cache-routing/README.md b/quest/m3/cache-routing/README.md index 20e34f08aa..6e621c0794 100644 --- a/quest/m3/cache-routing/README.md +++ b/quest/m3/cache-routing/README.md @@ -85,10 +85,20 @@ Decided 2026-11-08 (maintainer, /quest-plan): broadcast no relay carries rides the shared route and costs no message of its own. Item 4 rests safety on Babel's distance, which cannot let a relay join a copy at a neighbour that is no closer to the publisher: the join - the Goal wants, and why item 2 stopped at one hop. What keeps a warm route - free of loops and stale copies is - [the prototype](/quest/m1/route-stale-copies.md)'s to - measure, and the proof then holds that rule. Items 1 and 3 stand. + the Goal wants, and why item 2 stopped at one hop. Items 1 and 3 stand. + Decided 2026-11-11 (maintainer, on + [moq-dev/moq#8533](https://github.com/moq-dev/moq/pull/8533)): what keeps a warm + route free of loops and stale copies is its path, as for every route + (routes epic Plan item 27), and the proof then holds that rule. The + prototype in `rs/moq-net/src/route/tests/copies.rs` showed three things + the proof starts from. A warm route carries the path its copy took and + names the relay that carries it, and a relay refuses one whose path names + it, so it joins a copy at a neighbour farther from the publisher than + itself. Two relays that start carrying in the same instant trade copies + for good unless they are ranked, so a relay already pulling moves onto + another's copy only if that carrier ranks below it. And when a carrier's + uplink closes, it and a relay pulling from it point at each other until + the news crosses their link. [moq-dev/moq#8457](https://github.com/moq-dev/moq/pull/8457) proves items 2 and 4 as first decided. diff --git a/quest/m3/route-keep-next-hop.md b/quest/m3/route-keep-next-hop.md index 29e004f267..09d60fca8a 100644 --- a/quest/m3/route-keep-next-hop.md +++ b/quest/m3/route-keep-next-hop.md @@ -39,9 +39,10 @@ selection. down-only route still moves to an equal one it may export, and test that. - Check that every router test that names the next hop it expects still means what it says, rather than editing expectations to match. -- [The prototype](/quest/m1/route-stale-copies.md) may - replace how a route is selected. If it has, apply this to the rule that - stands. +- How a route is selected is being replaced + ([routes epic](/quest/m2/cluster-routing/routes/README.md) Plan item 27). + Apply this to the rule that stands when this starts. The prototype in + `route/tests/paths.rs` already stays on its next hop among equals. - Measure with `Sim`'s `hop_updates` count and, if it has landed, `mesh_sweep.rs`'s benchmark, before and after. diff --git a/rs/moq-net/src/route/tests/compare.rs b/rs/moq-net/src/route/tests/compare.rs new file mode 100644 index 0000000000..9e3e6c43f7 --- /dev/null +++ b/rs/moq-net/src/route/tests/compare.rs @@ -0,0 +1,1079 @@ +//! One script on two routers: [`Router`], which keeps Babel's distance, and the +//! prototype in `paths.rs`, which keeps a route loop-free by its path alone. +//! +//! A [`Lab`] builds the same mesh on either, applies one change at a time, and +//! delivers in rounds: every message queued when a round starts crosses its link +//! in that round, so a round is one hop of delay. After each delivery it follows +//! every relay's next hops, and after each round it counts the relays whose route +//! does not lead to the publisher and those with no route while a path exists. +//! `benchmark_route_rules` prints the table; the tests pin what it shows. + +use std::collections::BTreeSet; + +use super::paths::{PathSim, Rules}; +use super::*; + +/// A queue of messages, or of echoed ROUTE_ENDs, from one node to another. +#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)] +pub(super) enum Chan { + Msg(Node, Node), + Ack(Node, Node), +} + +/// What a router sent and showed so far. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub(super) struct Sent { + /// Route messages of every kind, requests and echoes included. + pub msgs: usize, + /// Routes those messages name. + pub items: usize, + /// Claims sent, each once per session it crossed. + pub claims: usize, + /// Claims a reader saw go offline: its relay let go of their route. + pub offline: usize, + /// Claims a reader saw turn unavailable: its relay holds their route. + pub unavailable: usize, + /// Claims a reader saw start. + pub starts: usize, +} + +impl std::ops::Sub for Sent { + type Output = Self; + + fn sub(self, was: Self) -> Self { + Self { + msgs: self.msgs - was.msgs, + items: self.items - was.items, + claims: self.claims - was.claims, + offline: self.offline - was.offline, + unavailable: self.unavailable - was.unavailable, + starts: self.starts - was.starts, + } + } +} + +/// A simulated network either router runs in. +pub(super) trait Net { + /// `node` passes on no route it learned, like a client. + fn client(&mut self, node: Node); + fn join(&mut self, a: Node, b: Node, cost: Metric); + fn open(&mut self, a: Node, b: Node); + fn close(&mut self, a: Node, b: Node); + /// The process at `node` is gone: nothing more passes between it and `peer`, + /// either way, though `peer` has yet to notice. + fn mute(&mut self, node: Node, peer: Node); + /// `peer` notices its session to `node` closed. `node`, which is gone, does nothing. + fn sever(&mut self, node: Node, peer: Node); + /// `node` restarts with no sessions and no state. + fn reset(&mut self, node: Node); + fn publish(&mut self, node: Node, id: Id); + fn claim(&mut self, node: Node, claim: &str, live: bool); + /// The queues with something to deliver, and how much. + fn ready(&self) -> Vec<(Chan, usize)>; + fn step(&mut self, chan: Chan); + /// One hop of delay passes. Whether a timer still runs. + fn tick(&mut self) -> bool { + false + } + /// What `node` holds for `id`, for a failure report. + fn dump(&self, _node: Node, _id: Id) -> String { + String::new() + } + /// The neighbour `node` routes `id` through: `Some(None)` at the publisher. + fn next(&self, node: Node, id: Id) -> Option>; + fn showing(&self, node: Node, id: Id, claim: &str) -> Option; + fn sent(&self) -> Sent; +} + +fn tally(kinds: impl Iterator) -> (usize, usize, usize) { + let (mut offline, mut unavailable, mut starts) = (0, 0, 0); + for kind in kinds { + match kind { + Kind::Offline => offline += 1, + Kind::Unavailable => unavailable += 1, + Kind::Start => starts += 1, + _ => {} + } + } + (offline, unavailable, starts) +} + +impl Net for Sim { + fn client(&mut self, node: Node) { + self.leaf(node); + } + + fn join(&mut self, a: Node, b: Node, cost: Metric) { + Sim::join(self, a, b, cost); + } + + fn open(&mut self, a: Node, b: Node) { + self.connect(a, b); + } + + fn close(&mut self, a: Node, b: Node) { + self.disconnect(a, b); + self.closed(a); + self.closed(b); + } + + fn mute(&mut self, node: Node, peer: Node) { + self.hold(node, peer); + self.hold(peer, node); + } + + fn sever(&mut self, node: Node, peer: Node) { + self.up.remove(&Self::pair(node, peer)); + for key in [(node, peer), (peer, node)] { + self.queues.remove(&key); + self.acks.remove(&key); + self.held.remove(&key); + } + self.nodes[peer].disconnect(node); + self.flush(); + self.closed(peer); + } + + fn reset(&mut self, node: Node) { + self.kill(node); + } + + fn publish(&mut self, node: Node, id: Id) { + self.originate(node, id, 0); + } + + fn claim(&mut self, node: Node, claim: &str, live: bool) { + match live { + true => self.announce(node, claim, 0, None), + false => self.unannounce(node, claim), + } + } + + fn ready(&self) -> Vec<(Chan, usize)> { + let msgs = (self.busy().into_iter()).map(|(from, to)| (Chan::Msg(from, to), self.queued(from, to))); + let acks = (self.busy_acks().into_iter()).map(|(from, to)| (Chan::Ack(from, to), self.queued_acks(from, to))); + let held = |chan: &Chan| match *chan { + Chan::Msg(from, to) | Chan::Ack(from, to) => self.held.contains(&(from, to)), + }; + msgs.chain(acks).filter(|(chan, _)| !held(chan)).collect() + } + + fn step(&mut self, chan: Chan) { + match chan { + Chan::Msg(from, to) => self.deliver(from, to), + Chan::Ack(from, to) => self.deliver_ack(from, to), + }; + } + + fn next(&self, node: Node, id: Id) -> Option> { + self.route(node, id).map(|route| route.next) + } + + fn showing(&self, node: Node, id: Id, claim: &str) -> Option { + self.shows(node, id, claim) + } + + fn sent(&self) -> Sent { + let (offline, unavailable, starts) = tally(self.events.iter().map(|event| event.kind)); + Sent { + msgs: self.stats.route_msgs + self.stats.requests, + items: self.stats.route_msgs + self.stats.requests, + claims: self.stats.claim_items, + offline, + unavailable, + starts, + } + } +} + +impl Sim { + /// A publisher advances its seqno when a session its route went out on closes, + /// as `route::Node` has it do. A new session lengthens no path, so it does not. + fn closed(&mut self, node: Node) { + if self.declared[node].is_some() { + self.bump(node); + } + } +} + +impl Net for PathSim { + fn client(&mut self, node: Node) { + self.leaf(node); + } + + fn join(&mut self, a: Node, b: Node, cost: Metric) { + PathSim::join(self, a, b, cost); + } + + fn open(&mut self, a: Node, b: Node) { + self.connect(a, b); + } + + fn close(&mut self, a: Node, b: Node) { + self.disconnect(a, b); + } + + fn mute(&mut self, node: Node, peer: Node) { + PathSim::mute(self, node, peer); + } + + fn sever(&mut self, node: Node, peer: Node) { + PathSim::sever(self, node, peer); + } + + fn reset(&mut self, node: Node) { + self.kill(node); + } + + fn publish(&mut self, node: Node, id: Id) { + self.originate(node, id); + } + + fn claim(&mut self, node: Node, claim: &str, live: bool) { + match live { + true => self.announce(node, claim), + false => self.unannounce(node, claim), + } + } + + fn ready(&self) -> Vec<(Chan, usize)> { + (self.busy().into_iter()) + .map(|(from, to)| (Chan::Msg(from, to), self.queued(from, to))) + .collect() + } + + fn dump(&self, node: Node, id: Id) -> String { + self.nodes[node].dump(id) + } + + fn step(&mut self, chan: Chan) { + let Chan::Msg(from, to) = chan else { + unreachable!("the prototype echoes nothing"); + }; + self.deliver(from, to); + } + + fn tick(&mut self) -> bool { + PathSim::tick(self) + } + + fn next(&self, node: Node, id: Id) -> Option> { + PathSim::next(self, node, id) + } + + fn showing(&self, node: Node, id: Id, claim: &str) -> Option { + self.shows(node, id, claim) + } + + fn sent(&self) -> Sent { + let (offline, unavailable, starts) = tally(self.events.iter().map(|(.., kind)| *kind)); + Sent { + msgs: self.stats.route_msgs, + items: self.stats.route_items, + claims: self.stats.claim_items, + offline, + unavailable, + starts, + } + } +} + +/// The routers compared, by the name the report gives each. +#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)] +pub(super) enum Rule { + /// Today's router: Babel's distance, seqno requests, holds, and echoes. + Distance, + /// Paths and nothing else. + Paths, + /// Paths, and a withdrawal that names the hop that died. + Causes, + /// Causes, one message per cause and session, and holds. + Holds, + /// Holds, and no move to a route through a node in doubt for 4 hops of delay. + Wait4, + /// The same for 32 hops of delay. + Wait32, + /// Holds, and no move to a route through a node in doubt until it advances its epoch. + Epochs, +} + +impl Rule { + pub const ALL: [Self; 7] = [ + Self::Distance, + Self::Paths, + Self::Causes, + Self::Holds, + Self::Wait4, + Self::Wait32, + Self::Epochs, + ]; + + pub fn as_str(self) -> &'static str { + match self { + Self::Distance => "distance", + Self::Paths => "paths", + Self::Causes => "paths+cause", + Self::Holds => "paths+cause+hold", + Self::Wait4 => "paths+cause+hold+wait4", + Self::Wait32 => "paths+cause+hold+wait32", + Self::Epochs => "paths+cause+hold+epoch", + } + } + + pub fn net(self, nodes: usize) -> Box { + match self { + Self::Distance => Box::new(Sim::new(nodes)), + Self::Paths => Box::new(PathSim::new(nodes, Rules::BARE)), + Self::Causes => Box::new(PathSim::new(nodes, Rules::CAUSES)), + Self::Holds => Box::new(PathSim::new(nodes, Rules::HOLDS)), + Self::Wait4 => Box::new(PathSim::new(nodes, Rules::waits(4))), + Self::Wait32 => Box::new(PathSim::new(nodes, Rules::waits(32))), + Self::Epochs => Box::new(PathSim::new(nodes, Rules::EPOCHS)), + } + } +} + +/// A mesh of relays, each publisher attached to one of them. +#[derive(Clone, Debug)] +pub(super) struct Mesh { + pub relays: usize, + pub links: Vec<(Node, Node, Metric)>, + /// The relay each publisher attaches to. + pub homes: Vec, + /// Claims each publisher announces. + pub claims: usize, +} + +impl Mesh { + /// A ring, the publisher on relay 0. + pub fn ring(relays: usize) -> Self { + let links = (0..relays).map(|a| (a, (a + 1) % relays, 1)).collect(); + Self { + relays, + links, + homes: vec![0], + claims: 4, + } + } + + /// Every relay linked to every other, with `publishers` on relay 0. + pub fn clique(relays: usize, publishers: usize) -> Self { + let links = (0..relays) + .flat_map(|a| (a + 1..relays).map(move |b| (a, b, 1))) + .collect(); + Self { + relays, + links, + homes: vec![0; publishers], + claims: 4, + } + } + + /// The mesh sweep's shape: a random spanning tree, then up to as many links + /// again, with two publishers on random relays. + pub fn sparse(relays: usize, rng: &mut XorShift) -> Self { + let mut linked = BTreeSet::new(); + for b in 1..relays { + linked.insert((rng.next(b as u64) as usize, b)); + } + for _ in 0..rng.next(relays as u64 + 1) { + let (a, b) = (rng.next(relays as u64) as usize, rng.next(relays as u64) as usize); + if a != b { + linked.insert((a.min(b), a.max(b))); + } + } + let links = linked.into_iter().map(|(a, b)| (a, b, 1 + rng.next(3))).collect(); + let homes = (0..2).map(|_| rng.next(relays as u64) as usize).collect(); + Self { + relays, + links, + homes, + claims: 4, + } + } + + pub fn publisher(&self, index: usize) -> (Node, Id) { + (self.relays + index, 100 + index as Id) + } + + fn claim(index: usize) -> String { + format!("live/{index}") + } +} + +/// How a relay's route fares against the mesh as it really is. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum Walk { + /// Its next hops lead to the publisher over open links. + Reached, + /// They stop at a relay with no route, or cross a closed link. + Broken, + Loop, +} + +/// What converging on one change cost. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub(super) struct Cost { + pub sent: Sent, + /// Hops of delay until nothing was left to deliver. + pub rounds: usize, + /// Deliveries after which some relay's next hops led round a loop. + pub loops: usize, + /// Relays, summed over rounds, whose route did not lead to the publisher. + pub stale: usize, + /// Relays, summed over rounds, with no route while a path to the publisher existed. + pub dark: usize, + /// Whether it settled at all. + pub settled: bool, +} + +pub(super) struct Lab { + pub net: Box, + pub mesh: Mesh, + /// Open links, relay to relay and publisher to relay. + up: BTreeSet<(Node, Node)>, + down: BTreeSet, + order: XorShift, + cost: Cost, +} + +fn pair(a: Node, b: Node) -> (Node, Node) { + (a.min(b), a.max(b)) +} + +impl Lab { + /// Build `mesh` on `rule`'s router, with every publisher announcing, and let it settle. + pub fn new(rule: Rule, mesh: &Mesh) -> Self { + let nodes = mesh.relays + mesh.homes.len(); + let mut lab = Self { + net: rule.net(nodes), + mesh: mesh.clone(), + up: BTreeSet::new(), + down: BTreeSet::new(), + order: XorShift(0x5eed_0001), + cost: Cost::default(), + }; + for &(a, b, cost) in &mesh.links { + lab.net.join(a, b, cost); + lab.up.insert(pair(a, b)); + } + lab.settle(); + for index in 0..mesh.homes.len() { + lab.arrive(index); + } + lab + } + + /// Publisher `index` joins its relay for the first time and announces its claims. + fn arrive(&mut self, index: usize) { + let (node, id) = self.mesh.publisher(index); + let home = self.mesh.homes[index]; + self.net.client(node); + self.net.join(node, home, 1); + self.up.insert(pair(node, home)); + self.net.publish(node, id); + for claim in 0..self.mesh.claims { + self.net.claim(node, &Mesh::claim(claim), true); + } + self.settle(); + } + + pub fn is_up(&self, a: Node, b: Node) -> bool { + self.up.contains(&pair(a, b)) + } + + fn peers(&self, node: Node) -> Vec { + (self.up.iter()) + .filter_map(|&(a, b)| match node { + _ if node == a => Some(b), + _ if node == b => Some(a), + _ => None, + }) + .collect() + } + + /// The relays with a path to publisher `index` over open links. + pub fn reach(&self, index: usize) -> BTreeSet { + self.reach_over(&self.up, index) + } + + fn reach_over(&self, up: &BTreeSet<(Node, Node)>, index: usize) -> BTreeSet { + let (node, _) = self.mesh.publisher(index); + let mut seen = BTreeSet::from([node]); + let mut queue = vec![node]; + while let Some(at) = queue.pop() { + for &(a, b) in up { + let peer = match at { + _ if at == a => b, + _ if at == b => a, + _ => continue, + }; + // A publisher forwards nothing, and neither does a relay that died, + // whether or not its neighbours have noticed. + if peer < self.mesh.relays && !self.down.contains(&peer) && seen.insert(peer) { + queue.push(peer); + } + } + } + seen.remove(&node); + seen + } + + fn walk(&self, from: Node, index: usize) -> Option { + let (origin, id) = self.mesh.publisher(index); + let nodes = self.mesh.relays + self.mesh.homes.len(); + self.net.next(from, id)?; + let mut at = from; + for _ in 0..=nodes { + match self.net.next(at, id) { + None => return Some(Walk::Broken), + Some(None) if at == origin => return Some(Walk::Reached), + Some(None) => return Some(Walk::Broken), + Some(Some(next)) if !self.is_up(at, next) || self.down.contains(&next) => return Some(Walk::Broken), + Some(Some(next)) => at = next, + } + } + Some(Walk::Loop) + } + + /// Whether any relay's next hops lead round a loop. + fn looping(&self) -> bool { + (0..self.mesh.homes.len()) + .any(|index| (0..self.mesh.relays).any(|relay| self.walk(relay, index) == Some(Walk::Loop))) + } + + /// Relays whose next hops loop, whose route does not lead to its publisher, and + /// with no route where a path exists. + pub fn wrong(&self) -> (usize, usize, usize) { + let (mut loops, mut stale, mut dark) = (0, 0, 0); + for index in 0..self.mesh.homes.len() { + let reach = self.reach(index); + for relay in (0..self.mesh.relays).filter(|relay| !self.down.contains(relay)) { + match self.walk(relay, index) { + Some(Walk::Reached) => {} + Some(Walk::Loop) => { + loops += 1; + stale += 1; + } + Some(Walk::Broken) => stale += 1, + None => dark += usize::from(reach.contains(&relay)), + } + } + } + (loops, stale, dark) + } + + /// Deliver everything queued now: one hop of delay. False if nothing was and no + /// timer runs. + pub fn round(&mut self) -> bool { + let mut ready = self.net.ready(); + if ready.is_empty() && !self.net.tick() && self.net.ready().is_empty() { + return false; + } + let idle = ready.is_empty(); + // Queues drain in a seeded random order, each in its own. + for i in (1..ready.len()).rev() { + ready.swap(i, self.order.next(i as u64 + 1) as usize); + } + for (chan, count) in ready { + for _ in 0..count { + self.net.step(chan); + self.cost.loops += usize::from(self.looping()); + } + } + if !idle { + self.net.tick(); + } + self.cost.rounds += 1; + let (_, stale, dark) = self.wrong(); + self.cost.stale += stale; + self.cost.dark += dark; + true + } + + /// Deliver in rounds until nothing is left. False if that never happens. + pub fn settle(&mut self) -> bool { + for _ in 0..2_000 { + if !self.round() { + return true; + } + } + false + } + + /// Apply `change`, settle, and report what that cost. + pub fn measure(&mut self, change: &dyn Fn(&mut Self)) -> Cost { + let before = self.net.sent(); + self.cost = Cost::default(); + change(self); + // What the change left wrong before any message moved. + let (_, stale, dark) = self.wrong(); + self.cost.stale += stale; + self.cost.dark += dark; + self.cost.settled = self.settle(); + self.cost.sent = self.net.sent() - before; + self.cost + } + + pub fn cut(&mut self, a: Node, b: Node) { + self.net.close(a, b); + self.up.remove(&pair(a, b)); + } + + pub fn mend(&mut self, a: Node, b: Node) { + self.net.open(a, b); + self.up.insert(pair(a, b)); + } + + /// The process at `relay` exits. Each neighbour notices in turn, a hop of delay apart. + pub fn crash(&mut self, relay: Node) { + let mut peers = self.peers(relay); + for &peer in &peers { + self.net.mute(relay, peer); + } + for i in (1..peers.len()).rev() { + peers.swap(i, self.order.next(i as u64 + 1) as usize); + } + self.down.insert(relay); + for peer in peers { + self.net.sever(relay, peer); + self.up.remove(&pair(relay, peer)); + self.round(); + } + self.net.reset(relay); + } + + /// `relay` starts again with no state, and its links and publishers reconnect. + pub fn revive(&mut self, relay: Node) { + self.down.remove(&relay); + let links: Vec<(Node, Node)> = (self.mesh.links.iter().map(|&(a, b, _)| (a, b))) + .chain((0..self.mesh.homes.len()).map(|index| (self.mesh.publisher(index).0, self.mesh.homes[index]))) + .filter(|&(a, b)| a == relay || b == relay) + .collect(); + for (a, b) in links { + if !self.down.contains(&a) && !self.down.contains(&b) && !self.is_up(a, b) { + self.mend(a, b); + } + } + } + + /// `relay` restarts at once: every session of its closes together and reopens. + pub fn restart(&mut self, relay: Node) { + let peers = self.peers(relay); + for &peer in &peers { + self.net.mute(relay, peer); + } + for peer in peers { + self.net.sever(relay, peer); + self.up.remove(&pair(relay, peer)); + } + self.net.reset(relay); + self.revive(relay); + } + + /// Deliver one message from a queue `pick` chooses. False if none waits. + pub fn deliver(&mut self, pick: &mut XorShift) -> bool { + let ready = self.net.ready(); + if ready.is_empty() { + return false; + } + let (chan, _) = ready[pick.next(ready.len() as u64) as usize]; + self.net.step(chan); + true + } + + /// Every relay with a path to a publisher routes to it and shows its claims as + /// available, and every other relay holds nothing of it. + pub fn check(&self) -> Result<(), String> { + for index in 0..self.mesh.homes.len() { + let (_, id) = self.mesh.publisher(index); + let reach = self.reach(index); + for relay in (0..self.mesh.relays).filter(|relay| !self.down.contains(relay)) { + let walk = self.walk(relay, index); + let shows = self.net.showing(relay, id, &Mesh::claim(0)); + let want = match reach.contains(&relay) { + true => (Some(Walk::Reached), Some(true)), + false => (None, None), + }; + if (walk, shows) != want { + return Err(format!( + "relay {relay} has {walk:?} to publisher {index} and shows {shows:?}, not {want:?}" + )); + } + } + } + Ok(()) + } + + /// The link of a publisher's cheapest-path tree that the most relays route + /// over, among those whose loss leaves every relay a path (`detour`) or cuts + /// some off. + pub fn tree_link(&self, index: usize, detour: bool) -> Option<(Node, Node)> { + let (_, id) = self.mesh.publisher(index); + let mut uses: BTreeMap<(Node, Node), usize> = BTreeMap::new(); + for relay in 0..self.mesh.relays { + let mut at = relay; + while let Some(Some(next)) = self.net.next(at, id) { + if next < self.mesh.relays { + *uses.entry(pair(at, next)).or_default() += 1; + } + at = next; + } + } + let before = self.reach(index).len(); + let mut links: Vec<((Node, Node), usize)> = uses.into_iter().collect(); + links.sort_by_key(|&(link, uses)| (std::cmp::Reverse(uses), link)); + links.into_iter().map(|(link, _)| link).find(|&(a, b)| { + let mut up = self.up.clone(); + up.remove(&pair(a, b)); + (self.reach_over(&up, index).len() == before) == detour + }) + } +} + +/// The changes the report measures, in the order a case applies them. +#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)] +pub(super) enum Change { + /// A publisher joins and announces its claims. + Arrive, + /// A link relays route over closes, and every relay still has a path. + Detour, + /// That link returns. + Shortcut, + /// A link closes and cuts some relays off. + Cut, + /// That link returns. + Mend, + /// A relay that others route through dies. Its neighbours notice one by one. + Transit, + /// It restarts. + Restart, + /// The publisher's own relay dies. + Home, + /// The publisher's session closes. + Leave, + /// The publisher withdraws one claim. + Withdraw, +} + +impl Change { + pub fn as_str(self) -> &'static str { + match self { + Self::Arrive => "publisher_arrives", + Self::Detour => "link_down_detour", + Self::Shortcut => "link_up_shortcut", + Self::Cut => "link_down_cut", + Self::Mend => "link_up_mend", + Self::Transit => "transit_relay_dies", + Self::Restart => "relay_restarts", + Self::Home => "home_relay_dies", + Self::Leave => "publisher_leaves", + Self::Withdraw => "claim_withdrawn", + } + } +} + +/// Run every change on `mesh` under `rule`, each from a settled mesh, checking the +/// mesh is right whenever it settles. +pub(super) fn run(rule: Rule, mesh: &Mesh) -> Vec<(Change, Cost)> { + let mut out = Vec::new(); + let mut lab = Lab::new(rule, mesh); + let mut apply = |lab: &mut Lab, change: Change, apply: &dyn Fn(&mut Lab)| { + let cost = lab.measure(apply); + assert!(cost.settled, "{}: {change:?} never settles on {mesh:?}", rule.as_str()); + if let Err(err) = lab.check() { + panic!("{}: after {change:?} on {mesh:?}: {err}", rule.as_str()); + } + out.push((change, cost)); + }; + // The last publisher leaves a settled mesh and comes back. + let last = mesh.homes.len() - 1; + let (node, _) = mesh.publisher(last); + let home = mesh.homes[last]; + apply(&mut lab, Change::Leave, &|lab| lab.cut(node, home)); + apply(&mut lab, Change::Arrive, &|lab| lab.mend(node, home)); + + for (detour, down, up) in [ + (true, Change::Detour, Change::Shortcut), + (false, Change::Cut, Change::Mend), + ] { + if let Some((a, b)) = lab.tree_link(0, detour) { + apply(&mut lab, down, &|lab| lab.cut(a, b)); + apply(&mut lab, up, &|lab| lab.mend(a, b)); + } + } + + // The relay the most others route through, the publisher's own aside. + let (node, id) = mesh.publisher(0); + let mut through = vec![0; mesh.relays]; + for relay in 0..mesh.relays { + let mut at = relay; + while let Some(Some(next)) = lab.net.next(at, id) { + if next < mesh.relays { + through[next] += 1; + } + at = next; + } + } + let transit = (0..mesh.relays) + .filter(|relay| !mesh.homes.contains(relay) && through[*relay] > 0) + .max_by_key(|&relay| (through[relay], std::cmp::Reverse(relay))); + if let Some(relay) = transit { + apply(&mut lab, Change::Transit, &|lab| lab.crash(relay)); + apply(&mut lab, Change::Restart, &|lab| lab.revive(relay)); + } + + apply(&mut lab, Change::Withdraw, &|lab| { + lab.net.claim(node, &Mesh::claim(1), false) + }); + lab.net.claim(node, &Mesh::claim(1), true); + assert!(lab.settle()); + + let home = mesh.homes[0]; + apply(&mut lab, Change::Home, &|lab| lab.crash(home)); + out +} + +/// What random churn showed, summed over its cases. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub(super) struct Churn { + pub steps: usize, + /// Steps after which some relay's next hops led round a loop. + pub loops: usize, + /// Relays, summed over steps, whose route did not lead to the publisher. + pub stale: usize, + /// Relays, summed over steps, with no route while a path to the publisher existed. + pub dark: usize, + pub sent: Sent, +} + +/// The script behind `sweep_no_route_outgrows_a_loop_free_path`, on either router: +/// random meshes of three to five relays where links flap, the publisher's most of +/// all, relays restart, and every message is delivered in a random order. Each rule +/// sees the same meshes and the same changes at the same steps. Once the churn +/// stops the mesh must settle and be right. +pub(super) fn churn(rule: Rule, cases: std::ops::Range) -> Churn { + let mut total = Churn::default(); + for case in cases { + let mut script = XorShift(0x0bad_c0de + case as u64 * 0x9e37_79b9); + let mut pick = XorShift(0x5eed_0002 + case as u64); + let relays = 3 + script.next(3) as usize; + let mut links = Vec::new(); + for a in 0..relays { + for b in a + 1..relays { + if b == a + 1 || script.chance(50) { + links.push((a, b, 1 + script.next(3))); + } + } + } + let mesh = Mesh { + relays, + links, + homes: vec![script.next(relays as u64) as usize], + claims: 2, + }; + let mut lab = Lab::new(rule, &mesh); + let (node, _) = mesh.publisher(0); + let before = lab.net.sent(); + for _ in 0..300 { + match script.next(40) { + 0 | 1 => { + let (a, b) = match script.chance(60) { + true => (node, mesh.homes[0]), + false => { + let (a, b, _) = mesh.links[script.next(mesh.links.len() as u64) as usize]; + (a, b) + } + }; + match lab.is_up(a, b) { + true => lab.cut(a, b), + false => lab.mend(a, b), + } + } + 2 if case % 4 == 0 && script.chance(25) => lab.restart(script.next(relays as u64) as usize), + _ => { + lab.deliver(&mut pick); + } + } + // A step is one delivery, so a hop of delay is several of them. + if total.steps % 8 == 7 { + lab.net.tick(); + } + let (loops, stale, dark) = lab.wrong(); + total.steps += 1; + total.loops += usize::from(loops > 0); + total.stale += stale; + total.dark += dark; + } + assert!(lab.settle(), "{}: case {case} never settles on {mesh:?}", rule.as_str()); + if let Err(err) = lab.check() { + let (_, id) = mesh.publisher(0); + let dump: Vec = (0..=relays) + .map(|node| format!("{node}: {}", lab.net.dump(node, id))) + .collect(); + panic!("{}: case {case} on {mesh:?}: {err}\n{}", rule.as_str(), dump.join("\n")); + } + let sent = lab.net.sent() - before; + total.sent = Sent { + msgs: total.sent.msgs + sent.msgs, + items: total.sent.items + sent.items, + claims: total.sent.claims + sent.claims, + offline: total.sent.offline + sent.offline, + unavailable: total.sent.unavailable + sent.unavailable, + starts: total.sent.starts + sent.starts, + }; + } + total +} + +/// Under churn every rule settles and is right once it does, and the distance +/// never loops, even for a step. +#[test] +fn every_rule_settles_after_churn() { + for rule in Rule::ALL { + let churn = churn(rule, 0..60); + if rule == Rule::Distance { + assert_eq!(churn.loops, 0); + } + } +} + +/// Loops, stale routes, and messages under the same churn, per rule. +#[test] +#[ignore = "a benchmark; Nightly reports it"] +fn benchmark_route_churn() { + // Nightly's report reads stdout and keeps the columns it has a unit for, so the + // message counts go there and the whole table to stderr. A claim sent is one + // message on the wire. + println!("rule,route_messages,claim_messages"); + eprintln!("rule,steps,loops,stale,dark,msgs,items,claims,offline,unavailable"); + for rule in Rule::ALL { + let churn = churn(rule, 0..400); + let (rule, steps, sent) = (rule.as_str(), churn.steps, churn.sent); + println!("{rule},{},{}", sent.msgs, sent.claims); + eprintln!( + "{rule},{steps},{},{},{},{},{},{},{},{}", + churn.loops, churn.stale, churn.dark, sent.msgs, sent.items, sent.claims, sent.offline, sent.unavailable, + ); + } +} + +/// Each rule's router carries every change on every shape, and is right whenever +/// it settles. +#[test] +fn every_rule_converges_on_every_shape() { + let mut rng = XorShift(0x6d6f_7100); + let mut meshes = vec![Mesh::ring(6), Mesh::clique(6, 1), Mesh::clique(4, 3)]; + meshes.extend((0..8).map(|_| Mesh::sparse(6, &mut rng))); + for mesh in &meshes { + for rule in Rule::ALL { + run(rule, mesh); + } + } +} + +/// What the report's table shows, pinned on six relays. +#[test] +fn the_rules_differ_as_reported() { + let cost = |rule: Rule, mesh: &Mesh, change: Change| { + let costs = run(rule, mesh); + costs.into_iter().find(|(found, _)| *found == change).unwrap().1 + }; + + // The publisher leaves a clique. A bare path hunts through every stale path; + // naming the hop that died ends them all at once, in fewer messages and hops of + // delay than the distance takes. + let clique = Mesh::clique(6, 1); + let bare = cost(Rule::Paths, &clique, Change::Leave); + let cause = cost(Rule::Causes, &clique, Change::Leave); + let distance = cost(Rule::Distance, &clique, Change::Leave); + assert!(bare.loops > 0 && bare.sent.msgs > 5 * cause.sent.msgs, "{bare:?}"); + assert_eq!(cause.loops, 0); + assert!(cause.sent.msgs < distance.sent.msgs && cause.rounds < distance.rounds); + + // A link of a ring closes, and the way round runs through relays that routed + // through the one that lost it. The distance waits for the publisher's next + // seqno. Without a hold the relay lets its claims go and they are sent again. + let ring = Mesh::ring(6); + let distance = cost(Rule::Distance, &ring, Change::Detour); + let holds = cost(Rule::Holds, &ring, Change::Detour); + assert!(holds.rounds < distance.rounds && holds.dark < distance.dark); + assert_eq!((holds.sent.claims, holds.sent.offline), (0, 0)); + assert!(cost(Rule::Causes, &ring, Change::Detour).sent.claims > 0); + + // The publisher's relay dies, and each neighbour reports its own session. A relay + // told of one takes a route through another for a moment, unless the dead relay + // itself is in doubt: until it advances its epoch, or for a while by the clock, + // which costs no message. + assert!(cost(Rule::Holds, &ring, Change::Home).loops > 0); + assert_eq!(cost(Rule::Epochs, &ring, Change::Home).loops, 0); + let wait = cost(Rule::Wait32, &ring, Change::Home); + assert_eq!(wait.loops, 0); + assert!(wait.sent.msgs <= cost(Rule::Epochs, &ring, Change::Home).sent.msgs); + // The clock's price: a detour through the far end of the closed link waits it out. + let (wait, epochs) = ( + cost(Rule::Wait32, &clique, Change::Detour), + cost(Rule::Epochs, &clique, Change::Detour), + ); + assert!(wait.rounds > 32 && epochs.rounds < 8 && wait.sent.msgs < epochs.sent.msgs); + assert_eq!(cost(Rule::Distance, &ring, Change::Home).loops, 0); +} + +/// Per change, mesh size, and rule, over fixed seeds: the mean of what converging cost. +#[test] +#[ignore = "a benchmark; Nightly reports it"] +fn benchmark_route_rules() { + // As above: message counts to stdout for the report, which keeps the largest mesh + // alone, and the whole table to stderr. + println!("shape,relays,change,rule,route_messages,claim_messages,loop_messages"); + eprintln!("shape,relays,change,rule,msgs,items,claims,rounds,stale,dark,loops,offline,unavailable"); + for relays in [3, 6, 12, 24] { + let mut rng = XorShift(0x6d6f_7100 + relays as u64); + let mut shapes = vec![ + ("ring", vec![Mesh::ring(relays)]), + ("clique", vec![Mesh::clique(relays, 1)]), + ]; + shapes.push(("clique_8_publishers", vec![Mesh::clique(relays, 8)])); + shapes.push(("sparse", (0..16).map(|_| Mesh::sparse(relays, &mut rng)).collect())); + for (shape, meshes) in &shapes { + for rule in Rule::ALL { + // A bare path hunts through every stale path of a clique: 65,219 messages + // for one publisher leaving twelve relays, and it grows as the factorial. + if rule == Rule::Paths && shape.starts_with("clique") && (relays > 12 || meshes[0].homes.len() > 1) { + continue; + } + let mut all: BTreeMap> = BTreeMap::new(); + for mesh in meshes { + for (change, cost) in run(rule, mesh) { + all.entry(change).or_default().push(cost); + } + } + for (change, costs) in all { + let mean = + |field: fn(&Cost) -> usize| costs.iter().map(field).sum::() as f64 / costs.len() as f64; + let (change, rule) = (change.as_str(), rule.as_str()); + if relays == 24 { + println!( + "{shape},{relays},{change},{rule},{:.1},{:.1},{:.1}", + mean(|cost| cost.sent.msgs), + mean(|cost| cost.sent.claims), + mean(|cost| cost.loops), + ); + } + eprintln!( + "{shape},{relays},{change},{rule},{:.1},{:.1},{:.1},{:.1},{:.1},{:.1},{:.1},{:.1},{:.1}", + mean(|cost| cost.sent.msgs), + mean(|cost| cost.sent.items), + mean(|cost| cost.sent.claims), + mean(|cost| cost.rounds), + mean(|cost| cost.stale), + mean(|cost| cost.dark), + mean(|cost| cost.loops), + mean(|cost| cost.sent.offline), + mean(|cost| cost.sent.unavailable), + ); + } + } + } + } +} diff --git a/rs/moq-net/src/route/tests/copies.rs b/rs/moq-net/src/route/tests/copies.rs new file mode 100644 index 0000000000..a7652355b0 --- /dev/null +++ b/rs/moq-net/src/route/tests/copies.rs @@ -0,0 +1,240 @@ +//! The cached-copy case of cache-aware routing, on the prototype in `paths.rs`. +//! +//! A relay with a reader for a broadcast carries a copy, and offers its neighbours +//! a route to that copy alone: the path the copy took from the publisher, and no +//! cost of its own. A relay pulls the broadcast from whichever is cheaper, the +//! publisher's route or a copy whose path does not name it, and passes a copy's +//! route on for as long as it beats the publisher's. + +use super::paths::{PathSim, Rules, Source}; +use super::*; + +const HOME: Node = 0; +const NEAR: Node = 1; +const FAR: Node = 2; +const TAIL: Node = 3; +const ORIGIN: Node = 4; + +/// Two relays with an uplink each to the publisher's relay, `NEAR`'s the cheaper, +/// and a cheap link between them. `TAIL` hangs off `FAR`. +fn uplinks(rules: Rules) -> PathSim { + let mut sim = PathSim::new(5, rules); + sim.leaf(ORIGIN); + sim.join(HOME, NEAR, 10); + sim.join(HOME, FAR, 12); + sim.join(NEAR, FAR, 3); + sim.join(FAR, TAIL, 1); + sim.join(ORIGIN, HOME, 1); + sim.originate(ORIGIN, 7); + sim.announce(ORIGIN, "live"); + assert!(sim.run()); + sim +} + +fn pulls(sim: &PathSim, node: Node) -> Option<(Node, Metric, bool)> { + let source: Source = sim.source(node, 7, "live")?; + Some((source.next, source.cost, source.warm)) +} + +/// How following each relay's source for the broadcast fares: relays whose sources +/// lead round a loop, and relays whose sources lead nowhere. +fn astray(sim: &PathSim) -> (usize, usize) { + let (mut loops, mut lost) = (0, 0); + for relay in [HOME, NEAR, FAR, TAIL] { + let mut at = relay; + let mut hops = 0; + loop { + match sim.source(at, 7, "live") { + None if at == relay => break, + None => { + lost += 1; + break; + } + Some(source) if source.next == ORIGIN => break, + Some(source) if !sim.connected(at, source.next) => { + lost += 1; + break; + } + Some(source) => at = source.next, + } + hops += 1; + if hops > 5 { + loops += 1; + break; + } + } + } + (loops, lost) +} + +/// Deliver one message at a time, and sum what was astray after each. +fn settle(sim: &mut PathSim) -> (usize, usize) { + let (mut loops, mut lost) = astray(sim); + for _ in 0..10_000 { + let Some(&(from, to)) = sim.busy().first() else { + return (loops, lost); + }; + sim.deliver(from, to); + let (a, b) = astray(sim); + loops += a; + lost += b; + } + panic!("never settles"); +} + +/// `FAR` carries the broadcast and is farther from the publisher than `NEAR`. +/// `NEAR` joins that copy over the cheap link rather than pull its own over its +/// uplink. Babel's distance cannot allow this: `FAR`'s cost to the publisher is +/// not below what `NEAR` advertised, which is the check cache-aware routing's +/// eligibility rule makes (moq-dev/moq#8457). +#[test] +fn a_relay_joins_a_copy_no_closer_to_the_publisher() { + for rules in [Rules::BARE, Rules::CAUSES, Rules::EPOCHS] { + let mut sim = uplinks(rules); + assert_eq!(pulls(&sim, NEAR), Some((HOME, 11, false))); + assert_eq!(pulls(&sim, FAR), Some((HOME, 13, false))); + + let before = sim.stats; + sim.carry(FAR, 7, "live"); + assert!(sim.run()); + assert_eq!(pulls(&sim, FAR), Some((HOME, 13, false)), "{rules:?}"); + assert_eq!(pulls(&sim, NEAR), Some((FAR, 3, true)), "{rules:?}"); + assert_eq!(pulls(&sim, TAIL), Some((FAR, 1, true)), "{rules:?}"); + // One message to each neighbour the copy's path does not name, and none onward: + // every further neighbour is on that path. + assert_eq!(sim.stats.warm_msgs - before.warm_msgs, 2, "{rules:?}"); + assert_eq!(sim.stats.route_msgs - before.route_msgs, 2, "{rules:?}"); + + sim.uncarry(FAR, 7, "live"); + assert!(sim.run()); + assert_eq!(pulls(&sim, NEAR), Some((HOME, 11, false)), "{rules:?}"); + assert_eq!(pulls(&sim, TAIL), Some((FAR, 14, false)), "{rules:?}"); + assert_eq!(sim.stats.warm_msgs - before.warm_msgs, 4, "{rules:?}"); + } + + // The same mesh on today's router: the neighbour with the copy is the farther one. + let mut sim = Sim::new(5); + sim.join(HOME, NEAR, 10); + sim.join(HOME, FAR, 12); + sim.join(NEAR, FAR, 3); + sim.join(FAR, TAIL, 1); + sim.join(ORIGIN, HOME, 1); + sim.originate(ORIGIN, 7, 0); + sim.announce(ORIGIN, "live", 0, None); + assert!(sim.run()); + let (near, far) = (sim.route(NEAR, 7).unwrap(), sim.route(FAR, 7).unwrap()); + assert_eq!((near.metric, far.metric), (11, 13)); + assert!(far.metric >= near.metric, "so the distance refuses the join"); +} + +/// Both relays carry a copy, `NEAR`'s pulled from `FAR`'s. `FAR` never takes +/// `NEAR`'s in return, since that copy came through `FAR`. When `FAR`'s uplink +/// closes the two swap, each still fed from the publisher. The swap loops for as +/// long as the news takes to cross their link, which a cause shortens. +#[test] +fn two_copies_never_feed_each_other() { + for rules in [Rules::BARE, Rules::CAUSES, Rules::EPOCHS] { + let mut sim = uplinks(rules); + sim.carry(FAR, 7, "live"); + assert_eq!(settle(&mut sim), (0, 0), "{rules:?}"); + sim.carry(NEAR, 7, "live"); + assert_eq!(settle(&mut sim), (0, 0), "{rules:?}"); + assert_eq!(pulls(&sim, FAR), Some((HOME, 13, false)), "{rules:?}"); + assert_eq!(pulls(&sim, NEAR), Some((FAR, 3, true)), "{rules:?}"); + + // Until `NEAR` hears, it still pulls from `FAR`, which now pulls from it. + sim.disconnect(HOME, FAR); + let (loops, _) = settle(&mut sim); + let limit = match rules.causes { + true => 3, + false => 6, + }; + assert!( + loops <= limit, + "{rules:?}: {loops} relays astray, summed over deliveries" + ); + assert_eq!(pulls(&sim, NEAR), Some((HOME, 11, false)), "{rules:?}"); + assert_eq!(pulls(&sim, FAR), Some((NEAR, 3, true)), "{rules:?}"); + assert_eq!(pulls(&sim, TAIL), Some((FAR, 1, true)), "{rules:?}"); + assert_eq!(astray(&sim), (0, 0), "{rules:?}"); + } +} + +/// `NEAR` pulls `FAR`'s copy, whose path crosses `HOME`, and then loses its own +/// session to `HOME`, which casts doubt on `HOME`. Doubt keeps a reader from moving +/// to a copy, not on the one it pulls, so `NEAR` stays on `FAR`'s throughout. +#[test] +fn a_doubt_leaves_the_copy_in_use_alone() { + for rules in [Rules::waits(4), Rules::EPOCHS] { + let mut sim = uplinks(rules); + sim.carry(FAR, 7, "live"); + assert_eq!(settle(&mut sim), (0, 0), "{rules:?}"); + sim.carry(NEAR, 7, "live"); + assert_eq!(settle(&mut sim), (0, 0), "{rules:?}"); + assert_eq!(pulls(&sim, NEAR), Some((FAR, 3, true)), "{rules:?}"); + + sim.disconnect(HOME, NEAR); + assert_eq!(pulls(&sim, NEAR), Some((FAR, 3, true)), "{rules:?}"); + for _ in 0..64 { + match sim.busy().first() { + Some(&(from, to)) => sim.deliver(from, to), + None => sim.tick(), + }; + assert_eq!(pulls(&sim, NEAR), Some((FAR, 3, true)), "{rules:?}"); + } + assert!(sim.busy().is_empty() && !sim.tick(), "{rules:?}"); + assert_eq!(astray(&sim), (0, 0), "{rules:?}"); + } +} + +/// Readers arrive at both relays in the same instant, so each pulls over its own +/// uplink and then hears of the other's copy. By their paths alone each would move +/// onto the other's, see its own copy's path come back, move off again, and so on +/// for good. Ranking the two lets exactly one move. +#[test] +fn two_copies_started_together_settle_on_one_uplink() { + for rules in [Rules::BARE, Rules::CAUSES, Rules::EPOCHS] { + let mut sim = uplinks(rules); + sim.carry(FAR, 7, "live"); + sim.carry(NEAR, 7, "live"); + assert_eq!(settle(&mut sim).0, 0, "{rules:?}"); + let (near, far) = (pulls(&sim, NEAR).unwrap(), pulls(&sim, FAR).unwrap()); + assert!( + [ + ((HOME, 11, false), (NEAR, 3, true)), + ((FAR, 3, true), (HOME, 13, false)) + ] + .contains(&(near, far)), + "{rules:?}: {near:?} {far:?}" + ); + assert_eq!(astray(&sim), (0, 0), "{rules:?}"); + + let unranked = Rules { ranked: false, ..rules }; + let mut sim = uplinks(unranked); + sim.carry(FAR, 7, "live"); + sim.carry(NEAR, 7, "live"); + assert!(!sim.run(), "{rules:?}: the two trade copies for good"); + } +} + +/// The publisher leaves while both relays carry a copy. Every route to either copy +/// goes with it and neither relay adopts the other's. A cause ends them without a +/// loop; a bare path leaves each relay to try the other's copy first. +#[test] +fn a_copy_that_goes_takes_its_routes_along() { + for rules in [Rules::BARE, Rules::CAUSES, Rules::EPOCHS] { + let mut sim = uplinks(rules); + sim.carry(FAR, 7, "live"); + assert_eq!(settle(&mut sim), (0, 0), "{rules:?}"); + sim.carry(NEAR, 7, "live"); + assert_eq!(settle(&mut sim), (0, 0), "{rules:?}"); + + sim.disconnect(ORIGIN, HOME); + let (loops, _) = settle(&mut sim); + for relay in [HOME, NEAR, FAR, TAIL] { + assert_eq!(pulls(&sim, relay), None, "{rules:?}: relay {relay}"); + assert_eq!(sim.nodes[relay].retained(), (0, 0), "{rules:?}: relay {relay}"); + } + assert_eq!(loops > 0, !rules.causes, "{rules:?}: {loops}"); + } +} diff --git a/rs/moq-net/src/route/tests/mod.rs b/rs/moq-net/src/route/tests/mod.rs index 61dc8a894b..0bee4fa960 100644 --- a/rs/moq-net/src/route/tests/mod.rs +++ b/rs/moq-net/src/route/tests/mod.rs @@ -5,7 +5,11 @@ //! are written with a trailing `/` where the contract means a wildcard; every //! lite-07 claim already covers the paths below it, so the slash is dropped. +mod compare; +mod copies; mod handoff; +mod paths; +mod residual; mod routing; mod rtt; diff --git a/rs/moq-net/src/route/tests/paths.rs b/rs/moq-net/src/route/tests/paths.rs new file mode 100644 index 0000000000..d550234db6 --- /dev/null +++ b/rs/moq-net/src/route/tests/paths.rs @@ -0,0 +1,1439 @@ +//! A prototype router that keeps a route free of loops by its path alone. +//! +//! It exists to be measured against [`Router`], which keeps Babel's distance: the +//! comparison in `compare.rs` runs one script on both. Nothing outside the tests +//! uses it. +//! +//! A route is one per publisher and names every relay it crossed, each with the +//! session that relay learned the route on. A relay uses the cheapest route whose +//! path does not name it, at once, with no distance, seqno, request, or echo. +//! [`Rules`] switches on what the comparison showed that leaves wanting: +//! +//! - **Causes.** When a session closes, the relay that learned routes on it names +//! that hop in a [`Msg::Dead`], and every relay drops every route through the hop, +//! whichever neighbour offered it. One message per session covers every publisher. +//! - **Holds.** A relay with no route left keeps its claims, unavailable, until the +//! neighbours that learned the route from it have answered: with another way, or +//! with nothing once those behind them have answered in turn. Then it lets go. +//! - **Epochs.** A closed session casts doubt on the node at its far end, since a +//! dead relay is reported one neighbour at a time. No relay moves to a route +//! through a node in doubt until that node advances its epoch. +//! - **Copies.** A relay with a reader for a broadcast offers a route to its copy, +//! which `copies.rs` exercises. +//! +//! Left out, since the comparison does not need them: scoped grants, qualified +//! keys, interests, GOAWAY, priorities, upstream marks, and RTT. + +use std::{ + collections::{BTreeMap, BTreeSet, VecDeque}, + sync::Arc, +}; + +use super::*; + +/// One relay on a route's path, and the session it learned the route on. The +/// publisher names itself with session 0. +#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)] +pub(super) struct Hop { + pub node: Id, + pub session: u32, +} + +/// Which parts of the prototype run. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(super) struct Rules { + /// A relay that loses a session names the hop that died, and every relay drops + /// every route through it. Off, each neighbour's copy lives until that neighbour + /// withdraws it: a bare path vector. + pub causes: bool, + /// One [`Msg::Dead`] says what became of every route through its hop on that + /// session. Off, each route is ended, held, or moved by its own message. + pub batch: bool, + /// A relay offers a neighbour no route whose path names that neighbour. + pub horizon: bool, + /// A relay with no route left keeps its claims, unavailable, while something + /// may still bring the route back, and its neighbours keep theirs. + pub holds: bool, + /// A closed session casts doubt on the node at its far end: no relay moves to a + /// route through that node until the node says it is still there, by advancing + /// its epoch. + pub epochs: bool, + /// The same doubt on a timer: it lifts by itself after this many hops of delay, + /// with no epoch and so no message. 0 is off. + pub wait: u64, + /// A relay already pulling a broadcast moves onto another relay's copy only if + /// that relay ranks below it, so two relays never trade copies. + pub ranked: bool, +} + +impl Rules { + /// Paths and nothing else: what lite-06 does per broadcast, here per publisher. + pub const BARE: Self = Self { + causes: false, + batch: false, + horizon: false, + holds: false, + epochs: false, + wait: 0, + ranked: true, + }; + + /// The quest's candidate: a withdrawal names the hop that died. + pub const CAUSES: Self = Self { + causes: true, + ..Self::BARE + }; + + /// Causes, one message per cause and session, and holds. + pub const HOLDS: Self = Self { + batch: true, + holds: true, + ..Self::CAUSES + }; + + /// Holds, and a doubt that lifts by itself after `wait` hops of delay. + pub const fn waits(wait: u64) -> Self { + Self { wait, ..Self::HOLDS } + } + + /// Holds, and epochs. + pub const EPOCHS: Self = Self { + epochs: true, + ..Self::HOLDS + }; +} + +/// A message on a session. +#[derive(Clone, Debug)] +pub(super) enum Msg { + /// Starts a route, or replaces its path and cost. + Offer { id: Id, path: Arc<[Hop]>, cost: Metric }, + /// The sender has no route left but keeps the claims: it may yet recover. + Hold { id: Id }, + /// Ends a route, and the claims that rode it. + End { id: Id }, + /// The session `hop` names closed, so every route through it is dead, whoever + /// offered it. Of the sender's own routes through it, each one `moved` lists + /// continues on its new path. The rest are ended if `gone`, and held if not. + /// + /// `far` is the node at the other end of that session, and its epoch as the + /// relay that saw the session close knew it. + Dead { + hop: Hop, + far: (Id, u64), + moved: Vec<(Id, Arc<[Hop]>, Metric)>, + gone: bool, + }, + /// `node` is at `epoch`: it was still there after every session it saw close + /// before that. + Epoch { node: Id, epoch: u64 }, + /// A copy of one broadcast is `cost` away through the sender, having come from + /// its publisher along `path`: 0 where the sender carries it. + /// `carrier` is the relay that carries it. + Warm { + id: Id, + claim: Path, + path: Arc<[Hop]>, + cost: Metric, + carrier: Id, + }, + /// The sender offers no copy of that broadcast any more. + Cold { id: Id, claim: Path }, + /// Changes to a route's claims, complete as of the publisher's revision `rev`. + Claims { + id: Id, + changes: Vec<(Path, Option)>, + rev: Rev, + }, +} + +/// A publisher's claims, each with the revision its life began at. +type Claims = BTreeMap; + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum State { + Live, + /// Its path holds a hop this relay knows is dead, and the neighbour has not + /// said what became of it yet. + Dead, + /// The neighbour lost the route and holds its claims. + Held, +} + +/// A route a neighbour offered. +#[derive(Clone, Debug)] +struct Heard { + /// The last path it offered. + path: Arc<[Hop]>, + cost: Metric, + state: State, + claims: Claims, + rev: Rev, +} + +/// What this relay last sent a neighbour for a route. +#[derive(Clone, Debug)] +struct Sent { + /// The last path it was offered. + path: Arc<[Hop]>, + cost: Metric, + held: bool, + claims: Claims, + rev: Rev, +} + +struct Session { + id: u32, + /// How the neighbour names itself in a path. + hop: Id, + /// Whether the neighbour passes routes on, as a relay does and a client does not. + forwards: bool, + cost: Metric, + rx: BTreeMap, + tx: BTreeMap, + /// The dead hops this neighbour was told of, while a route sent to it names one. + told: BTreeSet, + /// The epoch this neighbour was told each node is at. + epochs: BTreeMap, + /// The copies this neighbour offers. + copies: BTreeMap, + /// The copies this neighbour was offered. + casts: BTreeMap, Metric, Id)>, +} + +/// The route a relay uses for a publisher, and the claims it shows. +#[derive(Clone, Debug)] +struct Route { + /// The neighbour it goes through, or the one this relay holds under. `None` at + /// the publisher itself, and at a relay holding on its own account. + next: Option, + /// The path this relay offers: its next hop's, then itself. `None` while held. + path: Option>, + cost: Metric, + claims: Claims, + rev: Rev, +} + +/// One broadcast: its publisher and its claim. +pub(super) type Cast = (Id, Path); + +/// A copy of a broadcast a neighbour offers. +#[derive(Clone, Debug)] +struct Copy { + /// The path the copy took from its publisher to that neighbour. + path: Arc<[Hop]>, + cost: Metric, + carrier: Id, + /// Its path holds a hop this relay knows is dead. + dead: bool, +} + +/// Where a relay would pull a broadcast from. +#[derive(Clone, Debug, PartialEq, Eq)] +pub(super) struct Source { + pub next: Node, + pub cost: Metric, + /// The path the broadcast then took to this relay. + pub path: Arc<[Hop]>, + /// Whether it joins a copy, rather than follow the publisher's route. + pub warm: bool, + /// The relay carrying that copy. + pub carrier: Id, +} + +/// What a relay sends one neighbour for a route. +enum Want { + Path(Arc<[Hop]>, Metric), + Hold, + End, + /// Whatever it was sent last: the neighbour is the next hop this relay holds + /// under, and waits for this relay's answer. + Keep, +} + +pub(super) struct Relay { + hop: Id, + rules: Rules, + /// Sends only its own routes, like a client. + pub leaf: bool, + opened: u32, + sessions: BTreeMap, + own: Option, + routes: BTreeMap, + /// Dead hops, each kept while a neighbour's route still names it. + dead: BTreeSet, + /// The newest epoch heard of each node a neighbour's route names, its own included. + epochs: BTreeMap, + /// The epoch each such node was at when a session of its was last seen to close. + doubt: BTreeMap, + /// With a timer in place of epochs: when the doubt on each node lifts. + until: BTreeMap, + now: u64, + /// The broadcasts this relay has a reader for, so it carries a copy, each with + /// the neighbour it pulls from and whether that is a copy. + readers: BTreeMap>, + shown: BTreeMap<(Id, Path), (Rev, bool)>, + out: Vec<(Node, Msg)>, + events: Vec<(Id, Path, Kind)>, +} + +fn names(path: &[Hop], node: Id) -> bool { + path.iter().any(|hop| hop.node == node) +} + +/// Whether the relay that offered `path` learned it from `node`. +fn child(path: &[Hop], node: Id) -> bool { + path.len() >= 2 && path[path.len() - 2].node == node +} + +fn state(dead: &BTreeSet, path: &[Hop]) -> State { + match path.iter().any(|hop| dead.contains(hop)) { + true => State::Dead, + false => State::Live, + } +} + +impl Relay { + pub fn new(hop: Id, rules: Rules) -> Self { + Self { + hop, + rules, + leaf: false, + opened: 0, + sessions: BTreeMap::new(), + own: None, + routes: BTreeMap::new(), + dead: BTreeSet::new(), + epochs: BTreeMap::new(), + doubt: BTreeMap::new(), + until: BTreeMap::new(), + now: 0, + readers: BTreeMap::new(), + shown: BTreeMap::new(), + out: Vec::new(), + events: Vec::new(), + } + } + + pub fn drain(&mut self) -> Vec<(Node, Msg)> { + std::mem::take(&mut self.out) + } + + pub fn events(&mut self) -> Vec<(Id, Path, Kind)> { + std::mem::take(&mut self.events) + } + + pub fn connect(&mut self, peer: Node, hop: Id, forwards: bool, cost: Metric) { + self.opened += 1; + let session = Session { + id: self.opened, + hop, + forwards, + cost, + rx: BTreeMap::new(), + tx: BTreeMap::new(), + told: BTreeSet::new(), + epochs: BTreeMap::new(), + copies: BTreeMap::new(), + casts: BTreeMap::new(), + }; + self.sessions.insert(peer, session); + let ids: Vec = self.routes.keys().copied().collect(); + for id in ids { + self.advertise(id); + } + self.warm(None); + } + + /// The session to `peer` closed: every route learned on it is dead. + pub fn disconnect(&mut self, peer: Node) { + let Some(session) = self.sessions.remove(&peer) else { + return; + }; + let hop = Hop { + node: self.hop, + session: session.id, + }; + let lost = session.rx.keys().copied().collect(); + let far = (session.hop, self.epochs.get(&session.hop).copied().unwrap_or(0)); + if self.rules.epochs && session.forwards { + // This relay is still here, whatever its neighbour tells the others. A + // neighbour that passes nothing on tells nobody. + *self.epochs.entry(self.hop).or_default() += 1; + } + self.fail(hop, far, None, Vec::new(), false, lost); + self.spread(self.hop); + } + + pub fn originate(&mut self, id: Id) { + self.own = Some(id); + let path = Arc::from([Hop { + node: self.hop, + session: 0, + }]); + let route = Route { + next: None, + path: Some(path), + cost: 0, + claims: Claims::new(), + rev: 0, + }; + self.routes.insert(id, route); + self.advertise(id); + } + + pub fn announce(&mut self, claim: Path) { + self.change(claim, true); + } + + pub fn unannounce(&mut self, claim: Path) { + self.change(claim, false); + } + + fn change(&mut self, claim: Path, live: bool) { + let id = self.own.expect("not a publisher"); + let route = self.routes.get_mut(&id).unwrap(); + route.rev += 1; + match live { + true => route.claims.insert(claim, route.rev), + false => route.claims.remove(&claim), + }; + self.show(id); + self.advertise(id); + } + + pub fn recv(&mut self, from: Node, msg: Msg) { + let Some(session) = self.sessions.get_mut(&from) else { + return; + }; + match msg { + Msg::Offer { id, path, cost } => { + let state = state(&self.dead, &path); + match session.rx.get_mut(&id) { + Some(heard) => { + heard.path = path; + heard.cost = cost; + heard.state = state; + } + None => { + let heard = Heard { + path, + cost, + state, + claims: Claims::new(), + rev: 0, + }; + session.rx.insert(id, heard); + } + } + self.update(id); + } + Msg::Hold { id } => { + if let Some(heard) = session.rx.get_mut(&id) { + heard.state = State::Held; + } + self.update(id); + } + Msg::End { id } => { + session.rx.remove(&id); + self.update(id); + } + Msg::Claims { id, changes, rev } => { + let Some(heard) = session.rx.get_mut(&id) else { + return; + }; + for (claim, since) in changes { + match since { + Some(since) => heard.claims.insert(claim, since), + None => heard.claims.remove(&claim), + }; + } + heard.rev = rev; + self.update(id); + } + Msg::Dead { hop, far, moved, gone } => self.fail(hop, far, Some(from), moved, gone, BTreeSet::new()), + Msg::Epoch { node, epoch } => { + if self.epochs.get(&node).copied().unwrap_or(0) >= epoch { + return; + } + self.epochs.insert(node, epoch); + self.spread(node); + for id in self.heard() { + self.update(id); + } + self.warm(None); + } + Msg::Warm { + id, + claim, + path, + cost, + carrier, + } => { + let dead = state(&self.dead, &path) == State::Dead; + let copy = Copy { + path, + cost, + carrier, + dead, + }; + session.copies.insert((id, claim), copy); + self.warm(Some(id)); + self.collect(); + } + Msg::Cold { id, claim } => { + session.copies.remove(&(id, claim)); + self.warm(Some(id)); + self.collect(); + } + } + } + + /// The session `hop` names closed. `from` reported it, with what became of its + /// own routes, or this relay saw it close and `lost` what it had learned there. + fn fail( + &mut self, + hop: Hop, + far: (Id, u64), + from: Option, + moved: Vec<(Id, Arc<[Hop]>, Metric)>, + gone: bool, + lost: BTreeSet, + ) { + let mut touched = lost; + let (me, rules) = (self.hop, self.rules); + for (&peer, session) in self.sessions.iter_mut() { + let reported = from == Some(peer); + let through: Vec = (session.rx.iter()) + .filter(|(_, heard)| heard.path.contains(&hop)) + .map(|(id, _)| *id) + .collect(); + for id in through { + touched.insert(id); + let heard = session.rx.get_mut(&id).unwrap(); + let moved = moved.iter().find(|(named, ..)| *named == id).filter(|_| reported); + if let Some((_, path, cost)) = moved { + heard.state = state(&self.dead, path); + heard.path = path.clone(); + heard.cost = *cost; + } else if reported && rules.batch { + if gone || !rules.holds { + session.rx.remove(&id); + } else if !child(&heard.path, me) { + heard.state = State::Held; + } + // A neighbour that holds under this relay answers once those behind it have. + } else if heard.state == State::Live { + heard.state = State::Dead; + } + } + } + for session in self.sessions.values_mut() { + for copy in session.copies.values_mut() { + copy.dead |= copy.path.contains(&hop); + } + } + self.dead.insert(hop); + if rules.wait > 0 { + let until = self.until.entry(far.0).or_default(); + *until = (self.now + rules.wait).max(*until); + touched.extend(self.heard()); + } + if rules.epochs { + let doubt = self.doubt.entry(far.0).or_default(); + *doubt = far.1.max(*doubt); + // Any route through that node may have become unusable. + touched.extend(self.heard()); + } + for &id in &touched { + self.select(id); + self.show(id); + } + if rules.causes { + self.report(hop, far, from); + } + for &id in &touched { + self.advertise(id); + } + self.warm(None); + self.collect(); + } + + /// Tell each neighbour that was sent a route through `hop` what became of those + /// routes, in one message where the rules allow. + fn report(&mut self, hop: Hop, far: (Id, u64), from: Option) { + let peers: Vec = self.sessions.keys().copied().collect(); + for peer in peers { + let through: Vec = (self.sessions[&peer].tx.iter()) + .filter(|(_, sent)| sent.path.contains(&hop)) + .map(|(id, _)| *id) + .collect(); + if through.is_empty() { + let session = self.sessions.get_mut(&peer).unwrap(); + let copied = session.casts.values().any(|(path, ..)| path.contains(&hop)); + if copied && from != Some(peer) && session.told.insert(hop) { + let (moved, gone) = (Vec::new(), true); + self.out.push((peer, Msg::Dead { hop, far, moved, gone })); + } + continue; + } + let first = self.sessions.get_mut(&peer).unwrap().told.insert(hop); + if !self.rules.batch { + // Each route's own message follows, and the reporter knows. + if first && from != Some(peer) { + let (moved, gone) = (Vec::new(), false); + self.out.push((peer, Msg::Dead { hop, far, moved, gone })); + } + continue; + } + let (mut moved, mut held, mut ended, mut rest) = (Vec::new(), Vec::new(), Vec::new(), 0); + for id in through { + let want = self.want(id, peer); + let session = self.sessions.get_mut(&peer).unwrap(); + match want { + Want::Path(path, cost) => { + let sent = session.tx.get_mut(&id).unwrap(); + sent.path = path.clone(); + sent.cost = cost; + sent.held = false; + moved.push((id, path, cost)); + } + Want::Hold => { + let sent = session.tx.get_mut(&id).unwrap(); + if !sent.held { + sent.held = true; + // The neighbour it learned the route from takes silence to mean + // this relay holds under it, so it is told outright. + let parent = child(&sent.path, session.hop); + held.push((id, parent)); + } + rest += 1; + } + Want::Keep => rest += 1, + Want::End => { + session.tx.remove(&id); + ended.push(id); + } + } + } + let gone = rest == 0; + let paths: Vec> = moved.iter().map(|(_, path, _)| path.clone()).collect(); + if first || (gone && !ended.is_empty()) { + self.out.push((peer, Msg::Dead { hop, far, moved, gone })); + if !gone { + let held = held.into_iter().filter(|(_, parent)| *parent); + self.out.extend(held.map(|(id, _)| (peer, Msg::Hold { id }))); + self.out.extend(ended.into_iter().map(|id| (peer, Msg::End { id }))); + } + } else { + self.out.extend( + moved + .into_iter() + .map(|(id, path, cost)| (peer, Msg::Offer { id, path, cost })), + ); + self.out + .extend(held.into_iter().map(|(id, _)| (peer, Msg::Hold { id }))); + self.out.extend(ended.into_iter().map(|id| (peer, Msg::End { id }))); + } + // An epoch follows the route that names its node, or it would name nothing + // the neighbour holds yet. + for path in paths { + self.tell(peer, &path); + } + } + } + + fn update(&mut self, id: Id) { + self.select(id); + self.show(id); + self.advertise(id); + self.warm(Some(id)); + self.collect(); + } + + /// A reader for `cast` arrives at this relay, which carries a copy from now on. + pub fn carry(&mut self, cast: Cast) { + let id = cast.0; + self.readers.insert(cast, None); + self.warm(Some(id)); + } + + /// The reader for `cast` leaves. + pub fn uncarry(&mut self, cast: &Cast) { + self.readers.remove(cast); + self.warm(Some(cast.0)); + } + + /// Where this relay pulls `cast` from: the cheapest of its publisher's route and + /// any copy whose path does not name this relay. A tie goes to the copy. A relay + /// that already pulls it stays put unless the cheaper way is the publisher's + /// route, the same neighbour, or a copy whose carrier ranks below this relay. + pub fn source(&self, cast: &Cast) -> Option { + let mut options = Vec::new(); + options.extend(self.routes.get(&cast.0).and_then(|route| { + Some(Source { + next: route.next?, + cost: route.cost, + path: route.path.clone()?, + warm: false, + carrier: 0, + }) + })); + let pull = self.readers.get(cast).copied().flatten(); + for (&peer, session) in &self.sessions { + let Some(copy) = session.copies.get(cast) else { + continue; + }; + // As in `select`, doubt keeps a reader from moving to a copy, not on the one + // it pulls: that copy's own neighbour says when it goes. + let doubted = pull != Some((peer, true)) && self.doubted(©.path); + if copy.dead || names(©.path, self.hop) || doubted { + continue; + } + let Some(cost) = copy.cost.checked_add(session.cost) else { + continue; + }; + let mut path = copy.path.to_vec(); + path.push(Hop { + node: self.hop, + session: session.id, + }); + options.push(Source { + next: peer, + cost, + path: path.into(), + warm: true, + carrier: copy.carrier, + }); + } + let pull = pull.filter(|pull| options.iter().any(|source| (source.next, source.warm) == *pull)); + let rank = |node: Id| fnv(cast.1.as_str(), &[cast.0, node]); + let free = |source: &Source| match pull { + Some((next, _)) if self.rules.ranked => { + !source.warm || source.next == next || rank(source.carrier) < rank(self.hop) + } + _ => true, + }; + (options.into_iter()) + .filter(free) + .min_by_key(|source| (source.cost, !source.warm, source.next)) + } + + /// What this relay offers its neighbours for `cast`: its own copy at no cost, or + /// a copy it would join, for as long as that is cheaper than the publisher's route. + fn copy(&self, cast: &Cast) -> Option<(Arc<[Hop]>, Metric, Id)> { + if self.own == Some(cast.0) || self.leaf { + return None; + } + let source = self.source(cast)?; + match self.readers.contains_key(cast) { + true => Some((source.path, 0, self.hop)), + false if source.warm => Some((source.path, source.cost, source.carrier)), + false => None, + } + } + + /// Bring what each neighbour was offered up to date, for the copies of `id` or + /// for every copy. A neighbour is offered no copy whose path names it. + fn warm(&mut self, id: Option) { + let mut casts: BTreeSet = self.readers.keys().cloned().collect(); + for session in self.sessions.values() { + casts.extend(session.copies.keys().cloned()); + casts.extend(session.casts.keys().cloned()); + } + for cast in casts { + if id.is_some_and(|id| id != cast.0) { + continue; + } + let pull = self.source(&cast).map(|source| (source.next, source.warm)); + if let Some(reader) = self.readers.get_mut(&cast) { + *reader = pull; + } + let copy = self.copy(&cast); + let peers: Vec = self.sessions.keys().copied().collect(); + for peer in peers { + let session = &self.sessions[&peer]; + let want = copy.clone().filter(|(path, ..)| !names(path, session.hop)); + if session.casts.get(&cast) == want.as_ref() { + continue; + } + let (id, claim) = cast.clone(); + match want { + Some((path, cost, carrier)) => { + let session = self.sessions.get_mut(&peer).unwrap(); + session.casts.insert(cast.clone(), (path.clone(), cost, carrier)); + let msg = Msg::Warm { + id, + claim, + path: path.clone(), + cost, + carrier, + }; + self.out.push((peer, msg)); + self.tell(peer, &path); + } + None => { + self.sessions.get_mut(&peer).unwrap().casts.remove(&cast); + self.out.push((peer, Msg::Cold { id, claim })); + } + } + } + } + } + + /// Forget each dead hop no neighbour's route names any more. + fn collect(&mut self) { + let sessions = &self.sessions; + self.dead.retain(|hop| { + (sessions.values()).any(|session| { + let routes = session.rx.values().any(|heard| heard.path.contains(hop)); + routes || session.copies.values().any(|copy| copy.path.contains(hop)) + }) + }); + let me = self.hop; + let named = |node: &Id| { + *node == me + || (sessions.values()).any(|session| { + let routes = session.rx.values().any(|heard| names(&heard.path, *node)); + routes || session.copies.values().any(|copy| names(©.path, *node)) + }) + }; + self.epochs.retain(|node, _| named(node)); + self.doubt.retain(|node, _| named(node)); + self.until.retain(|node, _| named(node)); + for session in self.sessions.values_mut() { + let tx = &session.tx; + let casts = &session.casts; + session.told.retain(|hop| { + tx.values().any(|sent| sent.path.contains(hop)) || casts.values().any(|(path, ..)| path.contains(hop)) + }); + session.epochs.retain(|node, _| { + tx.values().any(|sent| names(&sent.path, *node)) || casts.values().any(|(path, ..)| names(path, *node)) + }); + } + } + + /// Every publisher a neighbour offered a route to. + fn heard(&self) -> BTreeSet { + (self.sessions.values()) + .flat_map(|session| session.rx.keys().copied()) + .collect() + } + + /// Whether `path` crosses a node that a closed session cast doubt on, and that + /// has not advanced its epoch since. + fn doubted(&self, path: &[Hop]) -> bool { + path.iter().any(|hop| { + let behind = + (self.doubt.get(&hop.node)).is_some_and(|at| self.epochs.get(&hop.node).copied().unwrap_or(0) <= *at); + behind || self.until.contains_key(&hop.node) + }) + } + + /// One hop of delay passes. Whether a doubt is still waiting to lift. + pub fn tick(&mut self) -> bool { + self.now += 1; + let before = self.until.len(); + let now = self.now; + self.until.retain(|_, until| *until > now); + if self.until.len() < before { + for id in self.heard() { + self.update(id); + } + self.warm(None); + } + !self.until.is_empty() + } + + /// Tell `peer` the epoch of each node `path` names, where it has an older one. + fn tell(&mut self, peer: Node, path: &[Hop]) { + let session = self.sessions.get_mut(&peer).unwrap(); + for hop in path { + let epoch = self.epochs.get(&hop.node).copied().unwrap_or(0); + if epoch > session.epochs.get(&hop.node).copied().unwrap_or(0) { + session.epochs.insert(hop.node, epoch); + let node = hop.node; + self.out.push((peer, Msg::Epoch { node, epoch })); + } + } + } + + /// Pass `node`'s epoch on to each neighbour that was sent a route naming it. + fn spread(&mut self, node: Id) { + let epoch = self.epochs.get(&node).copied().unwrap_or(0); + for (&peer, session) in self.sessions.iter_mut() { + let named = session.tx.values().any(|sent| names(&sent.path, node)) + || session.casts.values().any(|(path, ..)| names(path, node)); + if named && epoch > session.epochs.get(&node).copied().unwrap_or(0) { + session.epochs.insert(node, epoch); + self.out.push((peer, Msg::Epoch { node, epoch })); + } + } + } + + /// Pick the cheapest live route whose path does not name this relay, and take + /// the newest claims any such route carries. With none left, hold or let go. + fn select(&mut self, id: Id) { + if self.own == Some(id) { + return; + } + let was = self.routes.remove(&id); + let current = was.as_ref().and_then(|route| route.next); + let using = current.filter(|_| was.as_ref().is_some_and(|route| route.path.is_some())); + // Stay on the next hop among equals. + type Rank = (Metric, bool, Node); + let mut best: Option<(Rank, Arc<[Hop]>)> = None; + let mut hope = false; + for (&peer, session) in &self.sessions { + let Some(heard) = session.rx.get(&id) else { + continue; + }; + if heard.state != State::Live || names(&heard.path, self.hop) { + continue; + } + // Doubt keeps a relay from moving to a route, not on the one it uses: that + // one's own neighbour says when it goes. + if Some(peer) != using && self.doubted(&heard.path) { + // It may come back: the node in doubt has yet to say it is there. + hope = true; + continue; + } + let Some(cost) = heard.cost.checked_add(session.cost) else { + continue; + }; + let rank = (cost, Some(peer) != current, peer); + if best.as_ref().is_none_or(|(best, _)| rank < *best) { + let mut path = heard.path.to_vec(); + path.push(Hop { + node: self.hop, + session: session.id, + }); + best = Some((rank, path.into())); + } + } + let had = was.is_some(); + let (mut claims, mut rev) = was.map_or((Claims::new(), 0), |route| (route.claims, route.rev)); + for session in self.sessions.values() { + if let Some(heard) = session.rx.get(&id) + && heard.state == State::Live + && !names(&heard.path, self.hop) + && heard.rev > rev + { + claims = heard.claims.clone(); + rev = heard.rev; + } + } + let route = match best { + Some(((cost, _, peer), path)) => Route { + next: Some(peer), + path: Some(path), + cost, + claims, + rev, + }, + None => { + // Only a relay that had the route holds it. + if !self.rules.holds || !had { + return; + } + // Its next hop holds, or has yet to say what became of a dead route. A next + // hop whose own route names this relay waits on this relay instead. + let under = current + .and_then(|next| self.sessions.get(&next)) + .and_then(|session| session.rx.get(&id)) + .is_some_and(|heard| heard.state != State::Live && !names(&heard.path, self.hop)); + if !under && !hope && !self.pending(id) { + return; + } + Route { + next: current.filter(|_| under), + path: None, + cost: 0, + claims, + rev, + } + } + }; + self.routes.insert(id, route); + } + + /// Whether a neighbour that learned `id` from this relay has yet to answer its loss. + fn pending(&self, id: Id) -> bool { + self.sessions.values().any(|session| { + (session.rx.get(&id)).is_some_and(|heard| heard.state != State::Held && child(&heard.path, self.hop)) + }) + } + + /// Whether a neighbour offers a route to `id` that only doubt keeps this relay off. + fn hopes(&self, id: Id) -> bool { + self.sessions.values().any(|session| { + (session.rx.get(&id)).is_some_and(|heard| { + heard.state == State::Live && !names(&heard.path, self.hop) && self.doubted(&heard.path) + }) + }) + } + + fn want(&self, id: Id, peer: Node) -> Want { + let Some(route) = self.routes.get(&id) else { + return Want::End; + }; + if self.leaf && self.own != Some(id) { + return Want::End; + } + match &route.path { + Some(path) if self.rules.horizon && names(path, self.sessions[&peer].hop) => Want::End, + Some(path) => Want::Path(path.clone(), route.cost), + // The next hop it holds under hears nothing until those behind this relay answered. + None if route.next == Some(peer) => match self.pending(id) || self.hopes(id) { + true => Want::Keep, + false => Want::End, + }, + None => Want::Hold, + } + } + + /// Bring what each neighbour was sent for `id` up to date. + fn advertise(&mut self, id: Id) { + let peers: Vec = self.sessions.keys().copied().collect(); + for peer in peers { + let want = self.want(id, peer); + let session = self.sessions.get_mut(&peer).unwrap(); + let (path, cost) = match want { + Want::Path(path, cost) => (path, cost), + Want::Keep => continue, + Want::Hold => { + // A hold follows a route: there is nothing to hold before one. + if let Some(sent) = session.tx.get_mut(&id) + && !sent.held + { + sent.held = true; + self.out.push((peer, Msg::Hold { id })); + } + continue; + } + Want::End => { + if session.tx.remove(&id).is_some() { + self.out.push((peer, Msg::End { id })); + } + continue; + } + }; + let sent = session.tx.get(&id); + let offer = sent.is_none_or(|sent| sent.held || sent.path != path || sent.cost != cost); + let (was, at) = sent.map_or((Claims::new(), 0), |sent| (sent.claims.clone(), sent.rev)); + if offer { + let offered = path.clone(); + self.out.push(( + peer, + Msg::Offer { + id, + path: offered, + cost, + }, + )); + self.tell(peer, &path); + } + let route = &self.routes[&id]; + let session = self.sessions.get_mut(&peer).unwrap(); + if was != route.claims || at != route.rev { + let mut changes: Vec<(Path, Option)> = (route.claims.iter()) + .filter(|(claim, since)| was.get(*claim) != Some(*since)) + .map(|(claim, since)| (claim.clone(), Some(*since))) + .collect(); + let ended = was.keys().filter(|claim| !route.claims.contains_key(*claim)); + changes.extend(ended.map(|claim| (claim.clone(), None))); + let rev = route.rev; + self.out.push((peer, Msg::Claims { id, changes, rev })); + } + let sent = Sent { + path, + cost, + held: false, + claims: route.claims.clone(), + rev: route.rev, + }; + session.tx.insert(id, sent); + } + } + + /// Tell the reader what changed for `id`. + fn show(&mut self, id: Id) { + let route = self.routes.get(&id); + let available = route.is_some_and(|route| route.path.is_some()); + let empty = Claims::new(); + let claims = route.map_or(&empty, |route| &route.claims); + let was: Vec<(Path, (Rev, bool))> = (self.shown.range((id, Path::empty())..)) + .take_while(|((named, _), _)| *named == id) + .map(|((_, claim), shown)| (claim.clone(), *shown)) + .collect(); + for (claim, (since, shown)) in was { + if claims.get(&claim) != Some(&since) { + let kind = match route.is_some() { + true => Kind::End, + false => Kind::Offline, + }; + self.events.push((id, claim.clone(), kind)); + self.shown.remove(&(id, claim)); + } else if shown != available { + let kind = match available { + true => Kind::Available, + false => Kind::Unavailable, + }; + self.events.push((id, claim.clone(), kind)); + self.shown.insert((id, claim), (since, available)); + } + } + for (claim, since) in claims { + if !self.shown.contains_key(&(id, claim.clone())) { + self.events.push((id, claim.clone(), Kind::Start)); + self.shown.insert((id, claim.clone()), (*since, available)); + } + } + } + + /// The neighbour this relay routes `id` through: `Some(None)` at the publisher. + pub fn next(&self, id: Id) -> Option> { + let route = self.routes.get(&id)?; + route.path.as_ref().map(|_| route.next) + } + + pub fn cost(&self, id: Id) -> Option { + let route = self.routes.get(&id)?; + route.path.as_ref().map(|_| route.cost) + } + + /// Whether this relay shows `claim` from `id`, and whether it is available. + pub fn shows(&self, id: Id, claim: &Path) -> Option { + self.shown.get(&(id, claim.clone())).map(|(_, available)| *available) + } + + /// Dead hops remembered, and routes and copies heard. + pub fn retained(&self) -> (usize, usize) { + let heard = (self.sessions.values()) + .map(|session| session.rx.len() + session.copies.len()) + .sum(); + (self.dead.len(), heard) + } +} + +/// Messages sent, as protocol counts. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub(super) struct PathStats { + /// Offers, holds, ends, and causes. + pub route_msgs: usize, + /// Routes those messages name: a cause counts itself and each route it moved. + pub route_items: usize, + pub claim_msgs: usize, + pub claim_items: usize, + /// Copy offers and their ends, also counted as route messages. + pub warm_msgs: usize, + pub delivered: usize, +} + +/// [`Sim`] for the prototype: one [`Relay`] per node, joined by ordered queues. +pub(super) struct PathSim { + pub nodes: Vec, + rules: Rules, + restarts: Vec, + links: Vec>>, + up: BTreeSet<(Node, Node)>, + queues: BTreeMap<(Node, Node), VecDeque>, + held: BTreeSet<(Node, Node)>, + /// Nodes whose process is gone and has yet to restart. + stopped: BTreeSet, + pub stats: PathStats, + pub events: Vec<(Node, Id, String, Kind)>, +} + +impl PathSim { + pub fn new(nodes: usize, rules: Rules) -> Self { + Self { + nodes: (0..nodes).map(|node| Relay::new(hop(node, 0), rules)).collect(), + rules, + restarts: vec![0; nodes], + links: vec![vec![None; nodes]; nodes], + up: BTreeSet::new(), + queues: BTreeMap::new(), + held: BTreeSet::new(), + stopped: BTreeSet::new(), + stats: PathStats::default(), + events: Vec::new(), + } + } + + pub fn leaf(&mut self, node: Node) { + self.nodes[node].leaf = true; + } + + /// A link costing `ab` at `a` and `ba` at `b`. + pub fn link(&mut self, a: Node, b: Node, ab: Metric, ba: Metric) { + self.links[a][b] = Some(ab); + self.links[b][a] = Some(ba); + self.connect(a, b); + } + + pub fn join(&mut self, a: Node, b: Node, cost: Metric) { + self.link(a, b, cost, cost); + } + + fn pair(a: Node, b: Node) -> (Node, Node) { + (a.min(b), a.max(b)) + } + + pub fn connect(&mut self, a: Node, b: Node) { + assert!(!self.connected(a, b), "{a}-{b} already up"); + self.up.insert(Self::pair(a, b)); + for (x, y) in [(a, b), (b, a)] { + let cost = self.links[x][y].expect("a configured link"); + let peer = hop(y, self.restarts[y]); + let forwards = !self.nodes[y].leaf; + self.nodes[x].connect(y, peer, forwards, cost); + } + self.flush(); + } + + pub fn connected(&self, a: Node, b: Node) -> bool { + self.up.contains(&Self::pair(a, b)) + } + + pub fn disconnect(&mut self, a: Node, b: Node) { + self.up.remove(&Self::pair(a, b)); + for (x, y) in [(a, b), (b, a)] { + self.queues.remove(&(x, y)); + self.held.remove(&(x, y)); + self.nodes[y].disconnect(x); + } + self.flush(); + } + + /// `peer` notices its session to `node` closed. `node`, whose process is gone, + /// does nothing. + pub fn sever(&mut self, node: Node, peer: Node) { + self.up.remove(&Self::pair(node, peer)); + for key in [(node, peer), (peer, node)] { + self.queues.remove(&key); + self.held.remove(&key); + } + self.nodes[peer].disconnect(node); + self.flush(); + } + + fn peers(&self, node: Node) -> Vec { + (0..self.nodes.len()) + .filter(|&peer| self.links[node][peer].is_some() && self.connected(node, peer)) + .collect() + } + + /// The process at `node` stops: nothing passes between it and `peer` from now on, + /// though `peer` has yet to notice, and no timer of its runs until `kill`. + pub fn mute(&mut self, node: Node, peer: Node) { + self.held.insert((node, peer)); + self.held.insert((peer, node)); + self.stopped.insert(node); + } + + /// The process at `node` exits: its sessions close and its state is lost. It says + /// nothing on the way out, so each neighbour notices and it does not. + pub fn kill(&mut self, node: Node) { + for peer in self.peers(node) { + self.sever(node, peer); + } + self.stopped.remove(&node); + self.restarts[node] += 1; + let leaf = self.nodes[node].leaf; + self.nodes[node] = Relay::new(hop(node, self.restarts[node]), self.rules); + self.nodes[node].leaf = leaf; + } + + pub fn originate(&mut self, node: Node, id: Id) { + self.nodes[node].originate(id); + self.flush(); + } + + pub fn announce(&mut self, node: Node, claim: &str) { + self.nodes[node].announce(p(claim)); + self.flush(); + } + + pub fn unannounce(&mut self, node: Node, claim: &str) { + self.nodes[node].unannounce(p(claim)); + self.flush(); + } + + /// A reader for `claim` of `id` arrives at `node`. + pub fn carry(&mut self, node: Node, id: Id, claim: &str) { + self.nodes[node].carry((id, p(claim))); + self.flush(); + } + + pub fn uncarry(&mut self, node: Node, id: Id, claim: &str) { + self.nodes[node].uncarry(&(id, p(claim))); + self.flush(); + } + + /// Where `node` pulls `claim` of `id` from. + pub fn source(&self, node: Node, id: Id, claim: &str) -> Option { + self.nodes[node].source(&(id, p(claim))) + } + + /// One hop of delay passes at every running node. Whether any still waits on a timer. + pub fn tick(&mut self) -> bool { + let mut waiting = false; + for (node, relay) in self.nodes.iter_mut().enumerate() { + if !self.stopped.contains(&node) { + waiting |= relay.tick(); + } + } + self.flush(); + waiting + } + + pub fn queued(&self, from: Node, to: Node) -> usize { + self.queues.get(&(from, to)).map_or(0, VecDeque::len) + } + + /// The queues with a message to deliver that are not held. + pub fn busy(&self) -> Vec<(Node, Node)> { + (self.queues.iter()) + .filter(|(key, queue)| !queue.is_empty() && !self.held.contains(key)) + .map(|(key, _)| *key) + .collect() + } + + pub fn deliver(&mut self, from: Node, to: Node) -> bool { + let Some(msg) = self.queues.get_mut(&(from, to)).and_then(VecDeque::pop_front) else { + return false; + }; + self.nodes[to].recv(from, msg); + self.stats.delivered += 1; + self.flush(); + true + } + + /// Deliver round-robin until every unheld queue is empty, or give up. + pub fn run(&mut self) -> bool { + for _ in 0..10_000 { + let ready = self.busy(); + if ready.is_empty() { + return true; + } + for (from, to) in ready { + self.deliver(from, to); + } + } + false + } + + fn flush(&mut self) { + for node in 0..self.nodes.len() { + for (to, msg) in self.nodes[node].drain() { + match &msg { + Msg::Offer { .. } | Msg::Hold { .. } | Msg::End { .. } | Msg::Epoch { .. } => { + self.stats.route_msgs += 1; + self.stats.route_items += 1; + } + Msg::Warm { .. } | Msg::Cold { .. } => { + self.stats.route_msgs += 1; + self.stats.route_items += 1; + self.stats.warm_msgs += 1; + } + Msg::Dead { moved, .. } => { + self.stats.route_msgs += 1; + self.stats.route_items += 1 + moved.len(); + } + Msg::Claims { changes, .. } => { + self.stats.claim_msgs += 1; + self.stats.claim_items += changes.len(); + } + } + self.queues.entry((node, to)).or_default().push_back(msg); + } + for (id, claim, kind) in self.nodes[node].events() { + self.events.push((node, id, claim.as_str().to_string(), kind)); + } + } + } + + pub fn next(&self, node: Node, id: Id) -> Option> { + self.nodes[node].next(id) + } + + pub fn shows(&self, node: Node, id: Id, claim: &str) -> Option { + self.nodes[node].shows(id, &p(claim)) + } +} + +/// A relay on a ring loses its link toward the publisher. The only other way runs +/// through the neighbour that routed through it, so that neighbour moves first. +/// With no holds the relay lets its claims go and gets them back; with holds they +/// turn unavailable and back, and no claim is sent again. +#[test] +fn a_ring_takes_the_other_way_round() { + for rules in [Rules::BARE, Rules::CAUSES, Rules::HOLDS, Rules::EPOCHS] { + let origin = 4; + let mut sim = PathSim::new(5, rules); + for (a, b) in [(0, 1), (1, 2), (2, 3), (3, 0)] { + sim.join(a, b, 1); + } + sim.join(origin, 0, 1); + sim.originate(origin, 7); + sim.announce(origin, "a"); + assert!(sim.run()); + assert_eq!(sim.next(1, 7), Some(Some(0))); + assert_eq!(sim.next(2, 7), Some(Some(1))); + + let (events, claims) = (sim.events.len(), sim.stats.claim_items); + sim.disconnect(0, 1); + assert!(sim.run()); + assert_eq!(sim.next(1, 7), Some(Some(2)), "{rules:?}"); + assert_eq!(sim.nodes[1].cost(7), Some(4)); + assert_eq!(sim.shows(1, 7, "a"), Some(true)); + let seen: Vec = (sim.events[events..].iter()) + .filter(|(node, ..)| *node == 1) + .map(|(.., kind)| *kind) + .collect(); + match rules.holds { + true => { + assert_eq!(seen, [Kind::Unavailable, Kind::Available], "{rules:?}"); + assert_eq!(sim.stats.claim_items, claims, "{rules:?}: no claim is sent again"); + } + false => assert_eq!(seen, [Kind::Offline, Kind::Start], "{rules:?}"), + } + + sim.disconnect(origin, 0); + assert!(sim.run()); + for node in 0..4 { + assert_eq!(sim.next(node, 7), None, "{rules:?}: node {node}"); + assert_eq!(sim.shows(node, 7, "a"), None, "{rules:?}: node {node}"); + assert_eq!(sim.nodes[node].retained(), (0, 0), "{rules:?}: node {node}"); + } + } +} + +impl Relay { + pub fn dump(&self, id: Id) -> String { + let mut out = format!( + "hop {} route {:?} dead {:?} epochs {:?} doubt {:?}", + self.hop, + self.routes.get(&id).map(|r| (r.next, r.path.is_some())), + self.dead, + self.epochs, + self.doubt + ); + for (peer, session) in &self.sessions { + out += &format!( + "\n {peer}: rx {:?} tx {:?} told {:?}", + session + .rx + .get(&id) + .map(|h| (h.state, h.path.iter().map(|x| (x.node, x.session)).collect::>())), + session + .tx + .get(&id) + .map(|s| (s.held, s.path.iter().map(|x| (x.node, x.session)).collect::>())), + session.told + ); + } + out + } +} diff --git a/rs/moq-net/src/route/tests/residual.rs b/rs/moq-net/src/route/tests/residual.rs new file mode 100644 index 0000000000..d0fb1516ca --- /dev/null +++ b/rs/moq-net/src/route/tests/residual.rs @@ -0,0 +1,227 @@ +//! What the reviews of the hop list found on meshes where neighbours know different +//! parts of a publisher's claims, pinned as [`Router`] behaves today. +//! +//! A `residual_` test asserts a bug. Each goes when the router keeps a route per +//! scope (`quest/m2/cluster-routing/routes/scope.md`): flip its assertion there and +//! drop the prefix. The others here are rules any replacement must keep. + +use super::*; + +/// One advance of the origin's seqno and this mesh never goes quiet again under +/// round-robin delivery. Relay 5 files the origin under a qualified key for its +/// scoped peers 4 and 7, whose claims there come and go with whichever copy is +/// feasible, so 5 starts and ends the route to 4 on each advance and 4 asks for the +/// next seqno every time. +#[test] +fn residual_a_seqno_advance_storms_between_two_grants() { + let ga = Grant { + subject: "ga", + scope: "", + }; + let gb = Grant { + subject: "gb", + scope: "", + }; + let origin = 9; + let mut sim = Sim::new(10); + sim.link(origin, 4, SimLink::cost(1), SimLink::cost(2)); + sim.join(origin, 7, 1); + sim.join(1, 4, 1); + sim.join(1, 5, 1); + sim.join(2, 3, 1); + sim.join(2, 5, 1); + sim.join(3, 4, 1); + sim.join(3, 6, 1); + sim.link(4, 5, SimLink::cost(1).upstream(), SimLink::cost(1).grant(gb)); + sim.link(5, 6, SimLink::cost(1).grant(ga), SimLink::cost(1)); + sim.link(5, 7, SimLink::cost(1).grant(gb), SimLink::cost(1)); + sim.link(6, 7, SimLink::cost(3), SimLink::cost(1)); + sim.originate(origin, 8, 0); + sim.announce(origin, "", 0, None); + sim.watch(4, "a"); + sim.watch(7, "b"); + assert!(sim.settle(3000)); + assert_eq!(sim.seqno(origin), 0, "it converges with no request"); + + sim.bump(origin); + assert!(!sim.settle(500), "the storm is fixed: drop the residual"); + assert!(sim.seqno(origin) > 20, "each lap asks the origin for another seqno"); +} + +/// The relay's closest neighbour asked for `a` only. A neighbour as far from the +/// origin as the relay itself knows everything, and is not feasible, so the relay +/// neither shows `b/x` nor has a route to ask for it on. +#[test] +fn residual_a_relay_behind_a_scoped_neighbour_is_blind() { + let (origin, narrow, wide, far, relay, tail) = (0, 1, 2, 3, 4, 5); + let mut sim = Sim::new(6); + sim.watch(narrow, "a"); + sim.join(origin, narrow, 1); + sim.join(narrow, relay, 1); + sim.join(origin, far, 1); + sim.join(far, wide, 1); + sim.join(wide, relay, 1); + sim.join(relay, tail, 1); + sim.originate(origin, 7, 0); + sim.announce(origin, "a/x", 0, None); + sim.announce(origin, "b/x", 0, None); + assert!(sim.run()); + assert_eq!(sim.shows(wide, 7, "b/x"), Some(true)); + for node in [relay, tail] { + assert_eq!(sim.shows(node, 7, "a/x"), Some(true), "node {node}"); + assert_eq!( + sim.shows(node, 7, "b/x"), + None, + "node {node} sees it: drop the residual" + ); + } + let routes: Vec = sim.nodes[relay].routes(7).into_iter().map(|(next, ..)| next).collect(); + assert_eq!(routes, [narrow], "the wide neighbour is no route either"); +} + +/// The mesh the reviews found the forgotten claim on: two origins, and six relays +/// with a link to relay 0 from each of the others. `cut` leaves out the link from +/// relay 0 to relay 3, and `narrow` has relay 2 ask for `b` alone from the start. +fn forgetful(cut: bool, narrow: bool) -> Sim { + let (origin, other) = (6, 7); + let mut sim = Sim::new(8); + if narrow { + sim.watch(2, "b"); + } + sim.join(other, 4, 1); + sim.join(other, 5, 1); + sim.join(origin, 3, 1); + sim.join(origin, 2, 1); + sim.join(0, 1, 1); + sim.link(0, 2, SimLink::cost(3), SimLink::cost(1)); + if !cut { + sim.join(0, 3, 1); + } + sim.join(0, 4, 1); + sim.link(1, 2, SimLink::cost(2), SimLink::cost(1)); + sim.join(2, 3, 1); + sim.join(2, 5, 1); + sim.link(3, 4, SimLink::cost(1), SimLink::cost(2)); + sim.originate(origin, 7, 0); + sim.announce(origin, "", 0, None); + sim.originate(other, 8, 0); + sim.announce(other, "", 0, None); + assert!(sim.run()); + sim +} + +/// What a relay shows depends on the order things happened in. Relay 0 loses its +/// link to relay 3, and relay 2 then asks for `b` alone. Relay 0 is left routing +/// through relay 2 and shows nothing of the origin, though its neighbour relay 4 +/// shows the claim, and a relay that came up in the mesh as it now stands shows it +/// too. Nothing will bring the claim back. +#[test] +fn residual_a_relay_forgets_a_claim_a_fresh_relay_would_show() { + let mut sim = forgetful(false, false); + assert_eq!(sim.shows(0, 7, ""), Some(true)); + sim.disconnect(0, 3); + assert!(sim.run()); + sim.watch(2, "b"); + assert!(sim.run()); + let fresh = forgetful(true, true); + for sim in [&sim, &fresh] { + assert_eq!(sim.shows(4, 7, ""), Some(true)); + assert_eq!(sim.route(0, 7).map(|route| route.metric), Some(4)); + } + assert_eq!(fresh.shows(0, 7, ""), Some(true)); + assert_eq!(sim.shows(0, 7, ""), None, "it shows the claim: drop the residual"); +} + +/// A neighbour that stops certifying and returns brings back what it certified. +/// The fix first tried for the forgotten claim broke the two tests below. +#[test] +fn a_certifier_that_returns_brings_back_what_it_certifies() { + let mut sim = forgetful(false, false); + sim.watch(3, "b"); + assert!(sim.run()); + assert_eq!(sim.shows(4, 7, ""), Some(true)); + + sim.disconnect(6, 3); + assert!(sim.run()); + assert_eq!(sim.shows(4, 7, ""), Some(true)); + sim.disconnect(2, 3); + assert!(sim.run()); + assert_eq!(sim.shows(4, 7, ""), Some(true)); +} + +/// A neighbour at the relay's own distance is feasible for one message during a +/// seqno advance. A relay that takes its view back then flaps its claims on every +/// advance. +#[test] +fn a_seqno_advance_leaves_a_relays_claims_alone() { + let (origin, wide, narrow, relay) = (0, 1, 2, 3); + let mut sim = Sim::new(4); + sim.watch(narrow, "a"); + sim.link(origin, wide, SimLink::cost(1), SimLink::cost(2)); + sim.join(origin, narrow, 1); + sim.join(wide, relay, 1); + sim.join(narrow, relay, 1); + sim.originate(origin, 7, 0); + sim.announce(origin, "", 0, None); + assert!(sim.run()); + assert_eq!(sim.shows(relay, 7, "a"), Some(true)); + + let before = sim.events.len(); + sim.bump(origin); + assert!(sim.run()); + assert!(sim.events[before..].iter().all(|event| event.node != relay)); + assert_eq!(sim.shows(relay, 7, "a"), Some(true)); + assert_eq!(sim.shows(relay, 7, ""), None); +} + +/// The same flap makes a relay start and end its route on a neighbour's stream on +/// every advance, and that neighbour asks for the next seqno each time. +#[test] +fn two_narrowed_interests_settle() { + let (relay, wide, a, b, tail, origin) = (0, 1, 2, 3, 4, 5); + let mut sim = Sim::new(6); + sim.watch(a, "a"); + sim.watch(b, "b"); + sim.join(origin, a, 1); + sim.link(origin, wide, SimLink::cost(1), SimLink::cost(2)); + sim.join(relay, wide, 1); + sim.join(relay, a, 1); + sim.join(relay, b, 1); + sim.join(b, tail, 1); + sim.link(tail, a, SimLink::cost(4), SimLink::cost(1)); + sim.originate(origin, 7, 0); + sim.announce(origin, "", 0, None); + assert!(sim.settle(3000), "never settles: seqno {}", sim.seqno(origin)); + assert!(sim.seqno(origin) <= 1); +} + +/// Two relays behind a relay whose only way to the origin asked for `a`. One of +/// them knew everything through a dear link of its own, which closes, and the +/// origin then withdraws `b/x`. Neither may keep it alive through the other: a rule +/// that believes any neighbour whose route did not cross the relay does. +#[test] +fn two_relays_behind_a_scoped_relay_let_a_dead_claim_go() { + let (origin, narrow, relay, left, right) = (0, 1, 2, 3, 4); + let mut sim = Sim::new(5); + sim.watch(narrow, "a"); + sim.join(origin, narrow, 1); + sim.join(narrow, relay, 1); + sim.join(relay, left, 1); + sim.join(relay, right, 1); + sim.join(left, right, 1); + sim.join(origin, right, 5); + sim.originate(origin, 7, 0); + sim.announce(origin, "a/x", 0, None); + sim.announce(origin, "b/x", 0, None); + assert!(sim.run()); + assert_eq!(sim.shows(right, 7, "b/x"), Some(true)); + + sim.disconnect(origin, right); + assert!(sim.run()); + sim.unannounce(origin, "b/x"); + assert!(sim.run()); + for node in [relay, left, right] { + assert_eq!(sim.shows(node, 7, "a/x"), Some(true), "node {node}"); + assert_eq!(sim.shows(node, 7, "b/x"), None, "node {node}"); + } +} diff --git a/sh/gh/nightly.sh b/sh/gh/nightly.sh index 32750ae15e..6a554c61a3 100755 --- a/sh/gh/nightly.sh +++ b/sh/gh/nightly.sh @@ -102,6 +102,7 @@ case "$1" in nextest just rs test -p moq-uring --release -E 'test(benchmark_socket_queues)' --run-ignored only --success-output immediate nextest just rs test -p moq-tokio --release -E 'test(benchmark_datagram_delivery)' --run-ignored only --success-output immediate nextest just rs test -p moq-net --release --features async --test integration -E 'test(=mesh_sweep::benchmark_route_mesh)' --run-ignored only --success-output immediate + nextest just rs test -p moq-net --release --lib -E 'test(benchmark_route_rules) | test(benchmark_route_churn)' --run-ignored only --success-output immediate # One build for the crates whose benchmarks need no features of their own. nextest just rs test --release --no-fail-fast -p moq-relay -p moq-c -p moq-mux -p moq-cli \ -E 'test(benchmark_stun_sources) | test(benchmark_handles) | test(benchmark_damage_pid_fanout) | test(benchmark_pes_staging) | test(benchmark_stream_import)' \