Phase 1a: cluster module + LAN-first peer probe
First cut of the v2 distributed FS. Captures the architecture design in ARCHITECTURE-v2.md and lands the smallest useful new capability: probing cluster peers with a LAN-first policy so subsequent transport + gossip layers (Phase 1b, 1c) can build on a real routing decision. New: - ARCHITECTURE-v2.md: zones (fabric-10g/lan-1g/roaming), tier lifecycle (hot/warm/cold), fingerprint-keyed build cache design, smart-clean policy, phase plan, explicit non-goals. - claw-store/src/cluster.rs: RouteKind, RouteWinner, LanFirstProbe. LAN 200ms timeout, Tailscale 500ms fallback. 8 tests use real TCP listeners on 127.0.0.1 (no mocks); cover happy path, fall-through, both-fail, single-address, and elapsed reporting. - claw-store/src/config.rs: ClusterConfig + PeerEntry with validation (bind-address presence, no duplicate peer names, per-peer reachable address required). Optional at top level so pre-v2 configs still load unchanged. 6 new tests. - claw-store/src/main.rs: `claw-store cluster-probe <peer>` CLI subcommand that reads config, resolves the peer, probes, prints the winning route + elapsed time. All 16 new tests pass. Existing 45 pass. Sole failure (hot::tests::test_project_target_size_bytes) is a pre-existing macOS-only issue with `du -sb`; Linux CI unaffected. Follow-on Phase 1 cuts (subsequent sessions): - 1b: chitchat SWIM gossip for live membership state - 1c: quinn QUIC transport with fleet-CA mTLS - 1d: `claw-store cluster status` — live membership view Every file well under the 1300-line ceiling (cluster.rs 268, config.rs 428, main.rs 450).
This commit is contained in:
Omar Sobh
+278
-3
@@ -1,5 +1,7 @@
|
||||
use anyhow::{Context, Result};
|
||||
use anyhow::{bail, Context, Result};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use std::collections::HashSet;
|
||||
use std::net::SocketAddr;
|
||||
use std::path::{Path, PathBuf};
|
||||
|
||||
#[derive(Debug, Clone, Deserialize, Serialize, PartialEq)]
|
||||
@@ -51,6 +53,88 @@ pub struct PeerConfig {
|
||||
pub user: String,
|
||||
}
|
||||
|
||||
/// One peer node the cluster knows about. Every peer carries up to two reachable
|
||||
/// endpoints: a LAN socket (fast path, tried first) and a Tailscale socket
|
||||
/// (fallback, always reachable when the tailnet is up). At least one must be set.
|
||||
#[derive(Debug, Clone, Deserialize, Serialize, PartialEq)]
|
||||
pub struct PeerEntry {
|
||||
/// Stable human-facing name (e.g. `"architect"`). Must be unique in the config.
|
||||
pub name: String,
|
||||
/// Zone tag — one of `"fabric-10g"`, `"lan-1g"`, `"roaming"`, or a custom label.
|
||||
pub zone: String,
|
||||
/// LAN socket. Omit when this peer has no LAN presence from our perspective
|
||||
/// (e.g. a roaming laptop reachable only via Tailscale).
|
||||
#[serde(default)]
|
||||
pub lan_addr: Option<SocketAddr>,
|
||||
/// Tailscale socket. Omit only if the peer is LAN-only.
|
||||
#[serde(default)]
|
||||
pub tailscale_addr: Option<SocketAddr>,
|
||||
}
|
||||
|
||||
impl PeerEntry {
|
||||
/// Sanity check: at least one address must be reachable.
|
||||
pub fn validate(&self) -> Result<()> {
|
||||
if self.name.is_empty() {
|
||||
bail!("peer entry with empty name");
|
||||
}
|
||||
if self.zone.is_empty() {
|
||||
bail!("peer {} has empty zone", self.name);
|
||||
}
|
||||
if self.lan_addr.is_none() && self.tailscale_addr.is_none() {
|
||||
bail!(
|
||||
"peer {} has no reachable address (both lan_addr and tailscale_addr unset)",
|
||||
self.name
|
||||
);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
/// Cluster membership configuration. Optional at the top level so existing
|
||||
/// single-node deployments (pre-v2) keep loading. Once present, describes the
|
||||
/// local node's zone + bind addresses, and enumerates known peers.
|
||||
#[derive(Debug, Clone, Deserialize, Serialize, PartialEq)]
|
||||
pub struct ClusterConfig {
|
||||
/// This node's zone tag.
|
||||
pub zone: String,
|
||||
/// LAN listen socket (typically `0.0.0.0:7701`). Omit on roaming nodes.
|
||||
#[serde(default)]
|
||||
pub bind_lan: Option<SocketAddr>,
|
||||
/// Tailscale listen socket (Tailscale IP + port). Omit on strictly-LAN nodes.
|
||||
#[serde(default)]
|
||||
pub bind_tailscale: Option<SocketAddr>,
|
||||
/// Static seed list of peers. Runtime membership (Phase 1b) will extend this
|
||||
/// via gossip; the config list bootstraps discovery.
|
||||
#[serde(default)]
|
||||
pub peers: Vec<PeerEntry>,
|
||||
}
|
||||
|
||||
impl ClusterConfig {
|
||||
/// Sanity check the cluster config: at least one bind address, no duplicate
|
||||
/// peer names, every peer has at least one address.
|
||||
pub fn validate(&self) -> Result<()> {
|
||||
if self.zone.is_empty() {
|
||||
bail!("cluster.zone is empty");
|
||||
}
|
||||
if self.bind_lan.is_none() && self.bind_tailscale.is_none() {
|
||||
bail!("cluster has no bind address (both bind_lan and bind_tailscale unset)");
|
||||
}
|
||||
let mut names = HashSet::new();
|
||||
for peer in &self.peers {
|
||||
peer.validate()?;
|
||||
if !names.insert(peer.name.as_str()) {
|
||||
bail!("duplicate peer name in cluster.peers: {}", peer.name);
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Look up a peer by name.
|
||||
pub fn peer(&self, name: &str) -> Option<&PeerEntry> {
|
||||
self.peers.iter().find(|p| p.name == name)
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Deserialize, Serialize)]
|
||||
pub struct Config {
|
||||
pub node: NodeConfig,
|
||||
@@ -59,6 +143,10 @@ pub struct Config {
|
||||
pub cold: Option<ColdConfig>,
|
||||
pub replication: Option<ReplicationConfig>,
|
||||
pub peer: Option<PeerConfig>,
|
||||
/// v2 cluster membership (peers, zones, bind addresses). Optional so pre-v2
|
||||
/// configs still load; once populated, `Config::load` calls `validate()`.
|
||||
#[serde(default)]
|
||||
pub cluster: Option<ClusterConfig>,
|
||||
/// Optional Bearer token required on all HTTP POST endpoints.
|
||||
/// Set to a long random string, e.g. `openssl rand -hex 32`.
|
||||
/// If absent, POST endpoints are unauthenticated (internal-network use only).
|
||||
@@ -70,8 +158,14 @@ impl Config {
|
||||
pub fn load(path: &Path) -> Result<Self> {
|
||||
let content = std::fs::read_to_string(path)
|
||||
.with_context(|| format!("reading config at {}", path.display()))?;
|
||||
toml::from_str(&content)
|
||||
.with_context(|| format!("parsing config at {}", path.display()))
|
||||
let cfg: Config = toml::from_str(&content)
|
||||
.with_context(|| format!("parsing config at {}", path.display()))?;
|
||||
if let Some(cluster) = cfg.cluster.as_ref() {
|
||||
cluster
|
||||
.validate()
|
||||
.with_context(|| format!("validating cluster config in {}", path.display()))?;
|
||||
}
|
||||
Ok(cfg)
|
||||
}
|
||||
|
||||
pub fn default_path() -> PathBuf {
|
||||
@@ -119,6 +213,187 @@ peer_user = "osobh"
|
||||
assert_eq!(cfg.cold.unwrap().retain_weeks, 12);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_pre_v2_config_still_loads_without_cluster_section() {
|
||||
// Backwards compatibility: any config that worked before v2 must still
|
||||
// parse cleanly with `cluster` absent.
|
||||
let toml = r#"
|
||||
[node]
|
||||
name = "tank"
|
||||
role = "secondary"
|
||||
|
||||
[hot]
|
||||
path = "/hot/targets"
|
||||
max_gb = 200
|
||||
stale_hours = 48
|
||||
|
||||
[warm]
|
||||
projects_path = "/slab/projects"
|
||||
zfs_dataset = "slab/projects"
|
||||
snapshot_retain_hours = 24
|
||||
snapshot_retain_days = 7
|
||||
snapshot_retain_weeks = 4
|
||||
"#;
|
||||
let cfg: Config = toml::from_str(toml).unwrap();
|
||||
assert!(cfg.cluster.is_none(), "cluster must be optional");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_cluster_config_parses_with_peers() {
|
||||
let toml = r#"
|
||||
[node]
|
||||
name = "tank"
|
||||
role = "secondary"
|
||||
|
||||
[hot]
|
||||
path = "/hot/targets"
|
||||
max_gb = 200
|
||||
stale_hours = 48
|
||||
|
||||
[warm]
|
||||
projects_path = "/slab/projects"
|
||||
zfs_dataset = "slab/projects"
|
||||
snapshot_retain_hours = 24
|
||||
snapshot_retain_days = 7
|
||||
snapshot_retain_weeks = 4
|
||||
|
||||
[cluster]
|
||||
zone = "fabric-10g"
|
||||
bind_lan = "10.0.0.14:7701"
|
||||
bind_tailscale = "100.64.1.2:7701"
|
||||
|
||||
[[cluster.peers]]
|
||||
name = "architect"
|
||||
zone = "fabric-10g"
|
||||
lan_addr = "10.0.0.13:7701"
|
||||
tailscale_addr = "100.64.1.3:7701"
|
||||
|
||||
[[cluster.peers]]
|
||||
name = "morpheus"
|
||||
zone = "lan-1g"
|
||||
lan_addr = "192.168.1.50:7701"
|
||||
tailscale_addr = "100.64.1.4:7701"
|
||||
|
||||
[[cluster.peers]]
|
||||
name = "laptop"
|
||||
zone = "roaming"
|
||||
tailscale_addr = "100.64.1.5:7701"
|
||||
"#;
|
||||
let cfg: Config = toml::from_str(toml).unwrap();
|
||||
let cluster = cfg.cluster.expect("cluster section present");
|
||||
assert_eq!(cluster.zone, "fabric-10g");
|
||||
assert_eq!(cluster.peers.len(), 3);
|
||||
|
||||
let architect = cluster.peer("architect").expect("architect peer");
|
||||
assert_eq!(architect.zone, "fabric-10g");
|
||||
assert_eq!(
|
||||
architect.lan_addr.unwrap(),
|
||||
"10.0.0.13:7701".parse::<SocketAddr>().unwrap()
|
||||
);
|
||||
|
||||
let laptop = cluster.peer("laptop").expect("laptop peer");
|
||||
assert!(
|
||||
laptop.lan_addr.is_none(),
|
||||
"roaming peer has no LAN address"
|
||||
);
|
||||
assert!(laptop.tailscale_addr.is_some());
|
||||
|
||||
cluster.validate().expect("valid cluster config");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_peer_without_addresses_fails_validation() {
|
||||
let peer = PeerEntry {
|
||||
name: "ghost".into(),
|
||||
zone: "fabric-10g".into(),
|
||||
lan_addr: None,
|
||||
tailscale_addr: None,
|
||||
};
|
||||
let err = peer.validate().unwrap_err().to_string();
|
||||
assert!(
|
||||
err.contains("ghost") && err.contains("no reachable address"),
|
||||
"unexpected error: {err}"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_cluster_without_bind_addresses_fails_validation() {
|
||||
let cluster = ClusterConfig {
|
||||
zone: "fabric-10g".into(),
|
||||
bind_lan: None,
|
||||
bind_tailscale: None,
|
||||
peers: vec![],
|
||||
};
|
||||
let err = cluster.validate().unwrap_err().to_string();
|
||||
assert!(
|
||||
err.contains("no bind address"),
|
||||
"unexpected error: {err}"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_cluster_duplicate_peer_names_fails_validation() {
|
||||
let cluster = ClusterConfig {
|
||||
zone: "fabric-10g".into(),
|
||||
bind_lan: Some("10.0.0.14:7701".parse().unwrap()),
|
||||
bind_tailscale: None,
|
||||
peers: vec![
|
||||
PeerEntry {
|
||||
name: "architect".into(),
|
||||
zone: "fabric-10g".into(),
|
||||
lan_addr: Some("10.0.0.13:7701".parse().unwrap()),
|
||||
tailscale_addr: None,
|
||||
},
|
||||
PeerEntry {
|
||||
name: "architect".into(),
|
||||
zone: "fabric-10g".into(),
|
||||
lan_addr: Some("10.0.0.14:7701".parse().unwrap()),
|
||||
tailscale_addr: None,
|
||||
},
|
||||
],
|
||||
};
|
||||
let err = cluster.validate().unwrap_err().to_string();
|
||||
assert!(
|
||||
err.contains("duplicate peer name"),
|
||||
"unexpected error: {err}"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_cluster_config_load_invokes_validate() {
|
||||
// Same shape as a real config file but written to a tempfile so the
|
||||
// full path — read, parse, validate — is exercised.
|
||||
let toml = r#"
|
||||
[node]
|
||||
name = "tank"
|
||||
role = "secondary"
|
||||
|
||||
[hot]
|
||||
path = "/hot/targets"
|
||||
max_gb = 200
|
||||
stale_hours = 48
|
||||
|
||||
[warm]
|
||||
projects_path = "/slab/projects"
|
||||
zfs_dataset = "slab/projects"
|
||||
snapshot_retain_hours = 24
|
||||
snapshot_retain_days = 7
|
||||
snapshot_retain_weeks = 4
|
||||
|
||||
[cluster]
|
||||
zone = "fabric-10g"
|
||||
|
||||
[[cluster.peers]]
|
||||
name = "architect"
|
||||
zone = "fabric-10g"
|
||||
"#;
|
||||
let tmp = tempfile::NamedTempFile::new().unwrap();
|
||||
std::fs::write(tmp.path(), toml).unwrap();
|
||||
let err = Config::load(tmp.path()).unwrap_err().to_string();
|
||||
// Fails because cluster has no bind address AND peer has no reachable address.
|
||||
assert!(err.contains("validating cluster config"), "err: {err}");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_load_secondary_config() {
|
||||
let toml = r#"
|
||||
|
||||
Reference in New Issue
Block a user