refactor(network): turn the AddressBook/CandidateSet into tower services

[oxarbitrage]

Implements the plan approved in #1976 (see this comment), as one PR with three staged commits rather than three stacked PRs. Each commit compiles and passes the zebra-network suite on its own, so the series is bisect-clean.

Please review commit by commit — each maps to one stage of the approved plan.

The commits

1. promote AddressBookUpdater to a Buffer-wrapped AddressBookService — the address book is now held by a Buffer-wrapped tower service that serves requests directly, the same idiom as PeerSet. NextReconnectPeer picks the next candidate and marks it AttemptPending in a single request, fixing the ordering TODO at candidate_set.rs:414 by construction. Writes are serialised by the address book mutex, shared with the fire-and-forget change path. Hot reads (getpeerinfo, the inbound GetAddr cache) stay on the existing shared handle. init()'s signature is unchanged, so no other crate is affected. Rate-limit timers stay exactly where they were: zero network-visible behavior change.

2. replace CandidateSet manual rate-limit timers with a rate-limit layer — pacing moves out of min_next_handshake / min_next_crawl and into two middlewares in peer_set/candidate_set/rate_limit.rs. The intervals themselves are unchanged. This is the only commit with network-visible behavior in scope, and the field-revert target if one is ever needed. Equivalence notes below.

3. dissolve CandidateSet into crawl_and_dial — the struct and its Arc<futures::lock::Mutex<..>> are gone; crawl_and_dial clones the service handles into its spawned tasks. Rate limits are shared across clones, so pacing is identical without the lock.

Rate-limit equivalence (commit 2)

• RateLimitOnYield paces NextReconnectPeer: every call is forwarded immediately, and only responses that actually yield a candidate reserve a pacing slot and sleep. Reservation is atomic, so the clones introduced in commit 3 still yield at least one interval apart. After sleeping, the shared timer is recharged from the actual wake time — exactly like the replaced timer — so scheduler latency never accumulates as a pacing deficit.
• SkipRateLimit wraps the new CrawlFanout service (the former update_fanout + PEER_GET_ADDR_TIMEOUT logic). While rate-limited, calls are skipped and return None without contacting any peers. The limit is claimed at call start (so concurrent crawls are skipped) and recharged when a crawl completes, matching the replaced timer's charge-at-completion semantics.
• An empty address book still returns None immediately, without consuming rate-limit budget.
• The five existing rate-limit tests act as the equivalence oracle, with their assertions unchanged.

Changes from review

Both are amended into commit 1, so the series stays bisect-clean.

• The service holds the address book and serves requests directly, rather than proxying to the updater task over a channel — following @arya2's review comment. A worker task survives only to drain change_channel's fire-and-forget mpsc; collapsing that too needs handshake/tests/vectors.rs converted to a mock service, so it is left as a follow-up.
• Address metrics are refreshed once per batch rather than once per change. Every successful update() used to trigger a refresh, and a refresh walks the address book several times, so one peer-supplied Addr batch could hold the mutex across roughly 925k entry visits with no yield — on an async task, that stalls a runtime thread. Extend now applies its changes and refreshes once.

Deviations from the approved plan

• The updater worker is an async task, not spawn_blocking: a permanently-running blocking task inhibits tokio's paused-clock auto-advance, stalling every time::pause() test in the workspace.
• Arc<Mutex<AddressBook>> is retained as the hot-read handle described above.
• Charge-on-yield is a custom layer in the next-peer path, not the dial path.

Testing

The full zebra-network suite passes at every commit boundary. New tests: N concurrent NextReconnectPeer requests never return the same peer, and a vector test covering every request variant.

The three listener_*zcashd_compat* / listener_bans_* failures on a stock macOS dev machine are environmental (no 127.0.0.x loopback aliases configured); they fail identically on unmodified main and pass on Linux CI.

CI note

The coverage that matters for this PR is the Zebra full sync — it exercises handshaking, peer acquisition, and connection pacing against the real network, which is exactly what these commits touch. A full-sync run against this branch is the integration signal I'd like this PR judged on, and I'll post the result here.

