Phase 1b: chitchat SWIM gossip + cluster-status #2
Generated
+141
-1
@@ -11,6 +11,12 @@ dependencies = [
|
||||
"memchr",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "allocator-api2"
|
||||
version = "0.2.21"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "683d7910e743518b0e34f1186f92494becacb047c7b6bf616c96772180fef923"
|
||||
|
||||
[[package]]
|
||||
name = "android_system_properties"
|
||||
version = "0.1.5"
|
||||
@@ -191,6 +197,8 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "dad887fd958be91b5098c0248def011f4523ab786cd411be668777e55063501f"
|
||||
dependencies = [
|
||||
"find-msvc-tools",
|
||||
"jobserver",
|
||||
"libc",
|
||||
"shlex",
|
||||
]
|
||||
|
||||
@@ -200,6 +208,36 @@ version = "1.0.4"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801"
|
||||
|
||||
[[package]]
|
||||
name = "chacha20"
|
||||
version = "0.10.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "d524456ba66e72eb8b115ff89e01e497f8e6d11d78b70b1aa13c0fbd97540a81"
|
||||
dependencies = [
|
||||
"cfg-if",
|
||||
"cpufeatures",
|
||||
"rand_core",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "chitchat"
|
||||
version = "0.11.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "9e1cfd5d16c6fc43487006ab5b5c910f295cc53cf07d90491fb36c8dc479c754"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"async-trait",
|
||||
"bytes",
|
||||
"itertools",
|
||||
"lru",
|
||||
"rand",
|
||||
"serde",
|
||||
"tokio",
|
||||
"tokio-stream",
|
||||
"tracing",
|
||||
"zstd",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "chrono"
|
||||
version = "0.4.45"
|
||||
@@ -260,6 +298,7 @@ version = "0.3.0"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"axum",
|
||||
"chitchat",
|
||||
"chrono",
|
||||
"clap",
|
||||
"http-body-util",
|
||||
@@ -289,6 +328,15 @@ version = "0.8.7"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b"
|
||||
|
||||
[[package]]
|
||||
name = "cpufeatures"
|
||||
version = "0.3.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "8b2a41393f66f16b0823bb79094d54ac5fbd34ab292ddafb9a0456ac9f87d201"
|
||||
dependencies = [
|
||||
"libc",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "crossbeam-deque"
|
||||
version = "0.8.6"
|
||||
@@ -354,6 +402,12 @@ version = "0.1.5"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "d9c4f5dac5e15c24eb999c26181a6ca40b39fe946cbe4c263c7209467bc83af2"
|
||||
|
||||
[[package]]
|
||||
name = "foldhash"
|
||||
version = "0.2.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "77ce24cb58228fbb8aa041425bb1050850ac19177686ea6e0f41a70416f56fdb"
|
||||
|
||||
[[package]]
|
||||
name = "form_urlencoded"
|
||||
version = "1.2.2"
|
||||
@@ -411,6 +465,7 @@ dependencies = [
|
||||
"cfg-if",
|
||||
"libc",
|
||||
"r-efi",
|
||||
"rand_core",
|
||||
"wasip2",
|
||||
"wasip3",
|
||||
]
|
||||
@@ -421,7 +476,7 @@ version = "0.15.5"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "9229cfe53dfd69f0609a49f65461bd93001ea1ef889cd5529dd176593f5338a1"
|
||||
dependencies = [
|
||||
"foldhash",
|
||||
"foldhash 0.1.5",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -429,6 +484,11 @@ name = "hashbrown"
|
||||
version = "0.17.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a"
|
||||
dependencies = [
|
||||
"allocator-api2",
|
||||
"equivalent",
|
||||
"foldhash 0.2.0",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "heck"
|
||||
@@ -570,12 +630,31 @@ version = "1.70.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "a6cb138bb79a146c1bd460005623e142ef0181e3d0219cb493e02f7d08a35695"
|
||||
|
||||
[[package]]
|
||||
name = "itertools"
|
||||
version = "0.14.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "2b192c782037fadd9cfa75548310488aabdbf3d2da73885b31bd0abd03351285"
|
||||
dependencies = [
|
||||
"either",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "itoa"
|
||||
version = "1.0.18"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682"
|
||||
|
||||
[[package]]
|
||||
name = "jobserver"
|
||||
version = "0.1.35"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "1c00acbd29eabad4a2392fa0e921c874934dbbf4194312ad20f04a0ed67a3cb3"
|
||||
dependencies = [
|
||||
"getrandom",
|
||||
"libc",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "js-sys"
|
||||
version = "0.3.102"
|
||||
@@ -626,6 +705,15 @@ version = "0.4.32"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "953f07c43838f8e6f9758cab68bf5bed85465e7587ebe0b823f1bcd81978ad3a"
|
||||
|
||||
[[package]]
|
||||
name = "lru"
|
||||
version = "0.17.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "0e0b564323a0fb6d54b864f625ae139de9612e27edb944dda37c109f05aac531"
|
||||
dependencies = [
|
||||
"hashbrown 0.17.1",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "matchers"
|
||||
version = "0.2.0"
|
||||
@@ -748,6 +836,12 @@ version = "0.2.17"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd"
|
||||
|
||||
[[package]]
|
||||
name = "pkg-config"
|
||||
version = "0.3.33"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "19f132c84eca552bf34cab8ec81f1c1dcc229b811638f9d283dceabe58c5569e"
|
||||
|
||||
[[package]]
|
||||
name = "prettyplease"
|
||||
version = "0.2.37"
|
||||
@@ -782,6 +876,23 @@ version = "6.0.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf"
|
||||
|
||||
[[package]]
|
||||
name = "rand"
|
||||
version = "0.10.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "c7f5fa3a058cd35567ef9bfa5e75732bee0f9e4c55fa90477bef2dfcdbc4be80"
|
||||
dependencies = [
|
||||
"chacha20",
|
||||
"getrandom",
|
||||
"rand_core",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "rand_core"
|
||||
version = "0.10.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "63b8176103e19a2643978565ca18b50549f6101881c443590420e4dc998a3c69"
|
||||
|
||||
[[package]]
|
||||
name = "rayon"
|
||||
version = "1.12.0"
|
||||
@@ -1084,6 +1195,7 @@ dependencies = [
|
||||
"futures-core",
|
||||
"pin-project-lite",
|
||||
"tokio",
|
||||
"tokio-util",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -1669,3 +1781,31 @@ name = "zmij"
|
||||
version = "1.0.21"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "b8848ee67ecc8aedbaf3e4122217aff892639231befc6a1b58d29fff4c2cabaa"
|
||||
|
||||
[[package]]
|
||||
name = "zstd"
|
||||
version = "0.13.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "e91ee311a569c327171651566e07972200e76fcfe2242a4fa446149a3881c08a"
|
||||
dependencies = [
|
||||
"zstd-safe",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "zstd-safe"
|
||||
version = "7.2.4"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "8f49c4d5f0abb602a93fb8736af2a4f4dd9512e36f7f570d66e65ff867ed3b9d"
|
||||
dependencies = [
|
||||
"zstd-sys",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "zstd-sys"
|
||||
version = "2.0.16+zstd.1.5.7"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "91e19ebc2adc8f83e43039e79776e3fda8ca919132d68a1fed6a5faca2683748"
|
||||
dependencies = [
|
||||
"cc",
|
||||
"pkg-config",
|
||||
]
|
||||
|
||||
@@ -28,6 +28,10 @@ libc = "0.2"
|
||||
# v0.2.0 — sibling-tempfile + rename for atomic manifest persistence.
|
||||
# Was previously dev-dep only; promoted to main.
|
||||
tempfile = "3"
|
||||
# v0.11.1 — cluster membership via scuttlebutt gossip + phi-accrual failure
|
||||
# detection. Wraps in cluster/gossip.rs. UDP transport, keyed KV state per
|
||||
# node, seed_nodes bootstrap from [[cluster.peers]] config.
|
||||
chitchat = "0.11"
|
||||
|
||||
[dev-dependencies]
|
||||
tempfile = "3"
|
||||
|
||||
@@ -1,15 +1,21 @@
|
||||
//! Cluster peer routing.
|
||||
//! Cluster peer routing and gossip.
|
||||
//!
|
||||
//! Phase 1a of the v2 distributed FS. Every fleet node has two potential
|
||||
//! endpoints: a LAN socket (fast, direct, no WireGuard framing) and a
|
||||
//! Tailscale socket (fallback, always reachable when the tailnet is up).
|
||||
//! This module races LAN first and falls through to Tailscale, returning
|
||||
//! the winning route so the caller can cache it for the session.
|
||||
//! Two submodules of the v2 distributed FS:
|
||||
//! * this file — LAN-first peer probe (Phase 1a).
|
||||
//! * [`gossip`] — chitchat SWIM membership + KV state broadcast (Phase 1b).
|
||||
//!
|
||||
//! Every fleet node has two potential endpoints: a LAN socket (fast, direct,
|
||||
//! no WireGuard framing) and a Tailscale socket (fallback, always reachable
|
||||
//! when the tailnet is up). This file races LAN first and falls through to
|
||||
//! Tailscale, returning the winning route so the caller can cache it for
|
||||
//! the session.
|
||||
//!
|
||||
//! Cache the returned [`RouteWinner`] and re-probe on transport error or
|
||||
//! after ~5 min so route changes (node moved networks, LAN NIC came back)
|
||||
//! propagate without a daemon restart.
|
||||
|
||||
pub mod gossip;
|
||||
|
||||
use crate::config::PeerEntry;
|
||||
use anyhow::{bail, Result};
|
||||
use std::net::SocketAddr;
|
||||
|
||||
@@ -0,0 +1,549 @@
|
||||
//! Cluster membership + KV state broadcast via chitchat SWIM gossip.
|
||||
//!
|
||||
//! Wraps `chitchat::spawn_chitchat` with clawstor-specific keys and a
|
||||
//! typed [`PeerView`] readout. Bootstrapped from static
|
||||
//! `[[cluster.peers]]` seed nodes in the config; membership extends
|
||||
//! dynamically as nodes come and go.
|
||||
//!
|
||||
//! # Transport
|
||||
//!
|
||||
//! Chitchat's built-in UDP transport listens on the [`ClusterConfig`]
|
||||
//! bind address. RPC transport (Phase 1c, QUIC) will use a different
|
||||
//! UDP port so gossip and RPC don't collide — QUIC is UDP too.
|
||||
//!
|
||||
//! # State advertised per node
|
||||
//!
|
||||
//! Well-known keys enumerated in [`keys`]. Peers publish their zone,
|
||||
//! RPC endpoints, hot-tier occupancy, and the warm-tier projects they
|
||||
//! serve — the last of which drives Gitea runner label dynamics later.
|
||||
|
||||
use crate::config::ClusterConfig;
|
||||
use anyhow::{bail, Context, Result};
|
||||
use chitchat::transport::UdpTransport;
|
||||
use chitchat::{
|
||||
spawn_chitchat, Chitchat, ChitchatConfig, ChitchatHandle, ChitchatId,
|
||||
FailureDetectorConfig,
|
||||
};
|
||||
use std::net::SocketAddr;
|
||||
use std::sync::Arc;
|
||||
use std::time::{Duration, SystemTime, UNIX_EPOCH};
|
||||
use tokio::sync::Mutex;
|
||||
|
||||
/// Well-known kv keys every clawstor node broadcasts under its own state.
|
||||
pub mod keys {
|
||||
/// Human-facing node name (mirrors the `[[cluster.peers]].name` from
|
||||
/// each peer's own config).
|
||||
pub const NODE_NAME: &str = "clawstor.name";
|
||||
/// Zone tag: `"fabric-10g"`, `"lan-1g"`, `"roaming"`, or a custom label.
|
||||
pub const ZONE: &str = "clawstor.zone";
|
||||
/// LAN RPC socket (host:port) — set when the node has a LAN bind.
|
||||
pub const RPC_ADDR_LAN: &str = "clawstor.rpc.lan";
|
||||
/// Tailscale RPC socket (host:port) — set when the node has a Tailscale bind.
|
||||
pub const RPC_ADDR_TAILSCALE: &str = "clawstor.rpc.tailscale";
|
||||
/// Bytes currently used in the local hot tier.
|
||||
pub const HOT_USED_BYTES: &str = "clawstor.hot.used";
|
||||
/// Configured maximum for the local hot tier.
|
||||
pub const HOT_MAX_BYTES: &str = "clawstor.hot.max";
|
||||
/// Comma-separated list of `org/repo` names this node serves warm.
|
||||
/// Feeds runner-label dynamics in later phases.
|
||||
pub const WARM_PROJECTS: &str = "clawstor.warm.projects";
|
||||
/// Unix timestamp (seconds) when this node's daemon started.
|
||||
pub const UPTIME_UNIX: &str = "clawstor.uptime.unix";
|
||||
}
|
||||
|
||||
/// Cluster identifier — every node in the same fleet must agree on this
|
||||
/// string. Chitchat drops messages that carry a mismatched cluster id.
|
||||
const DEFAULT_CLUSTER_ID: &str = "clawstor";
|
||||
|
||||
/// How often chitchat runs its gossip round. 500ms is tighter than the
|
||||
/// crate's default (1s) — fits a small fleet where fast propagation matters
|
||||
/// more than saving inter-node bytes.
|
||||
const DEFAULT_GOSSIP_INTERVAL: Duration = Duration::from_millis(500);
|
||||
|
||||
/// After a peer stops responding for this long the failure detector
|
||||
/// declares it dead. Kept short (10s) so a rebooted peer is noticed
|
||||
/// quickly by the placement policy.
|
||||
const DEFAULT_DEAD_NODE_GRACE: Duration = Duration::from_secs(10);
|
||||
|
||||
/// Grace period for tombstoned KVs before hard GC. Long enough that a
|
||||
/// briefly-disconnected node still catches up on deletions on rejoin.
|
||||
const DEFAULT_MARKED_FOR_DELETION_GRACE: Duration = Duration::from_secs(60);
|
||||
|
||||
/// A snapshot of one peer's advertised state, gathered from gossip.
|
||||
///
|
||||
/// `alive` reflects the failure detector's phi-accrual decision at read
|
||||
/// time. Fields other than `name` and `zone` may be `None` if the peer
|
||||
/// hasn't advertised them yet (early bootstrap) or has never had them
|
||||
/// (e.g. a laptop with no LAN address).
|
||||
#[derive(Debug, Clone, PartialEq)]
|
||||
pub struct PeerView {
|
||||
pub name: String,
|
||||
pub zone: String,
|
||||
pub rpc_lan: Option<SocketAddr>,
|
||||
pub rpc_tailscale: Option<SocketAddr>,
|
||||
pub hot_used_bytes: Option<u64>,
|
||||
pub hot_max_bytes: Option<u64>,
|
||||
pub warm_projects: Vec<String>,
|
||||
pub uptime_unix: Option<u64>,
|
||||
pub alive: bool,
|
||||
}
|
||||
|
||||
impl PeerView {
|
||||
/// Fraction of hot tier used, when both used + max are known.
|
||||
pub fn hot_fill_ratio(&self) -> Option<f64> {
|
||||
match (self.hot_used_bytes, self.hot_max_bytes) {
|
||||
(Some(used), Some(max)) if max > 0 => Some(used as f64 / max as f64),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Running gossip service — owns the chitchat handle. Drop shuts down
|
||||
/// the background task via [`ChitchatHandle::abort`].
|
||||
pub struct ClusterGossip {
|
||||
chitchat: Arc<Mutex<Chitchat>>,
|
||||
handle: ChitchatHandle,
|
||||
}
|
||||
|
||||
impl std::fmt::Debug for ClusterGossip {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
f.debug_struct("ClusterGossip").finish_non_exhaustive()
|
||||
}
|
||||
}
|
||||
|
||||
/// Wrap a config value in the shape chitchat expects (`(String, String)`).
|
||||
fn kv(key: &str, value: impl Into<String>) -> (String, String) {
|
||||
(key.to_string(), value.into())
|
||||
}
|
||||
|
||||
/// Turn `ClusterConfig` peers into the string-form seed list chitchat wants.
|
||||
/// Prefers LAN address (fast path) but falls back to Tailscale if that's the
|
||||
/// only reachable one — a roaming peer can still be a bootstrap seed.
|
||||
fn seed_list(cluster: &ClusterConfig) -> Vec<String> {
|
||||
cluster
|
||||
.peers
|
||||
.iter()
|
||||
.filter_map(|p| p.lan_addr.or(p.tailscale_addr))
|
||||
.map(|a| a.to_string())
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// Now, as a Unix seconds timestamp. Falls back to 0 pre-1970 (never
|
||||
/// happens in practice but keeps this pure).
|
||||
fn unix_now() -> u64 {
|
||||
SystemTime::now()
|
||||
.duration_since(UNIX_EPOCH)
|
||||
.map(|d| d.as_secs())
|
||||
.unwrap_or(0)
|
||||
}
|
||||
|
||||
impl ClusterGossip {
|
||||
/// Bootstrap chitchat: bind UDP transport, publish initial state,
|
||||
/// contact seed peers, return the running service. The returned
|
||||
/// `ClusterGossip` MUST be held for the daemon's lifetime; dropping
|
||||
/// it aborts the gossip task and the node effectively leaves the
|
||||
/// cluster (peers observe it as dead within `dead_node_grace_period`).
|
||||
pub async fn bootstrap(
|
||||
cluster: &ClusterConfig,
|
||||
local_name: impl Into<String>,
|
||||
) -> Result<Self> {
|
||||
let local_name = local_name.into();
|
||||
if local_name.is_empty() {
|
||||
bail!("cluster gossip requires a non-empty local node name");
|
||||
}
|
||||
cluster
|
||||
.validate()
|
||||
.context("validating cluster config before gossip bootstrap")?;
|
||||
|
||||
let bind_addr = cluster
|
||||
.bind_lan
|
||||
.or(cluster.bind_tailscale)
|
||||
.context("cluster has no bind address; cannot bootstrap gossip")?;
|
||||
|
||||
// ChitchatId generation must monotonically increase every restart
|
||||
// so peers can tell "same node, new incarnation" apart from "stuck".
|
||||
// The unix timestamp gives us that naturally.
|
||||
let generation = unix_now();
|
||||
let chitchat_id = ChitchatId::new(local_name.clone(), generation, bind_addr);
|
||||
|
||||
let mut initial_kvs = vec![
|
||||
kv(keys::NODE_NAME, &local_name),
|
||||
kv(keys::ZONE, &cluster.zone),
|
||||
kv(keys::UPTIME_UNIX, generation.to_string()),
|
||||
];
|
||||
if let Some(lan) = cluster.bind_lan {
|
||||
initial_kvs.push(kv(keys::RPC_ADDR_LAN, lan.to_string()));
|
||||
}
|
||||
if let Some(ts) = cluster.bind_tailscale {
|
||||
initial_kvs.push(kv(keys::RPC_ADDR_TAILSCALE, ts.to_string()));
|
||||
}
|
||||
|
||||
let config = ChitchatConfig {
|
||||
cluster_id: DEFAULT_CLUSTER_ID.to_string(),
|
||||
chitchat_id,
|
||||
gossip_interval: DEFAULT_GOSSIP_INTERVAL,
|
||||
listen_addr: bind_addr,
|
||||
seed_nodes: seed_list(cluster),
|
||||
failure_detector_config: FailureDetectorConfig {
|
||||
dead_node_grace_period: DEFAULT_DEAD_NODE_GRACE,
|
||||
..FailureDetectorConfig::default()
|
||||
},
|
||||
marked_for_deletion_grace_period: DEFAULT_MARKED_FOR_DELETION_GRACE,
|
||||
catchup_callback: None,
|
||||
extra_liveness_predicate: None,
|
||||
};
|
||||
|
||||
let handle = spawn_chitchat(config, initial_kvs, &UdpTransport)
|
||||
.await
|
||||
.context("spawning chitchat gossip service")?;
|
||||
let chitchat = handle.chitchat();
|
||||
Ok(Self { chitchat, handle })
|
||||
}
|
||||
|
||||
/// Set an arbitrary key on our own advertised state. Peers see it on
|
||||
/// the next gossip round.
|
||||
pub async fn set(&self, key: &str, value: impl Into<String>) {
|
||||
let value = value.into();
|
||||
let mut cc = self.chitchat.lock().await;
|
||||
cc.self_node_state().set(key, &value);
|
||||
}
|
||||
|
||||
/// Publish current hot-tier usage. Called periodically by the daemon.
|
||||
pub async fn set_hot_used(&self, bytes: u64) {
|
||||
self.set(keys::HOT_USED_BYTES, bytes.to_string()).await;
|
||||
}
|
||||
|
||||
/// Publish configured hot-tier maximum. Called once at bootstrap and
|
||||
/// whenever config reloads change it.
|
||||
pub async fn set_hot_max(&self, bytes: u64) {
|
||||
self.set(keys::HOT_MAX_BYTES, bytes.to_string()).await;
|
||||
}
|
||||
|
||||
/// Publish the list of warm-tier `org/repo` projects this node serves.
|
||||
/// Later phases use this to bias runner scheduling.
|
||||
pub async fn set_warm_projects<S: AsRef<str>>(&self, projects: &[S]) {
|
||||
let joined = projects
|
||||
.iter()
|
||||
.map(|s| s.as_ref())
|
||||
.collect::<Vec<_>>()
|
||||
.join(",");
|
||||
self.set(keys::WARM_PROJECTS, joined).await;
|
||||
}
|
||||
|
||||
/// This node's identity as chitchat sees it (name, generation, gossip
|
||||
/// address). Useful for logs and self-filtering during peer reads.
|
||||
pub async fn self_chitchat_id(&self) -> ChitchatId {
|
||||
self.chitchat.lock().await.self_chitchat_id().clone()
|
||||
}
|
||||
|
||||
/// All known peers other than self, with their advertised state and
|
||||
/// liveness. Includes peers currently in the grace period (dead but
|
||||
/// not yet garbage-collected).
|
||||
pub async fn peers(&self) -> Vec<PeerView> {
|
||||
let cc = self.chitchat.lock().await;
|
||||
let self_id = cc.self_chitchat_id().clone();
|
||||
let live: std::collections::HashSet<ChitchatId> = cc.live_nodes().cloned().collect();
|
||||
let mut views = Vec::new();
|
||||
for id in cc.live_nodes().chain(cc.dead_nodes()) {
|
||||
if *id == self_id {
|
||||
continue;
|
||||
}
|
||||
if let Some(state) = cc.node_state(id) {
|
||||
views.push(peer_view_from_state(id, state, live.contains(id)));
|
||||
}
|
||||
}
|
||||
views
|
||||
}
|
||||
|
||||
/// Look up one peer by advertised name. Returns `None` if we've never
|
||||
/// heard from a peer with that name.
|
||||
pub async fn peer(&self, name: &str) -> Option<PeerView> {
|
||||
self.peers().await.into_iter().find(|p| p.name == name)
|
||||
}
|
||||
|
||||
/// Live peers in a given zone.
|
||||
pub async fn peers_in_zone(&self, zone: &str) -> Vec<PeerView> {
|
||||
self.peers()
|
||||
.await
|
||||
.into_iter()
|
||||
.filter(|p| p.alive && p.zone == zone)
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// Graceful shutdown: aborts the gossip task. Peers observe this node
|
||||
/// as dead within `dead_node_grace_period`.
|
||||
pub fn shutdown(self) {
|
||||
self.handle.abort();
|
||||
}
|
||||
}
|
||||
|
||||
/// Extract a well-known key from a peer's chitchat state. Returns `None`
|
||||
/// when the key hasn't been advertised (early bootstrap, or the peer
|
||||
/// simply doesn't publish it).
|
||||
fn get_str(state: &chitchat::NodeState, key: &str) -> Option<String> {
|
||||
state.get(key).map(|s| s.to_string())
|
||||
}
|
||||
|
||||
fn get_u64(state: &chitchat::NodeState, key: &str) -> Option<u64> {
|
||||
state.get(key).and_then(|s| s.parse().ok())
|
||||
}
|
||||
|
||||
fn get_socket(state: &chitchat::NodeState, key: &str) -> Option<SocketAddr> {
|
||||
state.get(key).and_then(|s| s.parse().ok())
|
||||
}
|
||||
|
||||
fn peer_view_from_state(
|
||||
id: &ChitchatId,
|
||||
state: &chitchat::NodeState,
|
||||
alive: bool,
|
||||
) -> PeerView {
|
||||
// Fall back to the chitchat node_id when the peer hasn't published a
|
||||
// separate NODE_NAME yet (should be almost never, but keeps startup
|
||||
// races graceful).
|
||||
let name = get_str(state, keys::NODE_NAME).unwrap_or_else(|| id.node_id.to_string());
|
||||
let zone = get_str(state, keys::ZONE).unwrap_or_default();
|
||||
let warm_projects = get_str(state, keys::WARM_PROJECTS)
|
||||
.map(|s| {
|
||||
s.split(',')
|
||||
.filter_map(|p| {
|
||||
let t = p.trim();
|
||||
if t.is_empty() { None } else { Some(t.to_string()) }
|
||||
})
|
||||
.collect()
|
||||
})
|
||||
.unwrap_or_default();
|
||||
PeerView {
|
||||
name,
|
||||
zone,
|
||||
rpc_lan: get_socket(state, keys::RPC_ADDR_LAN),
|
||||
rpc_tailscale: get_socket(state, keys::RPC_ADDR_TAILSCALE),
|
||||
hot_used_bytes: get_u64(state, keys::HOT_USED_BYTES),
|
||||
hot_max_bytes: get_u64(state, keys::HOT_MAX_BYTES),
|
||||
warm_projects,
|
||||
uptime_unix: get_u64(state, keys::UPTIME_UNIX),
|
||||
alive,
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::config::PeerEntry;
|
||||
use std::sync::atomic::{AtomicU16, Ordering};
|
||||
use std::time::Instant;
|
||||
|
||||
/// Fixed port range for gossip tests, so multiple parallel tests within
|
||||
/// the same binary don't collide. 41000-41999 is above the ephemeral
|
||||
/// range on most Linux boxes (32768-60999 minus the top part) yet still
|
||||
/// out of the well-known-service ranges. Two ports per test.
|
||||
static NEXT_PORT: AtomicU16 = AtomicU16::new(41001);
|
||||
fn next_lan_port() -> u16 {
|
||||
NEXT_PORT.fetch_add(1, Ordering::Relaxed)
|
||||
}
|
||||
|
||||
fn loopback(port: u16) -> SocketAddr {
|
||||
format!("127.0.0.1:{port}").parse().unwrap()
|
||||
}
|
||||
|
||||
/// Wait until the gossip service both knows about the named peer AND
|
||||
/// the failure detector has classified it as alive. Phi-accrual needs
|
||||
/// a handful of heartbeat samples before a newly-discovered node
|
||||
/// flips to live — this wait covers that ramp-up.
|
||||
async fn wait_until_peer_alive(
|
||||
gossip: &ClusterGossip,
|
||||
name: &str,
|
||||
deadline: Duration,
|
||||
) -> bool {
|
||||
let start = Instant::now();
|
||||
loop {
|
||||
if let Some(v) = gossip.peer(name).await {
|
||||
if v.alive {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
if start.elapsed() >= deadline {
|
||||
return gossip.peer(name).await.map(|v| v.alive).unwrap_or(false);
|
||||
}
|
||||
tokio::time::sleep(Duration::from_millis(100)).await;
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn bootstrap_publishes_our_own_state() {
|
||||
let port = next_lan_port();
|
||||
let cfg = ClusterConfig {
|
||||
zone: "fabric-10g".into(),
|
||||
bind_lan: Some(loopback(port)),
|
||||
bind_tailscale: None,
|
||||
peers: vec![],
|
||||
};
|
||||
let g = ClusterGossip::bootstrap(&cfg, "solo").await.unwrap();
|
||||
let id = g.self_chitchat_id().await;
|
||||
assert_eq!(id.node_id.as_ref(), "solo");
|
||||
assert_eq!(id.gossip_advertise_addr, loopback(port));
|
||||
g.shutdown();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn bootstrap_fails_when_local_name_empty() {
|
||||
let cfg = ClusterConfig {
|
||||
zone: "fabric-10g".into(),
|
||||
bind_lan: Some(loopback(next_lan_port())),
|
||||
bind_tailscale: None,
|
||||
peers: vec![],
|
||||
};
|
||||
let err = ClusterGossip::bootstrap(&cfg, "")
|
||||
.await
|
||||
.unwrap_err()
|
||||
.to_string();
|
||||
assert!(err.contains("non-empty local node name"), "err: {err}");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn bootstrap_fails_when_no_bind_address() {
|
||||
let cfg = ClusterConfig {
|
||||
zone: "fabric-10g".into(),
|
||||
bind_lan: None,
|
||||
bind_tailscale: None,
|
||||
peers: vec![],
|
||||
};
|
||||
// ClusterConfig::validate rejects this first — that's what we want:
|
||||
// the daemon should refuse to bootstrap gossip on a malformed config.
|
||||
let err_chain = format!("{:#}", ClusterGossip::bootstrap(&cfg, "solo").await.unwrap_err());
|
||||
assert!(
|
||||
err_chain.contains("no bind address"),
|
||||
"expected 'no bind address' in error chain, got: {err_chain}"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn two_node_cluster_converges_and_shares_state() {
|
||||
let port_a = next_lan_port();
|
||||
let port_b = next_lan_port();
|
||||
let addr_a = loopback(port_a);
|
||||
let addr_b = loopback(port_b);
|
||||
|
||||
// Node A: no seeds — starts alone.
|
||||
let cfg_a = ClusterConfig {
|
||||
zone: "fabric-10g".into(),
|
||||
bind_lan: Some(addr_a),
|
||||
bind_tailscale: None,
|
||||
peers: vec![],
|
||||
};
|
||||
// Node B: uses A as seed.
|
||||
let cfg_b = ClusterConfig {
|
||||
zone: "lan-1g".into(),
|
||||
bind_lan: Some(addr_b),
|
||||
bind_tailscale: None,
|
||||
peers: vec![PeerEntry {
|
||||
name: "a".into(),
|
||||
zone: "fabric-10g".into(),
|
||||
lan_addr: Some(addr_a),
|
||||
tailscale_addr: None,
|
||||
}],
|
||||
};
|
||||
|
||||
let gossip_a = ClusterGossip::bootstrap(&cfg_a, "a").await.unwrap();
|
||||
let gossip_b = ClusterGossip::bootstrap(&cfg_b, "b").await.unwrap();
|
||||
|
||||
// Publish some state we'll verify propagated.
|
||||
gossip_a.set_hot_used(1024).await;
|
||||
gossip_a.set_hot_max(1_000_000).await;
|
||||
gossip_a
|
||||
.set_warm_projects(&["osobh/clawverse", "osobh/clawmates"])
|
||||
.await;
|
||||
|
||||
// Wait for gossip convergence AND phi-accrual to classify both
|
||||
// peers as live. Allow a generous 10s since phi-accrual needs a
|
||||
// handful of heartbeat samples (500ms gossip interval → typically
|
||||
// 2-4s to flip to alive on first sight).
|
||||
let a_sees_b_alive =
|
||||
wait_until_peer_alive(&gossip_a, "b", Duration::from_secs(10)).await;
|
||||
assert!(a_sees_b_alive, "A should see B alive within 10s");
|
||||
let b_sees_a_alive =
|
||||
wait_until_peer_alive(&gossip_b, "a", Duration::from_secs(10)).await;
|
||||
assert!(b_sees_a_alive, "B should see A alive within 10s");
|
||||
|
||||
// Verify the state B sees for A matches what A published.
|
||||
let b_view_of_a = gossip_b.peer("a").await.expect("B has A");
|
||||
assert_eq!(b_view_of_a.zone, "fabric-10g");
|
||||
assert_eq!(b_view_of_a.rpc_lan, Some(addr_a));
|
||||
assert!(b_view_of_a.alive);
|
||||
assert_eq!(b_view_of_a.hot_used_bytes, Some(1024));
|
||||
assert_eq!(b_view_of_a.hot_max_bytes, Some(1_000_000));
|
||||
assert_eq!(
|
||||
b_view_of_a.warm_projects,
|
||||
vec![
|
||||
"osobh/clawverse".to_string(),
|
||||
"osobh/clawmates".to_string(),
|
||||
]
|
||||
);
|
||||
assert_eq!(b_view_of_a.hot_fill_ratio(), Some(1024.0 / 1_000_000.0));
|
||||
|
||||
// And what A sees for B — zone should be lan-1g.
|
||||
let a_view_of_b = gossip_a.peer("b").await.expect("A has B");
|
||||
assert_eq!(a_view_of_b.zone, "lan-1g");
|
||||
assert_eq!(a_view_of_b.rpc_lan, Some(addr_b));
|
||||
|
||||
// Zone filtering.
|
||||
let a_fabric = gossip_a.peers_in_zone("fabric-10g").await;
|
||||
assert!(a_fabric.is_empty(), "A alone in its zone from its view");
|
||||
let a_lan_1g = gossip_a.peers_in_zone("lan-1g").await;
|
||||
assert_eq!(a_lan_1g.len(), 1);
|
||||
assert_eq!(a_lan_1g[0].name, "b");
|
||||
|
||||
gossip_a.shutdown();
|
||||
gossip_b.shutdown();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn peers_excludes_self() {
|
||||
let port = next_lan_port();
|
||||
let cfg = ClusterConfig {
|
||||
zone: "fabric-10g".into(),
|
||||
bind_lan: Some(loopback(port)),
|
||||
bind_tailscale: None,
|
||||
peers: vec![],
|
||||
};
|
||||
let g = ClusterGossip::bootstrap(&cfg, "solo").await.unwrap();
|
||||
// Solo cluster — peers() must never include self.
|
||||
let peers = g.peers().await;
|
||||
assert!(peers.is_empty(), "solo cluster reports zero peers");
|
||||
g.shutdown();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn peer_view_hot_fill_ratio_handles_missing_or_zero() {
|
||||
let base = PeerView {
|
||||
name: "x".into(),
|
||||
zone: "z".into(),
|
||||
rpc_lan: None,
|
||||
rpc_tailscale: None,
|
||||
hot_used_bytes: None,
|
||||
hot_max_bytes: Some(1000),
|
||||
warm_projects: vec![],
|
||||
uptime_unix: None,
|
||||
alive: true,
|
||||
};
|
||||
assert_eq!(base.hot_fill_ratio(), None, "no used → None");
|
||||
|
||||
let with_zero_max = PeerView {
|
||||
hot_used_bytes: Some(500),
|
||||
hot_max_bytes: Some(0),
|
||||
..base.clone()
|
||||
};
|
||||
assert_eq!(
|
||||
with_zero_max.hot_fill_ratio(),
|
||||
None,
|
||||
"max=0 → None (no div by zero)"
|
||||
);
|
||||
|
||||
let half_full = PeerView {
|
||||
hot_used_bytes: Some(500),
|
||||
hot_max_bytes: Some(1000),
|
||||
..base
|
||||
};
|
||||
assert_eq!(half_full.hot_fill_ratio(), Some(0.5));
|
||||
}
|
||||
|
||||
}
|
||||
@@ -74,6 +74,15 @@ enum Cmd {
|
||||
/// Peer name as listed in `[[cluster.peers]]` in the config
|
||||
peer: String,
|
||||
},
|
||||
/// Bootstrap gossip, wait for convergence, print the peer membership table
|
||||
ClusterStatus {
|
||||
/// This node's name as advertised to peers.
|
||||
#[arg(long)]
|
||||
name: String,
|
||||
/// How long to wait for convergence before printing.
|
||||
#[arg(long, default_value = "3")]
|
||||
wait_secs: u64,
|
||||
},
|
||||
}
|
||||
|
||||
#[tokio::main]
|
||||
@@ -109,10 +118,57 @@ async fn main() -> Result<()> {
|
||||
Cmd::Pin { project } => cmd_set_pin(&manifest_path, &project, true)?,
|
||||
Cmd::Unpin { project } => cmd_set_pin(&manifest_path, &project, false)?,
|
||||
Cmd::ClusterProbe { peer } => cmd_cluster_probe(&cfg, &peer).await?,
|
||||
Cmd::ClusterStatus { name, wait_secs } => cmd_cluster_status(&cfg, &name, wait_secs).await?,
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
// ── cluster status ───────────────────────────────────────────────────────────
|
||||
|
||||
async fn cmd_cluster_status(cfg: &Config, name: &str, wait_secs: u64) -> Result<()> {
|
||||
let cluster_cfg = cfg
|
||||
.cluster
|
||||
.as_ref()
|
||||
.context("no [cluster] section in config; cannot start gossip")?;
|
||||
let gossip = cluster::gossip::ClusterGossip::bootstrap(cluster_cfg, name).await?;
|
||||
println!("gossip bootstrapped as {}; waiting {}s for convergence...", name, wait_secs);
|
||||
tokio::time::sleep(std::time::Duration::from_secs(wait_secs)).await;
|
||||
|
||||
let mut peers = gossip.peers().await;
|
||||
peers.sort_by(|a, b| a.name.cmp(&b.name));
|
||||
|
||||
println!();
|
||||
println!(
|
||||
"{:<20} {:<14} {:<8} {:<22} {:<22} {:<20}",
|
||||
"PEER", "ZONE", "STATE", "RPC LAN", "RPC TAILSCALE", "HOT USED / MAX"
|
||||
);
|
||||
println!("{}", "-".repeat(110));
|
||||
if peers.is_empty() {
|
||||
println!("(no peers yet — either alone in the fleet or gossip still converging)");
|
||||
} else {
|
||||
for p in &peers {
|
||||
let state = if p.alive { "alive" } else { "dead" };
|
||||
let hot = match (p.hot_used_bytes, p.hot_max_bytes) {
|
||||
(Some(u), Some(m)) => format!("{u} / {m}"),
|
||||
(Some(u), None) => format!("{u} / -"),
|
||||
_ => "-".into(),
|
||||
};
|
||||
println!(
|
||||
"{:<20} {:<14} {:<8} {:<22} {:<22} {:<20}",
|
||||
p.name,
|
||||
p.zone,
|
||||
state,
|
||||
p.rpc_lan.map(|a| a.to_string()).unwrap_or_else(|| "-".into()),
|
||||
p.rpc_tailscale.map(|a| a.to_string()).unwrap_or_else(|| "-".into()),
|
||||
hot,
|
||||
);
|
||||
}
|
||||
}
|
||||
println!();
|
||||
gossip.shutdown();
|
||||
Ok(())
|
||||
}
|
||||
|
||||
// ── cluster probe ────────────────────────────────────────────────────────────
|
||||
|
||||
async fn cmd_cluster_probe(cfg: &Config, peer_name: &str) -> Result<()> {
|
||||
|
||||
Reference in New Issue
Block a user