jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-27 04:30:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(mesh): expire dead peer records and instrument mesh boot (#8503)

Browse Source 
## 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>
This commit is contained in:
Luke Tornquist
2026-10-26 09:47:30 -04:00
committed by GitHub
parent d676c7b8db
commit 7ce422417b
11 changed files with 1018 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
crates/buzz-relay-mesh/Cargo.toml
View File

@@ -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 }

4
crates/buzz-relay-mesh/src/lib.rs
View File

@@ -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).

327
crates/buzz-relay-mesh/src/membership.rs
View File

@@ -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<String>,
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<GossipRecord> {
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<GossipRecord> {
self.records_at(crate::gossip::now_millis())
}
fn records_at(&self, now_millis: u64) -> Vec<GossipRecord> {
let mut records: Vec<GossipRecord> = 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<PeerInfo> {
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::<u64>(), 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));

460
crates/buzz-relay-mesh/src/runtime.rs
View File

@@ -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<Inner>, 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<Inner>) {
loop {
match inner.endpoint.accept().await {
@@ -327,66 +339,93 @@ async fn reconcile_loop(inner: Arc<Inner>) {
}
async fn reconcile_once(inner: &Arc<Inner>) {
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<RuntimeId> {
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<Inner>, 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<Inner>, 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<Inner>, 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<String>) {
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<String>) {
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<String>, 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<ConnectionState> {
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<String> = 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]);

149
crates/buzz-relay-mesh/src/telemetry.rs Normal file
View File

@@ -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);
}
}

76
crates/buzz-relay/src/mesh_boot.rs
View File

@@ -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]

18
crates/buzz-relay/src/metrics.rs
View File

@@ -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);

16
crates/buzz-relay/src/startup_steps.rs
View File

@@ -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",