Note that "Integration Tests on GCP" has been red on main since 2026-07-28 for unrelated reasons on the lightwalletd jobs (#10636), and none of its jobs are required checks (the required set is lint, test-crates, unit-tests, pr-gate-result). A red overall status here is not by itself evidence of a regression in this PR.

Please land it early in a release cycle so it soaks on main.

Closes #1976.

[v12-auditor]

Note

Complete: Audit complete. V12 found two issues worth reviewing.

Open the full results here.

Finding	Severity	Details
F-186231	🟡 Medium	Peer-supplied address batches run unbounded metric scans on async runtime thread The diff moved the address-book updater worker from tokio::task::spawn_blocking to a plain tokio::spawn async task, while the worker still takes a blocking std::sync::Mutex around the AddressBook. The in-code justification claims each request "only briefly locks the address book mutex ... so the task never blocks for long," but the ExtendGossiped arm holds that mutex across an entire remote-peer-supplied batch. AddressBook::extend calls AddressBook::update once per element, and every successful update unconditionally calls update_metrics, which recomputes address metrics by scanning the complete address book five times (four state_peers().count() traversals plus recently_live_peers()). With the production address-book limit of 5,000 entries and the 37-entry per-response cap enforced by PEER_ADDR_RESPONSE_LIMIT, one maximum accepted batch performs up to roughly 925,000 entry visits with no .await and therefore no scheduler yield. The same blocking-mutex-on-async-task pattern also inverts the contention direction against get_peer_info, which locks the address book directly on the async RPC path, whereas zebrad's health component deliberately wraps the identical call in spawn_blocking.
F-186233	🟡 Medium	Peer event bursts delay candidate selection through shared queue The diff routes fire-and-forget per-connection peer changes and control-plane service requests through the same bounded actor mpsc. AddressBookChangeSender::send writes directly into that channel, while NextReconnectPeer, ReadyPeerCount, CacheablePeers, and ExtendGossiped traverse the Buffer and then await the same sender. Connected peers generate direct Change events on handshake completion, receive and heartbeat errors, ping/response heartbeats, shutdown, and batched misbehavior. Because the single worker serves strictly FIFO and the channel capacity is max(peerset_total_connection_limit(), MIN_CHANNEL_SIZE), a set of peers occupying the bounded connection slots can place a burst of per-connection changes ahead of a control request and occupy the channel. This matters because the crawler holds its futures::lock::Mutex<CandidateSet> across the entire next() service round trip and the subsequent pacing sleep. The misbehavior batching task added specifically to stop peers holding the address-book lock does not cover this path: it only aggregates misbehavior events and still forwards each aggregate through the same shared sender.

And two more auto-invalidated findings.

Analyzed five files, diff 3dd9756...b101146.

[arya2]

the tower service should hold address book directly:

struct AddressBookService { address_book: AddressBook }

perhaps we should also make sure AddressBook is cheap to clone and use a watch channel:

struct AddressBookService { address_book: AddressBook, address_book_rx: WatchReceiver<AddressBook>, address_book_tx }

I was also just working on converting the address book to a service as part of the v2 p2p impl.

[oxarbitrage]

Done in 0173b534a — the service now holds the address book and serves requests directly; the forwarding client is gone. Named AddressBookHandler since the alias owns AddressBookService.

Left out for now:

• The updater task still drains change_channel's fire-and-forget mpsc. Collapsing it needs handshake/tests/vectors.rs converted to a mock service, so I kept it out of this PR.
• The watch channel — AddressBook isn't cheap to clone yet, so that wants its own change.

[oxarbitrage]

Both addressed.

F-186231 — correct, and the arithmetic checked out. Extend::extend called update() per element, and every successful update refreshed the address metrics, which walks the address book five times. Fixed in 0173b534a: metrics are refreshed once per batch, so a peer-supplied Addr batch costs 5 walks rather than 37 × 5.

F-186233 — resolved by the full series rather than disputed. The finding depends on the crawler holding futures::lock::Mutex across the next() round trip and the subsequent pacing sleep. That construct is byte-identical in base main at initialize.rs:1077, so commit 1 inherits it rather than introducing it, and commit 3 deletes it along with CandidateSet — at HEAD the only futures::lock::Mutex left in the crate is the handshake nonce set.

It also doesn't bite quantitatively: there are no per-message change events (the collector only fires on the error arm), so at 200 connections the steady rate is around 15 events/sec, and a maximal 200-event burst drains in 5-20 ms against a 100 ms MIN_OUTBOUND_PEER_CONNECTION_INTERVAL the delayed request has to wait for regardless.