crashlabs/huskies
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 26 Wiki Activity

huskies: merge 766

Browse Source
This commit is contained in:
dave
2026-04-28 08:54:44 +00:00
parent 0d14fffe1c
commit 38e828979c
6 changed files with 107 additions and 584 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/src/service/gateway/io.rs
-23
View File
@@ -5,7 +5,6 @@
//! spawning the Matrix bot task, and the notification poller background task.
use super::config::{GatewayConfig, ProjectEntry};
use super::registration::JoinedAgent;
pub use reqwest::Client;
use serde_json::{Value, json};
use std::collections::{BTreeMap, HashMap};
@@ -20,16 +19,6 @@ pub fn load_config(path: &Path) -> Result<GatewayConfig, String> {
toml::from_str(&contents).map_err(|e| format!("invalid projects.toml: {e}"))
}
/// Load persisted agents from `<config_dir>/gateway_agents.json`.
/// Returns an empty list if the file does not exist or cannot be parsed.
pub fn load_agents(config_dir: &Path) -> Vec<JoinedAgent> {
let path = config_dir.join("gateway_agents.json");
match std::fs::read(&path) {
Ok(data) => serde_json::from_slice(&data).unwrap_or_default(),
Err(_) => Vec::new(),
}
}
/// Persist the current projects map to `<config_dir>/projects.toml`.
/// Silently ignores write errors or skips when `config_dir` is empty.
pub async fn save_config(projects: &BTreeMap<String, ProjectEntry>, config_dir: &Path) {
@@ -45,18 +34,6 @@ pub async fn save_config(projects: &BTreeMap<String, ProjectEntry>, config_dir:
}
}
/// Persist the current agent list to `<config_dir>/gateway_agents.json`.
/// Silently ignores write errors.
pub async fn save_agents(agents: &[JoinedAgent], config_dir: &Path) {
if config_dir == Path::new("") {
return;
}
let path = config_dir.join("gateway_agents.json");
if let Ok(data) = serde_json::to_vec_pretty(agents) {
let _ = tokio::fs::write(&path, data).await;
}
}
// ── Bot config I/O ──────────────────────────────────────────────────────────
/// Read the current raw bot.toml as key/value pairs for the configuration UI.
server/src/service/gateway/mod.rs
+106 -75
View File
@@ -4,7 +4,6 @@
//! - `mod.rs` (this file) — public API, typed [`Error`], orchestration, `GatewayState`
//! - `io.rs` — the ONLY place that performs side effects (filesystem, network, process spawn)
//! - `config.rs` — pure config types and validation
//! - `registration.rs` — pure agent registration logic
//! - `aggregation.rs` — pure cross-project pipeline formatting
//! - `polling.rs` — pure notification event formatting
@@ -12,7 +11,6 @@ pub mod aggregation;
pub mod config;
pub(crate) mod io;
pub mod polling;
pub mod registration;
pub use aggregation::format_aggregate_status_compact;
pub use config::{GatewayConfig, ProjectEntry};
@@ -20,7 +18,6 @@ pub use io::{
fetch_all_project_pipeline_statuses, spawn_gateway_broadcaster_forwarder,
spawn_gateway_notification_poller,
};
pub use registration::JoinedAgent;
use io::Client;
use std::collections::{BTreeMap, HashMap};
@@ -29,6 +26,8 @@ use std::sync::Arc;
use tokio::sync::Mutex as TokioMutex;
use tokio::sync::RwLock;
pub use crate::crdt_state::NodePresenceView;
// ── Status event broadcaster ────────────────────────────────────────────────
/// Capacity of the gateway status event broadcast channel.
@@ -101,8 +100,6 @@ pub struct GatewayState {
pub active_project: Arc<RwLock<String>>,
/// HTTP client for proxying requests to project containers.
pub client: Client,
/// Build agents that have joined this gateway.
pub joined_agents: Arc<RwLock<Vec<JoinedAgent>>>,
/// One-time join tokens that have been issued but not yet consumed.
pub(crate) pending_tokens: Arc<RwLock<HashMap<String, PendingToken>>>,
/// Directory containing `projects.toml` and the `.huskies/` subfolder.
@@ -121,20 +118,18 @@ impl GatewayState {
/// Create a new gateway state from a config and config directory.
///
/// The first project in the config becomes the active project by default.
/// Previously registered agents are loaded from `gateway_agents.json`.
/// Agent registrations are stored in the CRDT nodes collection.
pub fn new(
gateway_config: GatewayConfig,
config_dir: PathBuf,
port: u16,
) -> Result<Self, String> {
let first = config::validate_config(&gateway_config)?;
let agents = io::load_agents(&config_dir);
let (event_tx, _) = tokio::sync::broadcast::channel(EVENT_CHANNEL_CAPACITY);
Ok(Self {
projects: Arc::new(RwLock::new(gateway_config.projects)),
active_project: Arc::new(RwLock::new(first)),
client: Client::new(),
joined_agents: Arc::new(RwLock::new(agents)),
pending_tokens: Arc::new(RwLock::new(HashMap::new())),
config_dir,
port,
@@ -187,82 +182,118 @@ pub async fn generate_join_token(state: &GatewayState) -> String {
token
}
/// Register a build agent with a join token.
/// Register a new build agent using a one-time join token.
///
/// Validates and consumes the token, then writes the agent's node presence
/// and metadata to the CRDT collection. Returns the newly-created node view.
#[allow(dead_code)]
pub async fn register_agent(
state: &GatewayState,
token: &str,
label: String,
address: String,
) -> Result<JoinedAgent, Error> {
// Validate and consume the token.
let mut tokens = state.pending_tokens.write().await;
if !tokens.contains_key(token) {
return Err(Error::DuplicateToken(
"invalid or already-used join token".into(),
));
) -> Result<NodePresenceView, Error> {
{
let mut tokens = state.pending_tokens.write().await;
if !tokens.contains_key(token) {
return Err(Error::InvalidAgent(
"invalid or already-used join token".into(),
));
}
tokens.remove(token);
}
tokens.remove(token);
drop(tokens);
let node_id = uuid::Uuid::new_v4().to_string();
let now = chrono::Utc::now().timestamp() as f64;
let agent = registration::create_agent(uuid::Uuid::new_v4().to_string(), label, address, now);
let now_ms = chrono::Utc::now().timestamp_millis() as f64;
crate::crdt_state::write_node_presence(&node_id, &address, now, true);
crate::crdt_state::write_node_metadata(&node_id, &label, None, now_ms);
crate::slog!(
"[gateway] Agent '{}' registered (id={})",
agent.label,
agent.id
"[gateway] Registered agent '{label}' node_id={:.12}…",
&node_id
);
{
let mut agents = state.joined_agents.write().await;
agents.push(agent.clone());
io::save_agents(&agents, &state.config_dir).await;
}
Ok(agent)
crate::crdt_state::read_all_node_presence()
.unwrap_or_default()
.into_iter()
.find(|n| n.node_id == node_id)
.ok_or_else(|| Error::Upstream("node write did not persist".into()))
}
/// Remove a registered agent by ID. Returns `true` if found and removed.
pub async fn remove_agent(state: &GatewayState, id: &str) -> bool {
let mut agents = state.joined_agents.write().await;
let removed = registration::remove_agent(&mut agents, id);
if removed {
io::save_agents(&agents, &state.config_dir).await;
crate::slog!("[gateway] Removed agent id={id}");
}
removed
/// Tombstone a registered agent in the CRDT (set `alive = false`).
///
/// Returns `true` if the node was found and tombstoned.
#[allow(dead_code)]
pub fn remove_agent(node_id: &str) -> bool {
let nodes = crate::crdt_state::read_all_node_presence().unwrap_or_default();
let Some(node) = nodes.iter().find(|n| n.node_id == node_id) else {
return false;
};
let now = chrono::Utc::now().timestamp() as f64;
crate::crdt_state::write_node_presence(node_id, &node.address, now, false);
true
}
/// Assign or unassign an agent to a project.
/// Assign (or unassign) an agent to a project in the CRDT.
///
/// Validates that the project exists in the gateway config (when assigning),
/// then writes the updated `assigned_project` field to the CRDT.
#[allow(dead_code)]
pub async fn assign_agent(
state: &GatewayState,
id: &str,
node_id: &str,
project: Option<String>,
) -> Result<JoinedAgent, Error> {
let project_clean = project.and_then(|p| if p.is_empty() { None } else { Some(p) });
let updated = {
) -> Result<NodePresenceView, Error> {
if let Some(ref p) = project {
let projects = state.projects.read().await;
let mut agents = state.joined_agents.write().await;
registration::assign_agent(&mut agents, id, project_clean, &projects)?
};
if !projects.contains_key(p.as_str()) {
return Err(Error::ProjectNotFound(format!("unknown project '{p}'")));
}
}
crate::slog!(
"[gateway] Agent '{}' (id={}) assigned to {:?}",
updated.label,
updated.id,
updated.assigned_project
let nodes = crate::crdt_state::read_all_node_presence().unwrap_or_default();
let node = nodes
.iter()
.find(|n| n.node_id == node_id)
.ok_or_else(|| Error::InvalidAgent(format!("agent not found: {node_id}")))?;
let now_ms = chrono::Utc::now().timestamp_millis() as f64;
crate::crdt_state::write_node_metadata(
node_id,
node.label.as_deref().unwrap_or(""),
project.as_deref(),
now_ms,
);
let agents = state.joined_agents.read().await.clone();
io::save_agents(&agents, &state.config_dir).await;
Ok(updated)
crate::crdt_state::read_all_node_presence()
.unwrap_or_default()
.into_iter()
.find(|n| n.node_id == node_id)
.ok_or_else(|| Error::Upstream("node write did not persist".into()))
}
/// Update an agent's heartbeat. Returns `true` if found.
pub async fn heartbeat_agent(state: &GatewayState, id: &str) -> bool {
/// Update an agent's heartbeat via CRDT. Returns `true` if the node was found.
#[allow(dead_code)]
pub fn heartbeat_agent(id: &str) -> bool {
let now = chrono::Utc::now().timestamp() as f64;
let mut agents = state.joined_agents.write().await;
registration::heartbeat(&mut agents, id, now)
let nodes = crate::crdt_state::read_all_node_presence().unwrap_or_default();
let Some(node) = nodes.iter().find(|n| n.node_id == id) else {
return false;
};
crate::crdt_state::write_node_presence(id, &node.address, now, node.alive);
true
}
/// List all registered build agents from the CRDT nodes collection.
#[allow(dead_code)]
pub fn list_agents() -> Vec<NodePresenceView> {
crate::crdt_state::read_all_node_presence()
.unwrap_or_default()
.into_iter()
.filter(|n| n.alive)
.collect()
}
/// Add a new project to the gateway config.
@@ -561,16 +592,18 @@ mod tests {
}
#[tokio::test]
async fn generate_and_register_agent() {
async fn register_agent_consumes_token_and_writes_crdt() {
crate::crdt_state::init_for_test();
let config = make_config(&[("test", "http://test:3001")]);
let state = GatewayState::new(config, PathBuf::new(), 3000).unwrap();
let token = generate_join_token(&state).await;
let agent = register_agent(&state, &token, "test-agent".into(), "ws://a".into())
let node = register_agent(&state, &token, "test-agent".into(), "ws://a".into())
.await
.unwrap();
assert_eq!(agent.label, "test-agent");
assert_eq!(node.label.as_deref(), Some("test-agent"));
assert!(state.pending_tokens.read().await.is_empty());
assert_eq!(state.joined_agents.read().await.len(), 1);
let agents = list_agents();
assert!(agents.iter().any(|n| n.node_id == node.node_id));
}
#[tokio::test]
@@ -582,31 +615,29 @@ mod tests {
}
#[tokio::test]
async fn remove_agent_success() {
async fn remove_agent_tombstones_crdt_node() {
crate::crdt_state::init_for_test();
let config = make_config(&[("test", "http://test:3001")]);
let state = GatewayState::new(config, PathBuf::new(), 3000).unwrap();
let token = generate_join_token(&state).await;
let agent = register_agent(&state, &token, "a".into(), "ws://a".into())
let node = register_agent(&state, &token, "a".into(), "ws://a".into())
.await
.unwrap();
assert!(remove_agent(&state, &agent.id).await);
assert!(state.joined_agents.read().await.is_empty());
assert!(remove_agent(&node.node_id));
let alive = list_agents();
assert!(!alive.iter().any(|n| n.node_id == node.node_id));
}
#[tokio::test]
async fn heartbeat_agent_updates_timestamp() {
async fn heartbeat_agent_returns_true_for_known_node() {
crate::crdt_state::init_for_test();
let config = make_config(&[("test", "http://test:3001")]);
let state = GatewayState::new(config, PathBuf::new(), 3000).unwrap();
let token = generate_join_token(&state).await;
let agent = register_agent(&state, &token, "a".into(), "ws://a".into())
let node = register_agent(&state, &token, "a".into(), "ws://a".into())
.await
.unwrap();
let old_ts = agent.last_seen;
// Small sleep to ensure timestamp differs.
tokio::time::sleep(std::time::Duration::from_millis(10)).await;
assert!(heartbeat_agent(&state, &agent.id).await);
let agents = state.joined_agents.read().await;
assert!(agents[0].last_seen >= old_ts);
assert!(heartbeat_agent(&node.node_id));
}
#[tokio::test]
server/src/service/gateway/registration.rs
-165
View File
@@ -1,165 +0,0 @@
//! Gateway agent registration — pure logic for managing build agents.
//!
//! Contains `JoinedAgent` and functions that validate and manipulate agent
//! state in memory. All persistence (disk I/O) lives in `io.rs`.
use serde::{Deserialize, Serialize};
use std::collections::BTreeMap;
use super::config::ProjectEntry;
/// A build agent that has registered with this gateway.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct JoinedAgent {
/// Unique ID assigned by the gateway on registration.
pub id: String,
/// Human-readable label provided by the agent (e.g. `build-agent-abc123`).
pub label: String,
/// The agent's CRDT-sync WebSocket address (e.g. `ws://host:3001/crdt-sync`).
pub address: String,
/// Unix timestamp when the agent registered.
pub registered_at: f64,
/// Unix timestamp of the last heartbeat from this agent.
#[serde(default)]
pub last_seen: f64,
/// Project this agent is assigned to, if any.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub assigned_project: Option<String>,
}
/// Create a new `JoinedAgent` from registration data.
pub fn create_agent(id: String, label: String, address: String, now: f64) -> JoinedAgent {
JoinedAgent {
id,
label,
address,
registered_at: now,
last_seen: now,
assigned_project: None,
}
}
/// Remove an agent by ID from the list. Returns `true` if found and removed.
pub fn remove_agent(agents: &mut Vec<JoinedAgent>, id: &str) -> bool {
let before = agents.len();
agents.retain(|a| a.id != id);
agents.len() < before
}
/// Assign (or unassign) an agent to a project.
///
/// Returns the updated agent on success, or an error if the agent or project
/// is not found.
pub fn assign_agent(
agents: &mut [JoinedAgent],
id: &str,
project: Option<String>,
projects: &BTreeMap<String, ProjectEntry>,
) -> Result<JoinedAgent, super::Error> {
// Validate project exists if assigning.
if let Some(ref p) = project
&& !projects.contains_key(p.as_str())
{
return Err(super::Error::ProjectNotFound(format!(
"unknown project '{p}'"
)));
}
match agents.iter_mut().find(|a| a.id == id) {
None => Err(super::Error::InvalidAgent(format!("agent not found: {id}"))),
Some(a) => {
a.assigned_project = project;
Ok(a.clone())
}
}
}
/// Update an agent's last-seen timestamp. Returns `true` if the agent was found.
pub fn heartbeat(agents: &mut [JoinedAgent], id: &str, now: f64) -> bool {
match agents.iter_mut().find(|a| a.id == id) {
None => false,
Some(a) => {
a.last_seen = now;
true
}
}
}
// ── Tests ────────────────────────────────────────────────────────────────────
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn create_agent_sets_fields() {
let agent = create_agent("id-1".into(), "lbl".into(), "ws://a".into(), 100.0);
assert_eq!(agent.id, "id-1");
assert_eq!(agent.label, "lbl");
assert_eq!(agent.address, "ws://a");
assert_eq!(agent.registered_at, 100.0);
assert_eq!(agent.last_seen, 100.0);
assert!(agent.assigned_project.is_none());
}
#[test]
fn remove_agent_by_id() {
let mut agents = vec![
create_agent("a".into(), "A".into(), "ws://a".into(), 0.0),
create_agent("b".into(), "B".into(), "ws://b".into(), 0.0),
];
assert!(remove_agent(&mut agents, "a"));
assert_eq!(agents.len(), 1);
assert_eq!(agents[0].id, "b");
}
#[test]
fn remove_agent_missing_returns_false() {
let mut agents = vec![];
assert!(!remove_agent(&mut agents, "x"));
}
#[test]
fn assign_agent_to_valid_project() {
let mut projects = BTreeMap::new();
projects.insert(
"proj".into(),
ProjectEntry {
url: "http://p".into(),
},
);
let mut agents = vec![create_agent("a".into(), "A".into(), "ws://a".into(), 0.0)];
let result = assign_agent(&mut agents, "a", Some("proj".into()), &projects);
assert!(result.is_ok());
assert_eq!(result.unwrap().assigned_project, Some("proj".into()));
}
#[test]
fn assign_agent_to_unknown_project_fails() {
let projects = BTreeMap::new();
let mut agents = vec![create_agent("a".into(), "A".into(), "ws://a".into(), 0.0)];
let result = assign_agent(&mut agents, "a", Some("nope".into()), &projects);
assert!(result.is_err());
}
#[test]
fn assign_agent_unknown_id_fails() {
let projects = BTreeMap::new();
let mut agents: Vec<JoinedAgent> = vec![];
let result = assign_agent(&mut agents, "x", None, &projects);
assert!(result.is_err());
}
#[test]
fn heartbeat_updates_last_seen() {
let mut agents = vec![create_agent("a".into(), "A".into(), "ws://a".into(), 0.0)];
assert!(heartbeat(&mut agents, "a", 999.0));
assert_eq!(agents[0].last_seen, 999.0);
}
#[test]
fn heartbeat_unknown_id_returns_false() {
let mut agents: Vec<JoinedAgent> = vec![];
assert!(!heartbeat(&mut agents, "x", 1.0));
}
}