From 7ce422417b139c2d622eccc6066b494c2e18bb9c Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Mon, 26 Oct 2026 09:47:30 -0400 Subject: [PATCH] fix(mesh): expire dead peer records and instrument mesh boot (#8503) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary Relay pods can stall in the `mesh_boot` startup step, and the mesh peer table grows without limit. One prod pod held 7,793 peer records. Most of those records belong to pods that no longer exist. **Cause.** Every boot gets a fresh runtime id, so a restarted pod leaves its old record behind. Nothing ever removed records, and scuttlebutt gossip kept spreading the dead ones around the fleet. Each reconcile pass, including the first one inside `boot_mesh`, then dials every record one at a time. This PR makes dead records expire. It also adds the measurements we need to see the next stall directly. ### Peer-table lifetime (`crates/buzz-relay-mesh/src/membership.rs`) - A peer record expires 5 minutes after its last heartbeat (`DEFAULT_PEER_RECORD_TTL`). For comparison, live runtimes refresh every 2s, and the Redis ready registry expires a silent runtime after 45s. - What happens to expired records: - They are **rejected on arrival**, so peers that haven't been upgraded can't re-add them. - They are **left out** of digests, deltas, and dial candidates. - They are **pruned** at the start of each reconcile pass. - Between prunes, they are not admitted (`has_peer`) and not routed (`peers`, `peer_has_capability`). - An expired entry that hasn't been pruned yet counts as absent, so any fresh record replaces it whatever its version. That lets a live peer's v1 Redis hint re-admit it before the next prune. - **The TTL never prunes or hides a peer with a live connection**, even if its heartbeat is stale. Its gossip can stop arriving while the transport stays up: a dead control stream, or a clock behind by more than the TTL. The connection proves the runtime is alive. Its stale record is still kept out of digests and dials. (The existing phi filter in `peers()` still drops a connected peer once its gossip stalls; this PR doesn't change that.) - **`Connected` tracks the transport map.** Because the TTL exemption trusts that flag, `install_peer` and `remove_peer` write it while holding the peers lock, so a connection that closes during install always ends `Disconnected`. A dial's `Connecting`/`Disconnected` writes are skipped when a connection is already installed, so a failed outbound dial can't overwrite a live inbound connection. Each reconcile pass also re-syncs the state from the transport map. - **Gossip no longer sets connection state.** Before, any gossiped record was marked `Connected`, even if this pod had never dialed it. That inflated the `/_mesh` counts. Now a record learned from gossip or the ready registry enters as `Disconnected`. Only the real connect and disconnect paths change the state. - Before building its candidate list, reconcile now skips peers that are already connected. A pass dials serially, and a dial to a dead runtime can take about a minute, so `dial_peer` re-reads the record before every address attempt. It stops once the record has expired or started draining, or once a connection has landed. ### Instrumentation - **`mesh_boot` sub-steps** use the existing startup-step gauges: `mesh_bind`, `mesh_publish`, and `mesh_initial_reconcile`. A pod stuck in `mesh_boot` now also shows which await it is stuck on. - **New metrics** (`crates/buzz-relay-mesh/src/telemetry.rs`): | Metric | Shows | |---|---| | `buzz_mesh_dial_total{outcome}` / `buzz_mesh_dial_duration_seconds` | Cost of each dial | | `buzz_mesh_reconcile_pass_seconds` | Full pass time | | `buzz_mesh_reconcile_candidates` | Dials planned for this pass (set after the registry scan) | | `buzz_mesh_peer_records{state,draining}` | Table size | | `buzz_mesh_peer_record_max_age_seconds` | Oldest heartbeat | | `buzz_mesh_peer_records_expired_total{reason=pruned\|rejected}` | Expiry activity | - The dial and pass histograms get buckets up to 600s. The default `_seconds` buckets stop at 5s, which would hide exactly the stall these metrics are meant to show. ### CI The `buzz-relay-mesh` tests and the `buzz-relay` `mesh_boot::tests` did not run in any CI lane. This PR adds both to `just test-unit`, and adds the matching stanzas to `scripts/run-tests.sh`. ### What got simpler - The peer table now has a defined size, about one record per live pod, where before it grew forever. - One wrong state write was deleted (gossip marking records `Connected`). - Two test suites that silently never ran now run. ### Rollout notes - Rejecting expired records works one pod at a time. During a rollout, pods that haven't been upgraded keep gossiping dead records. Upgraded pods reject them, which shows up as `reason="rejected"` on the expired counter. - A pod whose clock runs more than 5 minutes behind has all of its gossip rejected. It stays reachable through its Redis ready record and any live connection. A `reason="rejected"` count that keeps climbing outside a rollout points to that skew. - Upgraded pods leave expired records out of their digests. Pods that haven't been upgraded therefore resend the whole ghost set (about 7.8k records, roughly 1 MB) on every 2s gossip tick, on each old↔new connection. So during the rollout, `reason="rejected"` will climb by about the ghost count per tick per connection. This is expected and stops when the rollout finishes. ### Not in this PR The boot dial itself is still unbounded: it runs serially, has no per-dial timeout, and dials gossip-learned records. Bounding it changes how a pod joins the mesh, so it goes in a separate PR, and these metrics will measure it. ### Related issue None found. ### Testing - New tests: - `membership::tests`: expired gossip is rejected; prune drops only expired records; prune keeps a connected peer with a stale heartbeat; expired unconnected records are neither admitted nor routed before a prune; a fresh record replaces an expired, unpruned entry whatever its version; expired records are neither dialed nor gossiped before a prune; gossip never sets connection state; table stats. - `runtime::tests::reconcile_pass_publishes_dial_and_peer_table_metrics` - `runtime::tests::reconcile_pass_prunes_expired_peer_records` (binds the prune to the reconcile pass) - `runtime::tests::dial_skips_candidate_that_expired_after_the_pass_snapshot` - `runtime::tests::finishing_dial_does_not_overwrite_a_connection_that_landed` - `runtime::tests::connection_closed_during_install_ends_disconnected_and_expires` - `runtime::tests::reconcile_pass_resyncs_connection_state_from_transport` - `mesh_boot::tests::mesh_boot_sub_steps_name_the_step_that_failed` - Blox at `e06f55a4` (tree `866add1c`): `just fmt-check`, `just clippy`, `buzz-relay-mesh` nextest 3x, and the `buzz-relay` `mesh_boot`/`startup_steps`/`metrics` tests all pass. Three of the new tests fail against a mutant that removes the fix they pin (dial ignores expiry; dial state writes ignore the transport map; no reconcile resync). --------- Signed-off-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> --- Cargo.lock | 2 + Justfile | 14 +- crates/buzz-relay-mesh/Cargo.toml | 2 + crates/buzz-relay-mesh/src/lib.rs | 4 +- crates/buzz-relay-mesh/src/membership.rs | 327 +++++++++++++++- crates/buzz-relay-mesh/src/runtime.rs | 460 +++++++++++++++++++++-- crates/buzz-relay-mesh/src/telemetry.rs | 149 ++++++++ crates/buzz-relay/src/mesh_boot.rs | 76 ++++ crates/buzz-relay/src/metrics.rs | 18 + crates/buzz-relay/src/startup_steps.rs | 16 + scripts/run-tests.sh | 7 + 11 files changed, 1018 insertions(+), 57 deletions(-) create mode 100644 crates/buzz-relay-mesh/src/telemetry.rs diff --git a/Cargo.lock b/Cargo.lock index fd257fd6c0..1cb97df5cd 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1505,6 +1505,8 @@ dependencies = [ "hex", "hmac 0.13.0", "iroh", + "metrics", + "metrics-util", "nostr 0.44.7", "postcard", "proptest", diff --git a/Justfile b/Justfile index 172b7f4ae0..4bc865726e 100644 --- a/Justfile +++ b/Justfile @@ -417,6 +417,10 @@ test-unit: # already runs it. cargo test -p buzz-auth --doc cargo nextest run -p buzz-voice --lib + # Inter-relay mesh: membership, gossip, registry, and loopback-UDP + # runtime tests. Infra-free (Redis-backed paths are unit-faked or + # never dialed). Keep the Cargo fallback in step. + cargo nextest run -p buzz-relay-mesh cargo nextest run -p buzz-cli # buzz-sdk builder/validation unit tests: pure event-builder and input # validation (e.g. the canvas writer-discipline/skew guard and the @@ -532,12 +536,14 @@ test-unit: # (JWKS source + refresh-loop composition), which `--lib` cannot reach # because they live in the binary target; the nested # `tests::postgres_tests::` stays in the PostgreSQL lane. - # The fourth clause adds the startup step timers (startup_steps) and the - # env- and log-capture tests in telemetry and config. All three modules - # are infra-free; config parses env into a struct and never connects. + # The fourth clause adds the startup step timers (startup_steps), the + # mesh boot wiring (mesh_boot; Redis and Postgres point at refused + # ports), and the env- and log-capture tests in telemetry and config. + # All four modules are infra-free; config parses env into a struct and + # never connects. # `^config::` is anchored so it does not also select nip_fi_config. cargo nextest run -p buzz-relay --lib --bin buzz-relay \ - -E 'test(/^api::admin::/) + test(/^push_runtime::/) + test(/^operator_listener::tests::/) + test(/^handlers::channel_authz::/) + test(/^handlers::moderation_authz::/) + test(/^handlers::side_effects::tests::/) + test(/^handlers::system_channels::tests::/) + test(/^nip11::tests::/) + test(/^storage_sweep::tests::/) + test(/^nip_fi_core::tests::/) + test(/^nip_fi_http::tests::/) + test(/^nip_fi_config::tests::/) + test(/^nip_fi_resolver::tests::/) + test(/^readiness::tests::/) + test(/^router::tests::/) + test(/^api::parse_query_tests::/) + test(/^api::profiles::/) + test(/^api::git::transport::off_mode_precedence_tests::/) + test(/^audio::join::tests::/) + test(/^audio::handler::tests::/) + test(/^nip_fi_gate::tests::/) + test(/^nip_fi_session::tests::/) + test(/^nip_fi_shadow::tests::/) + test(/^nip_fi_shadow_session::tests::/) + test(/^startup_steps::tests::/) + test(/^telemetry::tests::/) + test(/^config::tests::/) + test(=state::tests::neither_a_confirmed_inactive_community_nor_a_failed_lookup_admits_the_socket) + test(=state::tests::periodic_revalidation_disconnects_inside_the_fenced_callback) + test(=state::tests::bare_community_disconnect_fails_closed_when_the_fence_is_unavailable) + test(=connection::tests::send_loop_sends_policy_close_when_community_is_archived) + test(=state::tests::community_disconnect_then_nip_fi_keeps_community_deleted_reason) + test(=state::tests::disconnect_community_wins_reason_losing_nip_fi_does_not_enqueue_frame) + test(=handlers::req::tests::timed_out_historical_read_deregisters_before_closed) + test(=handlers::req::tests::superseded_timeout_leaves_replacement_intact) + test(=handlers::req::tests::search_claim_retires_live_and_yields_to_replacement) + test(=handlers::req::tests::concurrent_claims_and_stale_teardowns_keep_the_last_owner) + test(=handlers::req::tests::timeout_closed_is_emitted_before_a_replacement_can_claim) + test(=handlers::req::tests::revoke_then_replacement_keeps_replacement_whole) + test(=handlers::req::tests::claims_after_connection_cleanup_are_refused) + test(=handlers::req::tests::dropped_terminal_frame_cancels_connection) + test(=handlers::req::tests::revoke_dropped_terminal_frame_cancels_connection) + test(=handlers::req::tests::count_fallback_covers_filters_that_can_match_private_deletions) + test(=handlers::req::tests::group_state_t_and_creator_tags_reach_sql_only_for_group_state_kinds) + test(=api::bridge::postgres_tests::bridge_search_mode_extension_accepts_prefix_snake_or_camel_case) + test(=api::bridge::postgres_tests::bridge_search_sort_extension_selects_newest_only_when_named) + (kind(bin) & (test(/^tests::/) + test(/^composition_tests::/) + test(/^env_filter_tests::/)) - test(/^tests::postgres_tests::/))' + -E 'test(/^api::admin::/) + test(/^push_runtime::/) + test(/^operator_listener::tests::/) + test(/^handlers::channel_authz::/) + test(/^handlers::moderation_authz::/) + test(/^handlers::side_effects::tests::/) + test(/^handlers::system_channels::tests::/) + test(/^nip11::tests::/) + test(/^storage_sweep::tests::/) + test(/^nip_fi_core::tests::/) + test(/^nip_fi_http::tests::/) + test(/^nip_fi_config::tests::/) + test(/^nip_fi_resolver::tests::/) + test(/^readiness::tests::/) + test(/^router::tests::/) + test(/^api::parse_query_tests::/) + test(/^api::profiles::/) + test(/^api::git::transport::off_mode_precedence_tests::/) + test(/^audio::join::tests::/) + test(/^audio::handler::tests::/) + test(/^nip_fi_gate::tests::/) + test(/^nip_fi_session::tests::/) + test(/^nip_fi_shadow::tests::/) + test(/^nip_fi_shadow_session::tests::/) + test(/^startup_steps::tests::/) + test(/^mesh_boot::tests::/) + test(/^telemetry::tests::/) + test(/^config::tests::/) + test(=state::tests::neither_a_confirmed_inactive_community_nor_a_failed_lookup_admits_the_socket) + test(=state::tests::periodic_revalidation_disconnects_inside_the_fenced_callback) + test(=state::tests::bare_community_disconnect_fails_closed_when_the_fence_is_unavailable) + test(=connection::tests::send_loop_sends_policy_close_when_community_is_archived) + test(=state::tests::community_disconnect_then_nip_fi_keeps_community_deleted_reason) + test(=state::tests::disconnect_community_wins_reason_losing_nip_fi_does_not_enqueue_frame) + test(=handlers::req::tests::timed_out_historical_read_deregisters_before_closed) + test(=handlers::req::tests::superseded_timeout_leaves_replacement_intact) + test(=handlers::req::tests::search_claim_retires_live_and_yields_to_replacement) + test(=handlers::req::tests::concurrent_claims_and_stale_teardowns_keep_the_last_owner) + test(=handlers::req::tests::timeout_closed_is_emitted_before_a_replacement_can_claim) + test(=handlers::req::tests::revoke_then_replacement_keeps_replacement_whole) + test(=handlers::req::tests::claims_after_connection_cleanup_are_refused) + test(=handlers::req::tests::dropped_terminal_frame_cancels_connection) + test(=handlers::req::tests::revoke_dropped_terminal_frame_cancels_connection) + test(=handlers::req::tests::count_fallback_covers_filters_that_can_match_private_deletions) + test(=handlers::req::tests::group_state_t_and_creator_tags_reach_sql_only_for_group_state_kinds) + test(=api::bridge::postgres_tests::bridge_search_mode_extension_accepts_prefix_snake_or_camel_case) + test(=api::bridge::postgres_tests::bridge_search_sort_extension_selects_newest_only_when_named) + (kind(bin) & (test(/^tests::/) + test(/^composition_tests::/) + test(/^env_filter_tests::/)) - test(/^tests::postgres_tests::/))' # Note on audio::join::tests scope: the full suite is infra-free (no DB, # no Redis). The infra-free audio/FI regression witnesses — bootstrap # ordering barrier, CommitConfirmed arm, pending-close invisibility, diff --git a/crates/buzz-relay-mesh/Cargo.toml b/crates/buzz-relay-mesh/Cargo.toml index 15d0c5ba30..5760f13770 100644 --- a/crates/buzz-relay-mesh/Cargo.toml +++ b/crates/buzz-relay-mesh/Cargo.toml @@ -24,7 +24,9 @@ hex = { workspace = true } nostr = { workspace = true } bytes = "1" futures-util = { workspace = true } +metrics = { workspace = true } [dev-dependencies] tokio = { workspace = true, features = ["test-util"] } proptest = { workspace = true } +metrics-util = { workspace = true } diff --git a/crates/buzz-relay-mesh/src/lib.rs b/crates/buzz-relay-mesh/src/lib.rs index 930f86d43a..9b7d8de9e0 100644 --- a/crates/buzz-relay-mesh/src/lib.rs +++ b/crates/buzz-relay-mesh/src/lib.rs @@ -25,12 +25,14 @@ pub mod peer; pub mod registry; pub mod runtime; pub mod status; +pub mod telemetry; pub mod wire; // Lane modules — one owner per file (see the mesh thread for lane map): // endpoint.rs, peer.rs — Mari (transport core) // registry.rs, gossip.rs, -// membership.rs, status.rs — Max (membership + /_mesh) +// membership.rs, status.rs, +// telemetry.rs — Max (membership + /_mesh + metrics) // Session directory + tunnel routing live relay-side (Perci), consuming the // seams below; huddle fan-out lives in buzz-relay's audio module (Dawn). diff --git a/crates/buzz-relay-mesh/src/membership.rs b/crates/buzz-relay-mesh/src/membership.rs index fc71cea7a4..3b1b082e66 100644 --- a/crates/buzz-relay-mesh/src/membership.rs +++ b/crates/buzz-relay-mesh/src/membership.rs @@ -3,19 +3,45 @@ //! This module implements the relay-facing [`RelayMeshMembership`] seam. It is //! deliberately incapable of electing session owners: peers here are dial/routing //! hints only, and liveness disagreement never performs takeover. +//! +//! **Record lifetime.** Every boot gets a fresh runtime id, so a restarted pod +//! leaves its old record behind. A live runtime refreshes `heartbeat_millis` +//! every gossip tick; a record whose heartbeat is older than the peer-record +//! TTL belongs to a dead process. Such records are rejected on arrival, left +//! out of digests, deltas, and dial candidates, and pruned from the table, so +//! scuttlebutt can no longer keep a dead runtime alive fleet-wide. The TTL +//! never prunes or hides a peer we hold a live connection to: the transport +//! itself proves the runtime is alive. (Phi suspicion still filters it from +//! [`RelayMeshMembership::peers`] once its gossip stalls.) The runtime keeps +//! `Connected` in step with its transport map, writing it under the map's lock +//! and re-syncing it every reconcile pass, so a closed connection cannot leave +//! a dead runtime exempt from expiry. use std::collections::HashMap; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::sync::{Arc, RwLock}; -use std::time::SystemTime; +use std::time::{Duration, SystemTime}; use crate::gossip::{system_time_from_millis, GossipRecord, PhiAccrual}; use crate::registry::ReadyRecord; use crate::status::{ConnectionState, MeshCounters, MeshPeerCounters, MeshPeerStatus, MeshStatus}; +use crate::telemetry::PeerTableStats; use crate::{PeerInfo, RelayMeshMembership, RuntimeId}; pub const DEFAULT_PHI_SUSPECT_THRESHOLD: f64 = 8.0; +/// How long a peer record survives without a fresher heartbeat. +/// +/// Live runtimes refresh every [`crate::runtime::DEFAULT_GOSSIP_INTERVAL`] +/// (2s), and the ready registry expires a silent runtime after 45s. Five +/// minutes sits well past both, so multi-hop gossip delay and cross-pod clock +/// skew *under the TTL* never expire a live runtime. A pod whose clock is +/// behind by more than the TTL has all of its gossip rejected fleet-wide; it +/// stays reachable through its Redis ready record and any live connection. +/// `buzz_mesh_peer_records_expired_total{reason="rejected"}` that keeps +/// climbing outside a rollout is the signal for that skew. +pub const DEFAULT_PEER_RECORD_TTL: Duration = Duration::from_secs(5 * 60); + #[derive(Clone, Debug)] struct PeerState { record: GossipRecord, @@ -40,6 +66,8 @@ pub struct MeshMembership { /// unanchored state is fail-closed, not accept-any. expected_relay_pubkey: Option, phi_suspect_threshold: f64, + /// Records whose heartbeat is older than this are dead; see the module docs. + peer_record_ttl: Duration, } impl MeshMembership { @@ -53,6 +81,7 @@ impl MeshMembership { foreign_relay_rejections: Arc::new(AtomicU64::new(0)), expected_relay_pubkey: None, phi_suspect_threshold: DEFAULT_PHI_SUSPECT_THRESHOLD, + peer_record_ttl: DEFAULT_PEER_RECORD_TTL, } } @@ -68,6 +97,28 @@ impl MeshMembership { self } + /// Override [`DEFAULT_PEER_RECORD_TTL`]. + pub fn with_peer_record_ttl(mut self, ttl: Duration) -> Self { + self.peer_record_ttl = ttl; + self + } + + /// Whether `record`'s heartbeat is older than the peer-record TTL at + /// `now_millis`. A heartbeat ahead of our clock (skew) is fresh. + fn is_expired(&self, record: &GossipRecord, now_millis: u64) -> bool { + let age = Duration::from_millis(now_millis.saturating_sub(record.heartbeat_millis)); + age > self.peer_record_ttl + } + + /// Whether `peer` still belongs in the table: its heartbeat is fresh, or + /// we hold a live connection to it. Governs prune, admission, and + /// routing. Digests, deltas, and dials use [`Self::is_expired`] alone so a + /// stale heartbeat is never re-gossiped. + fn is_live(&self, peer: &PeerState, now_millis: u64) -> bool { + peer.connection_state == ConnectionState::Connected + || !self.is_expired(&peer.record, now_millis) + } + pub fn local_record(&self) -> GossipRecord { self.local_record .read() @@ -116,19 +167,41 @@ impl MeshMembership { } } - /// Apply a gossiped record if it is newer than the local copy. + /// Apply a gossiped record if it is newer than the local copy and its + /// heartbeat has not expired. + /// + /// A record learned here is a dial hint, not a connection: a new peer + /// enters `Disconnected`, and an update never changes connection state. + /// Only the runtime's real connect/disconnect paths do that. + /// + /// An existing entry that is no longer live (expired, not yet pruned, and + /// not connected) counts as absent: any fresh record replaces it whatever + /// its version, so a live peer's v1 Redis hint can re-admit it before the + /// next prune. pub fn apply_gossip_record(&self, record: GossipRecord) -> bool { if record.runtime_id == self.local_runtime_id { return false; } + let now_millis = crate::gossip::now_millis(); + if self.is_expired(&record, now_millis) { + crate::telemetry::record_expired("rejected", 1); + return false; + } let heartbeat = system_time_from_millis(record.heartbeat_millis); let mut peers = self.peers.write().expect("membership lock poisoned"); match peers.get_mut(&record.runtime_id) { - Some(peer) if record.version <= peer.record.version => false, + Some(peer) + if self.is_live(peer, now_millis) && record.version <= peer.record.version => + { + false + } Some(peer) => { + if !self.is_live(peer, now_millis) { + // The old heartbeat history belongs to the dead stretch. + peer.phi = PhiAccrual::default(); + } peer.record = record; - peer.connection_state = ConnectionState::Connected; peer.phi.observe(heartbeat); true } @@ -144,7 +217,7 @@ impl MeshMembership { }, record, phi, - connection_state: ConnectionState::Connected, + connection_state: ConnectionState::Disconnected, }, ); true @@ -152,6 +225,36 @@ impl MeshMembership { } } + /// The current record for `runtime_id` if it is still worth dialing: + /// present, unexpired, and not draining. Dials re-read this before every + /// attempt, because a reconcile pass's candidate snapshot goes stale while + /// earlier dials are pending. + pub fn dial_record(&self, runtime_id: RuntimeId) -> Option { + let now_millis = crate::gossip::now_millis(); + self.peers + .read() + .expect("membership lock poisoned") + .get(&runtime_id) + .filter(|peer| !peer.record.draining && !self.is_expired(&peer.record, now_millis)) + .map(|peer| peer.record.clone()) + } + + /// Re-derive `Connected` from the transport: a peer in `connected` is + /// `Connected`; a `Connected` peer missing from it is `Disconnected`. + /// `Connecting` is left alone. Heals any write the event-driven updates + /// missed (for example a connection installed after its record was pruned + /// and later re-learned from gossip). + pub(crate) fn sync_connection_states(&self, connected: impl Fn(RuntimeId) -> bool) { + let mut peers = self.peers.write().expect("membership lock poisoned"); + for (runtime_id, peer) in peers.iter_mut() { + if connected(*runtime_id) { + peer.connection_state = ConnectionState::Connected; + } else if peer.connection_state == ConnectionState::Connected { + peer.connection_state = ConnectionState::Disconnected; + } + } + } + pub fn mark_connection_state(&self, runtime_id: RuntimeId, state: ConnectionState) { if let Some(peer) = self .peers @@ -188,22 +291,62 @@ impl MeshMembership { self.peers .read() .expect("membership lock poisoned") - .contains_key(&runtime_id) + .get(&runtime_id) + .is_some_and(|peer| self.is_live(peer, crate::gossip::now_millis())) } - /// All known gossip records (local + peers), for reconcile/dial decisions. + /// All live gossip records (local + unexpired peers), for reconcile/dial + /// decisions and gossip digests/deltas. Expired records are left out even + /// before [`Self::prune_expired`] removes them, so they are never dialed + /// or re-gossiped. pub fn records(&self) -> Vec { + self.records_at(crate::gossip::now_millis()) + } + + fn records_at(&self, now_millis: u64) -> Vec { let mut records: Vec = self .peers .read() .expect("membership lock poisoned") .values() + .filter(|peer| !self.is_expired(&peer.record, now_millis)) .map(|peer| peer.record.clone()) .collect(); records.push(self.local_record()); records } + /// Drop every peer whose heartbeat has expired, except peers we are still + /// connected to. Returns how many went. + pub fn prune_expired(&self) -> usize { + self.prune_expired_at(crate::gossip::now_millis()) + } + + fn prune_expired_at(&self, now_millis: u64) -> usize { + let mut peers = self.peers.write().expect("membership lock poisoned"); + let before = peers.len(); + peers.retain(|_, peer| self.is_live(peer, now_millis)); + let pruned = before - peers.len(); + drop(peers); + crate::telemetry::record_expired("pruned", pruned as u64); + pruned + } + + /// Publish peer-table size and staleness gauges. + pub(crate) fn publish_table_metrics(&self) { + self.table_stats(SystemTime::now()).publish(); + } + + fn table_stats(&self, now: SystemTime) -> PeerTableStats { + let mut stats = PeerTableStats::default(); + for status in self.peer_statuses(now) { + let heartbeat = system_time_from_millis(status.last_heartbeat_millis); + let age = now.duration_since(heartbeat).unwrap_or_default(); + stats.record(status.connection_state, status.draining, age); + } + stats + } + /// Scuttlebutt digest over every record this runtime knows (local + peers). pub fn digest(&self) -> crate::gossip::GossipMessage { let mut entries: Vec<_> = self @@ -358,10 +501,12 @@ impl MeshMembership { impl RelayMeshMembership for MeshMembership { fn peers(&self) -> Vec { let now = SystemTime::now(); + let now_millis = crate::gossip::now_millis(); self.peers .read() .expect("membership lock poisoned") .values() + .filter(|peer| self.is_live(peer, now_millis)) .filter_map(|peer| { let phi = peer.phi.phi_at(now); if phi.is_some_and(|p| p >= self.phi_suspect_threshold) { @@ -391,21 +536,22 @@ impl RelayMeshMembership for MeshMembership { .read() .expect("membership lock poisoned") .get(&runtime_id) + .filter(|peer| self.is_live(peer, crate::gossip::now_millis())) .is_some_and(|peer| peer.record.capabilities.iter().any(|c| c == capability)) } } #[cfg(test)] mod tests { - use std::time::{Duration, UNIX_EPOCH}; - use super::*; + use crate::gossip::now_millis; fn rid(byte: u8) -> RuntimeId { RuntimeId([byte; 32]) } - fn record(byte: u8, version: u64, heartbeat_secs: u64) -> GossipRecord { + /// A peer record whose heartbeat is `age_secs` old. + fn record(byte: u8, version: u64, age_secs: u64) -> GossipRecord { GossipRecord { runtime_id: rid(byte), endpoint_addrs: vec![format!("127.0.0.{byte}:3478")], @@ -414,10 +560,7 @@ mod tests { draining: false, capabilities: vec!["reliable-stream".to_string()], version, - heartbeat_millis: (UNIX_EPOCH + Duration::from_secs(heartbeat_secs)) - .duration_since(UNIX_EPOCH) - .unwrap() - .as_millis() as u64, + heartbeat_millis: now_millis() - age_secs * 1_000, } } @@ -498,6 +641,162 @@ mod tests { assert_eq!(status.peers[0].counters.stale_generation_rejections, 1); } + const TTL_SECS: u64 = DEFAULT_PEER_RECORD_TTL.as_secs(); + + #[test] + fn expired_gossip_record_is_rejected() { + let membership = MeshMembership::new(record(1, 1, 0)); + assert!(!membership.apply_gossip_record(record(2, 1, TTL_SECS + 1))); + assert!(membership.status().peers.is_empty()); + assert!(!membership.has_peer(rid(2))); + // A fresh record for the same runtime is still admitted. + assert!(membership.apply_gossip_record(record(2, 2, 0))); + assert!(membership.has_peer(rid(2))); + } + + #[test] + fn prune_drops_only_expired_records() { + let membership = MeshMembership::new(record(1, 1, 0)); + membership.apply_gossip_record(record(2, 1, 0)); + membership.apply_gossip_record(record(3, 1, 200)); + // 150s later: rid(3) is 350s old (expired), rid(2) is 150s (live). + let later = now_millis() + 150_000; + assert_eq!(membership.prune_expired_at(later), 1); + assert!(membership.has_peer(rid(2))); + assert!(!membership.has_peer(rid(3))); + assert_eq!(membership.prune_expired_at(later), 0); + } + + #[test] + fn prune_keeps_a_connected_peer_with_a_stale_heartbeat() { + // A live connection whose gossip stopped arriving (control stream + // died, or the peer's clock is behind) must not be pruned: a Redis + // rescan would re-add it as `Disconnected` and nothing would restore + // `Connected`. Its stale record still stays out of digests and dials. + let membership = MeshMembership::new(record(1, 1, 0)); + membership.apply_gossip_record(record(2, 1, 0)); + membership.mark_connection_state(rid(2), ConnectionState::Connected); + let later = now_millis() + (TTL_SECS + 1) * 1_000; + assert_eq!(membership.prune_expired_at(later), 0); + assert!(membership.has_peer(rid(2))); + let status = membership.status(); + assert_eq!(status.peers[0].connection_state, ConnectionState::Connected); + assert!(!membership + .records_at(later) + .iter() + .any(|r| r.runtime_id == rid(2))); + // Once the connection really closes, the next prune drops it. + membership.mark_connection_state(rid(2), ConnectionState::Disconnected); + assert_eq!(membership.prune_expired_at(later), 1); + assert!(!membership.has_peer(rid(2))); + } + + #[test] + fn expired_unconnected_records_are_not_admitted_or_routed_before_prune() { + // Prune runs once per reconcile pass; a slow pass must not leave + // ghosts admissible or routable in between. + let ttl = Duration::from_millis(50); + let membership = MeshMembership::new(record(1, 1, 0)).with_peer_record_ttl(ttl); + membership.apply_gossip_record(record(2, 1, 0)); + membership.apply_gossip_record(record(3, 1, 0)); + membership.mark_connection_state(rid(3), ConnectionState::Connected); + std::thread::sleep(ttl * 3); + assert!(!membership.has_peer(rid(2))); + assert!(!membership.peer_has_capability(rid(2), "reliable-stream")); + let routed: Vec<_> = membership + .peers() + .into_iter() + .map(|p| p.runtime_id) + .collect(); + assert!(!routed.contains(&rid(2))); + // The connected peer stays admitted and routable. + assert!(membership.has_peer(rid(3))); + assert!(membership.peer_has_capability(rid(3), "reliable-stream")); + assert!(routed.contains(&rid(3))); + } + + #[test] + fn a_fresh_record_replaces_an_expired_unpruned_entry_whatever_its_version() { + // A live peer whose stale entry outlived the TTL but not yet a prune + // must be re-admitted by its v1 Redis hint, or inbound from it is + // rejected until the next pass. + let ttl = Duration::from_millis(50); + let membership = MeshMembership::new(record(1, 1, 0)).with_peer_record_ttl(ttl); + assert!(membership.apply_gossip_record(record(2, 5, 0))); + std::thread::sleep(ttl * 3); + assert!(!membership.has_peer(rid(2))); + assert!(membership.apply_gossip_record(record(2, 1, 0))); + assert!(membership.has_peer(rid(2))); + // The dead gap is not kept as a phi sample: one fresh observation + // and no intervals reads as no suspicion yet. + assert_eq!(membership.status().peers[0].phi, None); + // A live entry still keeps the version guard. + assert!(!membership.apply_gossip_record(record(2, 1, 0))); + } + + #[test] + fn expired_records_are_not_dialed_or_gossiped_before_prune() { + let membership = MeshMembership::new(record(1, 1, 0)); + membership.apply_gossip_record(record(2, 1, 0)); + membership.apply_gossip_record(record(3, 1, 200)); + let later = now_millis() + 150_000; + let ids: Vec<_> = membership + .records_at(later) + .into_iter() + .map(|r| r.runtime_id) + .collect(); + assert!(ids.contains(&rid(1)), "local record is always present"); + assert!(ids.contains(&rid(2))); + assert!( + !ids.contains(&rid(3)), + "expired record must not be a candidate" + ); + // Still in the table until a prune runs. + assert!(membership.has_peer(rid(3))); + } + + #[test] + fn gossip_never_sets_connection_state() { + let membership = MeshMembership::new(record(1, 1, 0)); + let state = |m: &MeshMembership| m.status().peers[0].connection_state; + + membership.apply_gossip_record(record(2, 1, 0)); + assert_eq!(state(&membership), ConnectionState::Disconnected); + + membership.mark_connection_state(rid(2), ConnectionState::Connected); + membership.apply_gossip_record(record(2, 2, 0)); + assert_eq!(state(&membership), ConnectionState::Connected); + + membership.mark_connection_state(rid(2), ConnectionState::Disconnected); + membership.apply_gossip_record(record(2, 3, 0)); + assert_eq!(state(&membership), ConnectionState::Disconnected); + } + + #[test] + fn table_stats_count_state_and_draining_and_track_oldest_heartbeat() { + use crate::telemetry::CONNECTION_STATES; + + let membership = MeshMembership::new(record(1, 1, 0)); + membership.apply_gossip_record(record(2, 1, 10)); + membership.apply_gossip_record(record(3, 1, 120)); + membership.mark_connection_state(rid(3), ConnectionState::Connected); + let mut draining = record(4, 1, 30); + draining.draining = true; + membership.apply_gossip_record(draining); + + let stats = membership.table_stats(SystemTime::now()); + let count = |state, draining: bool| { + let index = CONNECTION_STATES.iter().position(|s| *s == state).unwrap(); + stats.counts[index][usize::from(draining)] + }; + assert_eq!(count(ConnectionState::Disconnected, false), 1); + assert_eq!(count(ConnectionState::Disconnected, true), 1); + assert_eq!(count(ConnectionState::Connected, false), 1); + assert_eq!(stats.counts.iter().flatten().sum::(), 3); + assert!(stats.max_age >= Duration::from_secs(120)); + assert!(stats.max_age < Duration::from_secs(130)); + } + #[test] fn begin_drain_updates_local_record() { let membership = MeshMembership::new(record(1, 1, 1)); diff --git a/crates/buzz-relay-mesh/src/runtime.rs b/crates/buzz-relay-mesh/src/runtime.rs index 897de6ca8c..e56af33206 100644 --- a/crates/buzz-relay-mesh/src/runtime.rs +++ b/crates/buzz-relay-mesh/src/runtime.rs @@ -24,7 +24,7 @@ use std::collections::HashMap; use std::sync::{Arc, Mutex, RwLock}; -use std::time::Duration; +use std::time::{Duration, Instant}; use tokio::sync::mpsc; use tokio::task::JoinHandle; @@ -35,6 +35,7 @@ use crate::membership::MeshMembership; use crate::peer::MeshPeer; use crate::registry::{ReadyRecord, ReadyRegistry}; use crate::status::ConnectionState; +use crate::telemetry; use crate::wire::{MeshStreamFrame, StreamHello, StreamRole}; use crate::{InboundHandler, MeshDatagram, MeshError, MeshStream, RelayPeerTransport, RuntimeId}; @@ -255,29 +256,40 @@ fn install_peer(inner: &Arc, peer: MeshPeer, dialed_by_us: bool) -> bool tasks, }, ); - drop(peers); + // Still under the peers lock: a receive task that sees this connection + // close must wait for it, so its `Disconnected` lands after `Connected`. inner .membership .mark_connection_state(remote, ConnectionState::Connected); + drop(peers); tracing::info!(peer = %remote, dialed_by_us, "mesh: peer connected"); true } fn remove_peer(inner: &Inner, runtime_id: RuntimeId) { - if let Some(entry) = inner - .peers - .write() - .expect("peer lock poisoned") - .remove(&runtime_id) - { + let mut peers = inner.peers.write().expect("peer lock poisoned"); + if let Some(entry) = peers.remove(&runtime_id) { entry.abort(); inner .membership .mark_connection_state(runtime_id, ConnectionState::Disconnected); + drop(peers); tracing::info!(peer = %runtime_id, "mesh: peer disconnected"); } } +/// Write a dial's `state` unless a connection to `runtime_id` is installed +/// (an inbound accept can land while our dial is pending). Returns whether +/// it wrote. Holding the peers lock orders it against install/remove. +fn mark_dial_state(inner: &Inner, runtime_id: RuntimeId, state: ConnectionState) -> bool { + let peers = inner.peers.read().expect("peer lock poisoned"); + if peers.contains_key(&runtime_id) { + return false; + } + inner.membership.mark_connection_state(runtime_id, state); + true +} + async fn accept_loop(inner: Arc) { loop { match inner.endpoint.accept().await { @@ -327,66 +339,93 @@ async fn reconcile_loop(inner: Arc) { } async fn reconcile_once(inner: &Arc) { + let started = Instant::now(); + { + let peers = inner.peers.read().expect("peer lock poisoned"); + inner + .membership + .sync_connection_states(|runtime_id| peers.contains_key(&runtime_id)); + } + let pruned = inner.membership.prune_expired(); + if pruned > 0 { + tracing::info!(pruned, "mesh: pruned expired peer records"); + } + if let Some(registry) = &inner.registry { match registry.scan_ready().await { Ok(records) => inner.membership.apply_ready_records(records), Err(err) => tracing::warn!(%err, "mesh: registry scan failed"), } } + inner.membership.publish_table_metrics(); + let candidates = dial_candidates(inner); + metrics::gauge!(telemetry::RECONCILE_CANDIDATES).set(candidates.len() as f64); + for runtime_id in candidates { + dial_peer(inner, runtime_id).await; + } + metrics::histogram!(telemetry::RECONCILE_PASS_SECONDS).record(started.elapsed().as_secs_f64()); +} + +/// Unconnected, non-draining, unexpired peers, snapshotted at pass start. +fn dial_candidates(inner: &Inner) -> Vec { let local = inner.endpoint.runtime_id(); - let candidates: Vec<_> = inner + let peers = inner.peers.read().expect("peer lock poisoned"); + inner .membership .records() .into_iter() - .filter(|record| record.runtime_id != local && !record.draining) - .collect(); - - for record in candidates { - let already_connected = inner - .peers - .read() - .expect("peer lock poisoned") - .contains_key(&record.runtime_id); - if already_connected { - continue; - } - dial_peer(inner, &record).await; - } + .filter(|record| { + record.runtime_id != local + && !record.draining + && !peers.contains_key(&record.runtime_id) + }) + .map(|record| record.runtime_id) + .collect() } -async fn dial_peer(inner: &Arc, record: &crate::gossip::GossipRecord) { +/// Dial `runtime_id`'s addresses in turn. Each dial can take a minute against +/// a dead runtime, so before every attempt this re-reads the record and stops +/// once it has expired or started draining, or a connection has landed +/// (inbound accept, or a tie-break). +async fn dial_peer(inner: &Arc, runtime_id: RuntimeId) { + let Some(record) = inner.membership.dial_record(runtime_id) else { + return; + }; for addr in &record.endpoint_addrs { + if inner.membership.dial_record(runtime_id).is_none() + || !mark_dial_state(inner, runtime_id, ConnectionState::Connecting) + { + break; + } let sock = match addr.parse() { Ok(sock) => sock, Err(err) => { - tracing::warn!(peer = %record.runtime_id, addr, %err, "mesh: bad peer addr"); + tracing::warn!(peer = %runtime_id, addr, %err, "mesh: bad peer addr"); continue; } }; - let endpoint_addr = match direct_addr(record.runtime_id, sock) { + let endpoint_addr = match direct_addr(runtime_id, sock) { Ok(ea) => ea, Err(err) => { - tracing::warn!(peer = %record.runtime_id, %err, "mesh: bad peer id"); - return; + tracing::warn!(peer = %runtime_id, %err, "mesh: bad peer id"); + break; } }; - inner - .membership - .mark_connection_state(record.runtime_id, ConnectionState::Connecting); - match inner.endpoint.connect(endpoint_addr).await { + let started = Instant::now(); + let result = inner.endpoint.connect(endpoint_addr).await; + telemetry::record_dial(result.is_ok(), started.elapsed()); + match result { Ok(peer) => { install_peer(inner, peer, true); return; } Err(err) => { - tracing::warn!(peer = %record.runtime_id, addr, %err, "mesh: dial failed"); + tracing::warn!(peer = %runtime_id, addr, %err, "mesh: dial failed"); } } } - inner - .membership - .mark_connection_state(record.runtime_id, ConnectionState::Disconnected); + mark_dial_state(inner, runtime_id, ConnectionState::Disconnected); } async fn datagram_recv_loop(inner: Arc, peer: MeshPeer) { @@ -626,7 +665,20 @@ mod tests { SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0) } + const RECONCILE_INTERVAL: Duration = Duration::from_millis(200); + /// Long enough that the background reconcile loop never runs a second + /// pass during a test, so the test drives every dial and state sync. + const QUIET: Duration = Duration::from_secs(3600); + async fn runtime(key_byte: u8) -> (MeshRuntime, Vec) { + runtime_with(key_byte, |membership| membership, RECONCILE_INTERVAL).await + } + + async fn runtime_with( + key_byte: u8, + configure: impl FnOnce(MeshMembership) -> MeshMembership, + reconcile_interval: Duration, + ) -> (MeshRuntime, Vec) { let endpoint = MeshEndpoint::bind_with_secret_key( SecretKey::from_bytes(&[key_byte; 32]), loopback_any(), @@ -644,17 +696,69 @@ mod tests { .collect(); assert!(!addrs.is_empty(), "endpoint must expose a loopback addr"); let record = GossipRecord::new(endpoint.runtime_id(), addrs.clone(), 1); - let membership = MeshMembership::new(record); + let membership = configure(MeshMembership::new(record)); let rt = MeshRuntime::start_with_intervals( endpoint, membership, None, Duration::from_millis(100), - Duration::from_millis(200), + reconcile_interval, ); (rt, addrs) } + /// A bare endpoint (no runtime) that accepts every dial. `drop_accepted` + /// closes each connection as soon as it lands; otherwise they are held. + async fn accepting_endpoint( + key_byte: u8, + drop_accepted: bool, + ) -> (MeshEndpoint, Vec, JoinHandle<()>) { + let endpoint = MeshEndpoint::bind_with_secret_key( + SecretKey::from_bytes(&[key_byte; 32]), + loopback_any(), + ) + .await + .unwrap(); + let addrs = endpoint.ip_addrs().iter().map(|s| s.to_string()).collect(); + let accept = tokio::spawn({ + let endpoint = endpoint.clone(); + async move { + let mut held = Vec::new(); + while let Ok(Some(peer)) = endpoint.accept().await { + if !drop_accepted { + held.push(peer); + } + } + } + }); + (endpoint, addrs, accept) + } + + fn state_of(rt: &MeshRuntime, runtime_id: RuntimeId) -> Option { + rt.membership() + .status() + .peers + .into_iter() + .find(|peer| peer.runtime_id == runtime_id.to_string()) + .map(|peer| peer.connection_state) + } + + fn dial_count(snapshotter: &metrics_util::debugging::Snapshotter) -> u64 { + snapshotter + .snapshot() + .into_vec() + .into_iter() + .filter_map(|(key, _, _, value)| match value { + metrics_util::debugging::DebugValue::Counter(n) + if key.key().name() == telemetry::DIAL_TOTAL => + { + Some(n) + } + _ => None, + }) + .sum() + } + /// Seed b's record into a's membership so a dials b. fn seed(a: &MeshRuntime, b: &MeshRuntime, b_addrs: &[String]) { a.membership().apply_gossip_record(GossipRecord::new( @@ -882,6 +986,286 @@ mod tests { b.shutdown(); } + #[tokio::test] + async fn reconcile_pass_publishes_dial_and_peer_table_metrics() { + use metrics_util::debugging::{DebugValue, DebuggingRecorder}; + + // Current-thread runtime: the local recorder also sees the spawned + // reconcile loop, which is why the dial assertions are lower bounds. + let recorder = DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + let _guard = metrics::set_default_local_recorder(&recorder); + + let (a, _) = runtime(5).await; + // A bare endpoint (no runtime) so only `a` publishes metrics. + let b = MeshEndpoint::bind_with_secret_key(SecretKey::from_bytes(&[6; 32]), loopback_any()) + .await + .unwrap(); + let b_addrs: Vec = b.ip_addrs().iter().map(|s| s.to_string()).collect(); + // Accept every dial (the spawned reconcile loop may dial too) and + // keep the connections open; report the first. + let (accepted_tx, mut accepted_rx) = mpsc::channel(8); + let b_accept = tokio::spawn({ + let b = b.clone(); + async move { + let mut held = Vec::new(); + while let Ok(Some(peer)) = b.accept().await { + let _ = accepted_tx.try_send(()); + held.push(peer); + } + } + }); + a.membership() + .apply_gossip_record(GossipRecord::new(b.runtime_id(), b_addrs, 1)); + + a.reconcile_now().await; + timeout(Duration::from_secs(5), accepted_rx.recv()) + .await + .expect("b should accept a's dial"); + + let mut dial_ok = 0; + let mut dial_samples = 0; + let mut pass_samples = 0; + let mut peer_records = 0.0; + let mut saw_candidates = false; + let mut saw_max_age = false; + for (key, _, _, value) in snapshotter.snapshot().into_vec() { + let key = key.key(); + let label = |name: &str| { + key.labels() + .find(|l| l.key() == name) + .map(|l| l.value().to_owned()) + }; + match (key.name(), value) { + (telemetry::DIAL_TOTAL, DebugValue::Counter(n)) + if label("outcome").as_deref() == Some("ok") => + { + dial_ok += n + } + (telemetry::DIAL_DURATION_SECONDS, DebugValue::Histogram(v)) => { + dial_samples += v.len() + } + (telemetry::RECONCILE_PASS_SECONDS, DebugValue::Histogram(v)) => { + pass_samples += v.len() + } + (telemetry::PEER_RECORDS, DebugValue::Gauge(v)) => peer_records += v.into_inner(), + (telemetry::RECONCILE_CANDIDATES, DebugValue::Gauge(_)) => saw_candidates = true, + (telemetry::PEER_RECORD_MAX_AGE_SECONDS, DebugValue::Gauge(_)) => { + saw_max_age = true + } + _ => {} + } + } + assert!(dial_ok >= 1, "successful dial must be counted as ok"); + assert!(dial_samples >= 1, "dial duration must be recorded"); + assert!( + pass_samples >= 1, + "reconcile pass duration must be recorded" + ); + assert_eq!( + peer_records, 1.0, + "exactly one peer record (b) across all series" + ); + assert!(saw_candidates && saw_max_age); + a.shutdown(); + b_accept.abort(); + } + + #[tokio::test] + async fn reconcile_pass_prunes_expired_peer_records() { + use metrics_util::debugging::{DebugValue, DebuggingRecorder}; + + let recorder = DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + let _guard = metrics::set_default_local_recorder(&recorder); + + let ttl = Duration::from_millis(100); + let (a, _) = runtime_with(7, |m| m.with_peer_record_ttl(ttl), RECONCILE_INTERVAL).await; + // A dead runtime's record: draining, so no pass tries to dial it. + let mut ghost = GossipRecord::new(RuntimeId([9; 32]), vec!["127.0.0.1:9".into()], 1); + ghost.draining = true; + assert!(a.membership().apply_gossip_record(ghost)); + assert!(a.membership().has_peer(RuntimeId([9; 32]))); + + tokio::time::sleep(ttl * 3).await; + a.reconcile_now().await; + + assert!( + a.membership().status().peers.is_empty(), + "the reconcile pass must prune the expired record from the table" + ); + let pruned: u64 = snapshotter + .snapshot() + .into_vec() + .into_iter() + .filter_map(|(key, _, _, value)| { + let key = key.key(); + let is_pruned = key.name() == telemetry::PEER_RECORDS_EXPIRED_TOTAL + && key + .labels() + .any(|l| l.key() == "reason" && l.value() == "pruned"); + match value { + DebugValue::Counter(n) if is_pruned => Some(n), + _ => None, + } + }) + .sum(); + assert!(pruned >= 1, "prune must count reason=\"pruned\""); + a.shutdown(); + } + + #[tokio::test] + async fn dial_skips_candidate_that_expired_after_the_pass_snapshot() { + use metrics_util::debugging::DebuggingRecorder; + + let recorder = DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + let _guard = metrics::set_default_local_recorder(&recorder); + + let ttl = Duration::from_millis(300); + let (a, _) = runtime_with(11, |m| m.with_peer_record_ttl(ttl), QUIET).await; + let (b, b_addrs, b_accept) = accepting_endpoint(12, false).await; + let b_id = b.runtime_id(); + // Let the background loop's startup pass finish (empty table), so + // every dial below is one this test made. + timeout(Duration::from_secs(5), async { + while !snapshotter + .snapshot() + .into_vec() + .into_iter() + .any(|(key, _, _, _)| key.key().name() == telemetry::RECONCILE_PASS_SECONDS) + { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("startup reconcile pass should run"); + a.membership() + .apply_gossip_record(GossipRecord::new(b_id, b_addrs.clone(), 1)); + + let candidates = dial_candidates(&a.inner); + assert_eq!(candidates, vec![b_id]); + // Stands in for earlier dials in the same pass: b's record expires + // while the pass is still working through its snapshot. + tokio::time::sleep(ttl * 2).await; + for runtime_id in candidates { + dial_peer(&a.inner, runtime_id).await; + } + assert_eq!( + dial_count(&snapshotter), + 0, + "an expired candidate must not be dialed" + ); + assert!(a.connected_peers().is_empty()); + + // Control: the same address dials fine once b's record is fresh. + a.membership() + .apply_gossip_record(GossipRecord::new(b_id, b_addrs, 2)); + dial_peer(&a.inner, b_id).await; + assert_eq!(a.connected_peers(), vec![b_id]); + assert_eq!(state_of(&a, b_id), Some(ConnectionState::Connected)); + a.shutdown(); + b_accept.abort(); + } + + #[tokio::test] + async fn finishing_dial_does_not_overwrite_a_connection_that_landed() { + let (a, b) = connected_pair().await; + let b_id = b.local_runtime_id(); + assert_eq!(state_of(&a, b_id), Some(ConnectionState::Connected)); + + // A dial that was pending when the connection landed ends with a + // failure write; it must not clobber the live connection's state. + assert!(!mark_dial_state( + &a.inner, + b_id, + ConnectionState::Disconnected + )); + assert!(!mark_dial_state( + &a.inner, + b_id, + ConnectionState::Connecting + )); + dial_peer(&a.inner, b_id).await; + assert_eq!(state_of(&a, b_id), Some(ConnectionState::Connected)); + assert_eq!(a.connected_peers(), vec![b_id]); + a.shutdown(); + b.shutdown(); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + // Installs an already-closed connection and checks the peer ends + // Disconnected and then expires. This does not deterministically hit the + // close-between-insert-and-state-write window inside install_peer; the + // reconcile re-sync (reconcile_pass_resyncs_connection_state_from_transport) + // is what guarantees a stale Connected flag cannot outlive that race. + async fn installing_closed_connection_ends_disconnected_and_expires() { + let ttl = Duration::from_millis(300); + let (a, _) = runtime_with(13, |m| m.with_peer_record_ttl(ttl), QUIET).await; + let (b, b_addrs, b_accept) = accepting_endpoint(14, true).await; + let b_id = b.runtime_id(); + // Draining, as in a rollout: no reconcile pass dials it, so nothing + // but the connection state stands between it and expiry. + let mut record = GossipRecord::new(b_id, b_addrs.clone(), 1); + record.draining = true; + a.membership().apply_gossip_record(record); + + let sock = b_addrs[0].parse().unwrap(); + let peer = a + .inner + .endpoint + .connect(direct_addr(b_id, sock).unwrap()) + .await + .unwrap(); + // b drops the connection; wait until a's side has observed the close, + // so the receive tasks install_peer spawns fail straight away. + timeout(Duration::from_secs(5), peer.recv_datagram()) + .await + .expect("b should close the connection") + .expect_err("closed connection yields no datagram"); + install_peer(&a.inner, peer, true); + + timeout(Duration::from_secs(5), async { + while !a.connected_peers().is_empty() { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("the closed connection must be removed"); + assert_eq!(state_of(&a, b_id), Some(ConnectionState::Disconnected)); + + // Without a stale Connected flag, the TTL exemption does not apply. + tokio::time::sleep(ttl * 2).await; + assert_eq!(a.membership().prune_expired(), 1); + assert_eq!(state_of(&a, b_id), None); + a.shutdown(); + b_accept.abort(); + } + + #[tokio::test] + async fn reconcile_pass_resyncs_connection_state_from_transport() { + let (a, b) = connected_pair().await; + let b_id = b.local_runtime_id(); + let mut ghost = GossipRecord::new(RuntimeId([9; 32]), vec!["127.0.0.1:9".into()], 1); + ghost.draining = true; + assert!(a.membership().apply_gossip_record(ghost)); + + // Simulate writes the event-driven paths missed, in both directions. + a.membership() + .mark_connection_state(b_id, ConnectionState::Disconnected); + a.membership() + .mark_connection_state(RuntimeId([9; 32]), ConnectionState::Connected); + a.reconcile_now().await; + + assert_eq!(state_of(&a, b_id), Some(ConnectionState::Connected)); + assert_eq!( + state_of(&a, RuntimeId([9; 32])), + Some(ConnectionState::Disconnected) + ); + a.shutdown(); + b.shutdown(); + } + #[test] fn tie_break_is_symmetric() { let small = RuntimeId([1u8; 32]); diff --git a/crates/buzz-relay-mesh/src/telemetry.rs b/crates/buzz-relay-mesh/src/telemetry.rs new file mode 100644 index 0000000000..e83547e5e3 --- /dev/null +++ b/crates/buzz-relay-mesh/src/telemetry.rs @@ -0,0 +1,149 @@ +//! Prometheus metrics for the mesh reconcile loop and peer table. +//! +//! These answer two operational questions without exec-ing into a pod: +//! +//! - **Where is a reconcile pass spending its time?** A pass first rescans the +//! ready registry, then dials every peer it is not yet connected to, one at +//! a time. `buzz_mesh_reconcile_candidates` is set once the scan finishes, +//! so a pass stuck before it is stuck in the scan; the dial counter and +//! histogram show progress and cost inside the dial loop. +//! - **How big and how stale is the peer table?** `buzz_mesh_peer_records` +//! and `buzz_mesh_peer_record_max_age_seconds` are refreshed every pass; +//! `buzz_mesh_peer_records_expired_total` counts records dropped for age. +//! +//! Label values are closed vocabularies: `outcome` ∈ {`ok`, `error`}, +//! `state` ∈ [`ConnectionState`] in snake_case, `draining` ∈ {`true`, +//! `false`}, `reason` ∈ {`pruned`, `rejected`}. + +use std::time::Duration; + +use crate::status::ConnectionState; + +/// Gauge: peers the current reconcile pass will dial. +pub const RECONCILE_CANDIDATES: &str = "buzz_mesh_reconcile_candidates"; +/// Histogram: wall-clock duration of one full reconcile pass. +pub const RECONCILE_PASS_SECONDS: &str = "buzz_mesh_reconcile_pass_seconds"; +/// Counter: outbound connect attempts by `outcome`. +pub const DIAL_TOTAL: &str = "buzz_mesh_dial_total"; +/// Histogram: duration of one outbound connect attempt. +pub const DIAL_DURATION_SECONDS: &str = "buzz_mesh_dial_duration_seconds"; +/// Gauge: peer-table records by effective `state` and `draining`. +pub const PEER_RECORDS: &str = "buzz_mesh_peer_records"; +/// Gauge: age of the oldest peer-table heartbeat, in seconds. +pub const PEER_RECORD_MAX_AGE_SECONDS: &str = "buzz_mesh_peer_record_max_age_seconds"; +/// Counter: records dropped for an expired heartbeat, by `reason`. +pub const PEER_RECORDS_EXPIRED_TOTAL: &str = "buzz_mesh_peer_records_expired_total"; + +/// Every effective connection state, in `state` label order. +pub(crate) const CONNECTION_STATES: [ConnectionState; 4] = [ + ConnectionState::Disconnected, + ConnectionState::Connecting, + ConnectionState::Connected, + ConnectionState::Suspect, +]; + +/// Register metric descriptions. Call once after the global recorder exists. +pub fn describe_metrics() { + metrics::describe_gauge!( + RECONCILE_CANDIDATES, + "Mesh peers the current reconcile pass will dial (set after the registry scan)" + ); + metrics::describe_histogram!( + RECONCILE_PASS_SECONDS, + metrics::Unit::Seconds, + "Wall-clock duration of one mesh reconcile pass (registry scan plus dials)" + ); + metrics::describe_counter!( + DIAL_TOTAL, + "Outbound mesh connect attempts, by outcome (ok, error)" + ); + metrics::describe_histogram!( + DIAL_DURATION_SECONDS, + metrics::Unit::Seconds, + "Duration of one outbound mesh connect attempt" + ); + metrics::describe_gauge!( + PEER_RECORDS, + "Mesh peer-table records, by effective connection state and draining flag" + ); + metrics::describe_gauge!( + PEER_RECORD_MAX_AGE_SECONDS, + metrics::Unit::Seconds, + "Age of the oldest heartbeat in the mesh peer table" + ); + metrics::describe_counter!( + PEER_RECORDS_EXPIRED_TOTAL, + "Mesh peer records dropped for an expired heartbeat (pruned from the table, or rejected on arrival)" + ); +} + +pub(crate) const fn state_label(state: ConnectionState) -> &'static str { + match state { + ConnectionState::Disconnected => "disconnected", + ConnectionState::Connecting => "connecting", + ConnectionState::Connected => "connected", + ConnectionState::Suspect => "suspect", + } +} + +/// Position of `state` in [`CONNECTION_STATES`]. +const fn state_index(state: ConnectionState) -> usize { + match state { + ConnectionState::Disconnected => 0, + ConnectionState::Connecting => 1, + ConnectionState::Connected => 2, + ConnectionState::Suspect => 3, + } +} + +const fn draining_label(draining: bool) -> &'static str { + if draining { + "true" + } else { + "false" + } +} + +/// Peer-table shape at one instant, published as gauges. +#[derive(Clone, Debug, Default, PartialEq)] +pub(crate) struct PeerTableStats { + /// `counts[state][draining as usize]`, indexed like [`CONNECTION_STATES`]. + pub counts: [[u64; 2]; CONNECTION_STATES.len()], + /// Oldest heartbeat age; zero for an empty table. + pub max_age: Duration, +} + +impl PeerTableStats { + pub(crate) fn record(&mut self, state: ConnectionState, draining: bool, age: Duration) { + self.counts[state_index(state)][usize::from(draining)] += 1; + self.max_age = self.max_age.max(age); + } + + /// Set every `(state, draining)` series, zeros included, so a series + /// that empties reads 0 instead of holding its last non-zero value. + pub(crate) fn publish(&self) { + for (index, state) in CONNECTION_STATES.iter().enumerate() { + for draining in [false, true] { + metrics::gauge!( + PEER_RECORDS, + "state" => state_label(*state), + "draining" => draining_label(draining), + ) + .set(self.counts[index][usize::from(draining)] as f64); + } + } + metrics::gauge!(PEER_RECORD_MAX_AGE_SECONDS).set(self.max_age.as_secs_f64()); + } +} + +pub(crate) fn record_dial(ok: bool, elapsed: Duration) { + let outcome = if ok { "ok" } else { "error" }; + metrics::counter!(DIAL_TOTAL, "outcome" => outcome).increment(1); + metrics::histogram!(DIAL_DURATION_SECONDS).record(elapsed.as_secs_f64()); +} + +pub(crate) fn record_expired(reason: &'static str, count: u64) { + if count > 0 { + metrics::counter!(PEER_RECORDS_EXPIRED_TOTAL, "reason" => reason).increment(count); + } +} diff --git a/crates/buzz-relay/src/mesh_boot.rs b/crates/buzz-relay/src/mesh_boot.rs index 63a346258f..28fe158c56 100644 --- a/crates/buzz-relay/src/mesh_boot.rs +++ b/crates/buzz-relay/src/mesh_boot.rs @@ -32,6 +32,7 @@ use buzz_relay_mesh::{ }; use crate::config::Config; +use crate::startup_steps::{StartupStep, StepTimer}; use crate::tunnel::directory::SessionDirectory; use crate::tunnel::reliable::{ReliableFrame, ReliableInbound, ReliableStreamRouter}; @@ -527,6 +528,9 @@ pub async fn boot_mesh( return Ok(None); } + // Each await that can stall boot is its own startup step, so a pod stuck + // in `mesh_boot` also shows which part (see `startup_steps`). + let step = StepTimer::start(StartupStep::MeshBind); let endpoint = MeshEndpoint::bind(config.mesh.bind_addr) .await .map_err(|e| { @@ -535,6 +539,7 @@ pub async fn boot_mesh( config.mesh.bind_addr ) })?; + step.finish(); let runtime_id = endpoint.runtime_id(); let addrs = advertise_addrs(&endpoint); tracing::info!( @@ -563,10 +568,12 @@ pub async fn boot_mesh( // First publish is part of boot: if Redis can't take the attested record, // peers can never find us — fail loudly now, not quietly forever. + let step = StepTimer::start(StartupStep::MeshPublish); registry .publish_ready(&ready_record) .await .map_err(|e| anyhow::anyhow!("mesh ready-registry publish failed: {e}"))?; + step.finish(); tracing::info!(runtime_id = %runtime_id, "mesh ready record published"); // The heartbeat is shutdown-gated only. Readiness terms deliberately do @@ -581,7 +588,9 @@ pub async fn boot_mesh( let runtime = MeshRuntime::start(endpoint, membership, Some(registry)); let owners = Arc::new(crate::audio::join::HuddleOwnerRegistry::new()); // Dial seed peers now rather than waiting for the first reconcile tick. + let step = StepTimer::start(StartupStep::MeshInitialReconcile); runtime.reconcile_now().await; + step.finish(); // Drain watcher: SIGTERM flips `shutting_down`; gossip `draining=true` so // peers stop routing new sessions here, then actively drain locally-owned @@ -657,6 +666,73 @@ mod tests { assert!(handle.is_none()); } + /// Each await in `boot_mesh` that can stall startup is its own startup + /// step, so a stuck or failed boot names the step. With Redis unreachable, + /// `mesh_bind` and `mesh_publish` both end and the reconcile step never + /// starts. + #[test] + fn mesh_boot_sub_steps_name_the_step_that_failed() { + use metrics_util::debugging::{DebugValue, DebuggingRecorder}; + + use crate::startup_steps::{CURRENT_METRIC, SECONDS_METRIC}; + + // Held around `block_on`, not across an `.await`: the step timers fire + // the log callsites `startup_steps::tests` capture. + let _tracing = crate::test_support::tracing_dispatch_lock(); + let recorder = DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + let result = metrics::with_local_recorder(&recorder, || { + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap() + .block_on(async { + let mut config = crate::config::Config::for_test(); // [FI-TRACE-ENV-RACE] + config.mesh.enabled = true; + config.mesh.bind_addr = "127.0.0.1:0".parse().unwrap(); + let pool = deadpool_redis::Config::from_url("redis://127.0.0.1:1") // refused + .create_pool(Some(deadpool_redis::Runtime::Tokio1)) + .unwrap(); + let db = buzz_db::Db::from_pool( + sqlx::postgres::PgPoolOptions::new() + .max_connections(1) + .connect_lazy("postgres://unused:unused@127.0.0.1:1/unused") // sadscan:disable np.postgres.1 + .expect("lazy database pool"), + ); + let keys = nostr::Keys::generate(); + boot_mesh(&config, pool, db, &keys, Arc::new(AtomicBool::new(false))).await + }) + }); + let Err(err) = result else { + panic!("boot must fail when Redis is unreachable"); + }; + assert!(err.to_string().contains("publish"), "{err}"); + + let gauges: std::collections::HashMap<(String, String), f64> = snapshotter + .snapshot() + .into_vec() + .into_iter() + .filter_map(|(key, _, _, value)| { + let key = key.key(); + let phase = key + .labels() + .find(|l| l.key() == "phase")? + .value() + .to_owned(); + match value { + DebugValue::Gauge(v) => Some(((key.name().to_owned(), phase), v.into_inner())), + _ => None, + } + }) + .collect(); + let get = |metric: &str, phase: &str| gauges.get(&(metric.to_owned(), phase.to_owned())); + for phase in ["mesh_bind", "mesh_publish"] { + assert_eq!(get(CURRENT_METRIC, phase), Some(&0.0), "{phase} must end"); + assert!(get(SECONDS_METRIC, phase).is_some(), "{phase} duration"); + } + assert_eq!(get(CURRENT_METRIC, "mesh_initial_reconcile"), None); + } + /// Blocker fix (Wren review of 8b077fdb): absent `BUZZ_MESH`, the mesh is /// OFF — an env-untouched image upgrade must not bind or write Redis. #[test] diff --git a/crates/buzz-relay/src/metrics.rs b/crates/buzz-relay/src/metrics.rs index 67e4d3b35e..893813a6fb 100644 --- a/crates/buzz-relay/src/metrics.rs +++ b/crates/buzz-relay/src/metrics.rs @@ -132,6 +132,13 @@ const GIT_DURATION_BUCKETS_S: [f64; 13] = [ 0.01, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 30.0, 60.0, 120.0, 300.0, ]; +/// Mesh dial and reconcile-pass buckets. A pass dials peers serially, so a +/// large or stale peer table can stretch it to minutes; the default +/// five-second ceiling would hide exactly the stall these exist to show. +const MESH_DURATION_BUCKETS_S: [f64; 14] = [ + 0.01, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 30.0, 60.0, 120.0, 300.0, 600.0, +]; + /// Byte buckets for hydrated repositories and streamed clone/fetch responses. const GIT_BYTES_BUCKETS: [f64; 9] = [ 0.0, @@ -234,6 +241,16 @@ fn configured_prometheus_builder(gauge_idle_timeout_secs: u64) -> PrometheusBuil &GIT_PACK_BUCKETS, ) .expect("valid git compaction output pack-count bucket boundaries") + .set_buckets_for_metric( + Matcher::Full(buzz_relay_mesh::telemetry::DIAL_DURATION_SECONDS.to_owned()), + &MESH_DURATION_BUCKETS_S, + ) + .expect("valid mesh dial bucket boundaries") + .set_buckets_for_metric( + Matcher::Full(buzz_relay_mesh::telemetry::RECONCILE_PASS_SECONDS.to_owned()), + &MESH_DURATION_BUCKETS_S, + ) + .expect("valid mesh reconcile pass bucket boundaries") .set_buckets_for_metric(Matcher::Suffix("_seconds".to_owned()), &DURATION_BUCKETS_S) .expect("valid seconds bucket boundaries") .set_buckets_for_metric( @@ -298,6 +315,7 @@ pub fn try_install(port: u16, gauge_idle_timeout_secs: u64) -> Result<(), Metric describe_db_pool_metrics(); describe_auth_metrics(); crate::startup_steps::describe_metrics(); + buzz_relay_mesh::telemetry::describe_metrics(); initialize_auth_metric_series(); initialize_partition_audit_metric_series(); tokio::spawn(exporter); diff --git a/crates/buzz-relay/src/startup_steps.rs b/crates/buzz-relay/src/startup_steps.rs index 6d95057414..1ef069ccd7 100644 --- a/crates/buzz-relay/src/startup_steps.rs +++ b/crates/buzz-relay/src/startup_steps.rs @@ -13,6 +13,10 @@ //! //! The `phase` label values are a closed vocabulary ([`StartupStep::ALL`]). //! They never overlap the early lifecycle phase names. +//! +//! A step may contain sub-steps: `mesh_boot` runs `mesh_bind`, +//! `mesh_publish`, then `mesh_initial_reconcile`, so a pod stuck in mesh boot +//! shows both `mesh_boot` and the sub-step it is on as current. use std::time::Instant; @@ -63,6 +67,12 @@ pub enum StartupStep { NipFiSnapshotLoad, /// Start the inter-relay mesh (no-op when the mesh is off). MeshBoot, + /// Inside `MeshBoot`: bind the mesh UDP endpoint. + MeshBind, + /// Inside `MeshBoot`: publish this runtime's ready record to Redis. + MeshPublish, + /// Inside `MeshBoot`: the first registry scan and peer-dial pass. + MeshInitialReconcile, /// Run the git object-store conformance probe (when enabled). GitConformanceProbe, /// Verify the channel roster fence. @@ -90,6 +100,9 @@ impl StartupStep { Self::NipFiJwksWarm, Self::NipFiSnapshotLoad, Self::MeshBoot, + Self::MeshBind, + Self::MeshPublish, + Self::MeshInitialReconcile, Self::GitConformanceProbe, Self::ChannelRosterFence, Self::LargeRosterReconcile, @@ -112,6 +125,9 @@ impl StartupStep { Self::NipFiJwksWarm => "nip_fi_jwks_warm", Self::NipFiSnapshotLoad => "nip_fi_snapshot_load", Self::MeshBoot => "mesh_boot", + Self::MeshBind => "mesh_bind", + Self::MeshPublish => "mesh_publish", + Self::MeshInitialReconcile => "mesh_initial_reconcile", Self::GitConformanceProbe => "git_conformance_probe", Self::ChannelRosterFence => "channel_roster_fence", Self::LargeRosterReconcile => "large_roster_reconcile", diff --git a/scripts/run-tests.sh b/scripts/run-tests.sh index 8bd7842bff..c6a83b2a09 100755 --- a/scripts/run-tests.sh +++ b/scripts/run-tests.sh @@ -101,6 +101,10 @@ run_unit_tests() { run_test_step "buzz-voice tests" \ cargo test -p buzz-voice --lib -- --nocapture + # Inter-relay mesh crate (infra-free). Mirrors `just test-unit`. + run_test_step "buzz-relay-mesh tests" \ + cargo test -p buzz-relay-mesh -- --nocapture + run_test_step "buzz-cli tests" \ cargo test -p buzz-cli -- --nocapture @@ -236,6 +240,9 @@ run_unit_tests() { run_test_step "buzz-relay startup step tests" \ cargo test -p buzz-relay --lib startup_steps::tests:: -- --nocapture + run_test_step "buzz-relay mesh boot tests" \ + cargo test -p buzz-relay --lib mesh_boot::tests:: -- --nocapture + # telemetry and config tests mutate process env. libtest runs one process # with many threads (nextest, used by `just test-unit`, isolates each test), # so run them single-threaded here. `--skip` keeps the substring filter