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

huskies: merge 1154 story Extend gateway_health with a relay-working signal — is each sled actually delivering events?

Browse Source
This commit is contained in:
dave
2026-05-20 00:42:24 +00:00
parent a3ac09f8a3
commit 2a5359051e
3 changed files with 118 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/src/chat/transport/matrix/health.rs
+38 -5
View File
@@ -73,6 +73,16 @@ impl HealthLine {
}
}
/// Append `relay=X` information to this line's detail field.
fn with_relay(mut self, relay: &str) -> Self {
let relay_text = format!("relay={relay}");
self.detail = Some(match self.detail {
Some(d) => format!("{d} | {relay_text}"),
None => relay_text,
});
self
}
/// Format as a single Markdown-friendly line.
fn format(&self) -> String {
let status = match self.status {
@@ -216,7 +226,8 @@ async fn check_build_hash(project_root: &std::path::Path) -> HealthLine {
/// Check each registered sled's `/health` endpoint with a 5-second timeout.
///
/// Returns one [`HealthLine`] per sled. PASS when the sled responds with HTTP
/// 2xx; FAIL when the request times out or returns an error status.
/// 2xx; FAIL when the request times out or returns an error status. Each line
/// includes a `relay=X` indicator derived from the gateway CRDT event log.
async fn check_sleds(
store: &tokio::sync::RwLock<BTreeMap<String, crate::service::gateway::config::ProjectEntry>>,
) -> Vec<HealthLine> {
@@ -235,31 +246,53 @@ async fn check_sleds(
)];
}
let event_entries = crate::crdt_state::read_all_event_log_entries();
let now_secs = chrono::Utc::now().timestamp() as f64;
let client = reqwest::Client::new();
let mut lines = Vec::new();
for (name, url_opt) in entries {
let subsystem = format!("sled:{name}");
let relay = {
let latest = event_entries
.iter()
.filter(|e| e.sled_id == name)
.map(|e| e.timestamp)
.fold(f64::NEG_INFINITY, f64::max);
if latest == f64::NEG_INFINITY {
"never"
} else if now_secs - latest <= crate::service::gateway::RELAY_MAX_AGE_SECS {
"ok"
} else {
"silent"
}
};
let line = match url_opt {
None => HealthLine::warn(subsystem, "no URL configured", "set url in projects.toml"),
None => HealthLine::warn(subsystem, "no URL configured", "set url in projects.toml")
.with_relay(relay),
Some(url) => {
let health_url = format!("{}/health", url.trim_end_matches('/'));
let result = timeout(Duration::from_secs(5), client.get(&health_url).send()).await;
match result {
Err(_) => {
HealthLine::fail(subsystem, "timed out", "check container is running")
.with_relay(relay)
}
Ok(Err(e)) => HealthLine::fail(
subsystem,
format!("unreachable: {}", short_error(&e.to_string())),
"check container is running",
),
Ok(Ok(resp)) if resp.status().is_success() => HealthLine::pass(subsystem),
)
.with_relay(relay),
Ok(Ok(resp)) if resp.status().is_success() => {
HealthLine::pass(subsystem).with_relay(relay)
}
Ok(Ok(resp)) => HealthLine::fail(
subsystem,
format!("HTTP {}", resp.status().as_u16()),
"check container logs",
),
)
.with_relay(relay),
}
}
};
server/src/http/gateway/mcp.rs
+76 -1
View File
@@ -497,6 +497,28 @@ async fn handle_gateway_status_tool(state: &GatewayState, id: Option<Value>) ->
}
}
/// Returns `"ok"`, `"silent"`, or `"never"` for a project based on its most recent
/// CRDT event log entry. The gateway appends events using the project name as `sled_id`,
/// so filtering by project name identifies all events received from that sled.
fn relay_status(
entries: &[crate::crdt_state::EventLogEntryRaw],
project: &str,
now_secs: f64,
) -> &'static str {
let latest = entries
.iter()
.filter(|e| e.sled_id == project)
.map(|e| e.timestamp)
.fold(f64::NEG_INFINITY, f64::max);
if latest == f64::NEG_INFINITY {
"never"
} else if now_secs - latest <= crate::service::gateway::RELAY_MAX_AGE_SECS {
"ok"
} else {
"silent"
}
}
async fn handle_gateway_health_tool(state: &GatewayState, id: Option<Value>) -> JsonRpcResponse {
let mut results = BTreeMap::new();
@@ -510,6 +532,8 @@ async fn handle_gateway_health_tool(state: &GatewayState, id: Option<Value>) ->
.iter()
.map(|(n, e)| (n.clone(), e.url.clone()))
.collect();
let event_entries = crate::crdt_state::read_all_event_log_entries();
let now_secs = chrono::Utc::now().timestamp() as f64;
let sled_conns = state.sled_connections.read().await;
for (name, url_opt) in &project_names {
let status = if let Some(conn) = sled_conns.get(name) {
@@ -527,7 +551,8 @@ async fn handle_gateway_health_tool(state: &GatewayState, id: Option<Value>) ->
} else {
"no uplink and no url configured".to_string()
};
results.insert(name.clone(), status);
let relay = relay_status(&event_entries, name, now_secs);
results.insert(name.clone(), format!("{status} relay={relay}"));
}
drop(sled_conns);
@@ -1044,6 +1069,56 @@ mod tests {
Arc::new(GatewayState::new(config, config_dir.to_path_buf(), 3000).unwrap())
}
#[tokio::test]
async fn gateway_health_relay_status_distinguishes_active_and_silent_sleds() {
crate::crdt_state::init_for_test();
let dir = tempfile::tempdir().unwrap();
let mut projects = BTreeMap::new();
projects.insert(
"project-a".to_string(),
ProjectEntry::with_url("http://127.0.0.1:9001"),
);
projects.insert(
"project-b".to_string(),
ProjectEntry::with_url("http://127.0.0.1:9002"),
);
let config = GatewayConfig {
projects,
sled_tokens: BTreeMap::new(),
};
let state = Arc::new(GatewayState::new(config, dir.path().to_path_buf(), 3000).unwrap());
// Fire a recent StageTransition for project-a only.
let now_ms = chrono::Utc::now().timestamp_millis() as u64;
gateway::broadcast_status_event(
&state,
"project-a".to_string(),
crate::service::events::StoredEvent::StageTransition {
story_id: "1_story_test".to_string(),
story_name: String::new(),
from_stage: "Backlog".to_string(),
to_stage: "Current".to_string(),
timestamp_ms: now_ms,
},
);
let resp = handle_gateway_health_tool(&state, Some(json!(1))).await;
assert!(resp.result.is_some(), "expected result, got error");
let text = resp.result.unwrap()["content"][0]["text"]
.as_str()
.unwrap()
.to_string();
assert!(
text.contains("relay=ok"),
"project-a should report relay=ok; got:\n{text}"
);
assert!(
text.contains("relay=never") || text.contains("relay=silent"),
"project-b should report relay=never or relay=silent; got:\n{text}"
);
}
#[tokio::test]
async fn adopt_project_tool_missing_name_returns_error() {
let dir = tempfile::tempdir().unwrap();
server/src/service/gateway/mod.rs
+4
View File
@@ -56,6 +56,10 @@ pub struct GatewayStatusEvent {
/// considers the connection stale (story 899 AC 3).
pub const HEARTBEAT_MAX_AGE_MS: i64 = 30_000;
/// Maximum event age in seconds before a sled relay is considered `silent` rather than `ok`.
/// Default is 10 minutes; configurable at the call site via this constant.
pub const RELAY_MAX_AGE_SECS: f64 = 600.0;
/// Default per-request timeout, in milliseconds, for an MCP call proxied over
/// a sled uplink WebSocket. Mirrors the existing reqwest-based path which has
/// no explicit cap; we set a generous bound so long-running tools (e.g.