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

huskies: merge 1180 story Sled↔gateway goes WS-only: remove the deprecated HTTP fallback paths

Browse Source
This commit is contained in:
Huskies Agent
2026-07-16 18:10:14 +00:00
parent 0e684fb06f
commit de7d22cb9f
13 changed files with 203 additions and 1053 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/src/service/events/buffer.rs
+5 -110
View File
@@ -1,16 +1,12 @@
//! Pure event-buffer types — no side effects.
//! Pure event types — no side effects.
//!
//! `StoredEvent` and `EventBuffer` contain only data-transformation and
//! structural logic; all I/O (clocks, spawned tasks) lives in `io.rs`.
//! `StoredEvent` is the wire format shared by the gateway's WS event-push
//! relay (`gateway_relay.rs`, `http/gateway/websocket.rs`) — data-transformation
//! only; all I/O lives in `io.rs`.
use serde::{Deserialize, Serialize};
use std::collections::VecDeque;
use std::sync::{Arc, Mutex};
/// Maximum number of events retained in the in-memory buffer.
pub const MAX_BUFFER_SIZE: usize = 500;
/// A pipeline event stored in the event buffer with a timestamp.
/// A pipeline event pushed to the gateway over the WS event-push relay.
#[derive(Clone, Debug, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum StoredEvent {
@@ -62,88 +58,10 @@ impl StoredEvent {
}
}
/// Shared, thread-safe ring buffer of recent pipeline events.
///
/// Wrapped in `Arc` so it can be shared between the background subscriber
/// task and the HTTP handler. The inner `Mutex` guards the `VecDeque`.
#[derive(Clone, Debug)]
pub struct EventBuffer(Arc<Mutex<VecDeque<StoredEvent>>>);
impl EventBuffer {
/// Create a new, empty event buffer.
pub fn new() -> Self {
EventBuffer(Arc::new(Mutex::new(VecDeque::new())))
}
/// Append an event to the buffer, evicting the oldest entry if the buffer
/// exceeds [`MAX_BUFFER_SIZE`].
pub fn push(&self, event: StoredEvent) {
let mut buf = self.0.lock().unwrap();
if buf.len() >= MAX_BUFFER_SIZE {
buf.pop_front();
}
buf.push_back(event);
}
/// Return all events whose `timestamp_ms` is strictly greater than `since_ms`.
pub fn events_since(&self, since_ms: u64) -> Vec<StoredEvent> {
let buf = self.0.lock().unwrap();
buf.iter()
.filter(|e| e.timestamp_ms() > since_ms)
.cloned()
.collect()
}
}
impl Default for EventBuffer {
fn default() -> Self {
Self::new()
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn push_and_retrieve_events() {
let buf = EventBuffer::new();
buf.push(StoredEvent::MergeFailure {
story_id: "42_story_x".to_string(),
story_name: String::new(),
reason: "conflict".to_string(),
timestamp_ms: 1000,
});
buf.push(StoredEvent::StoryBlocked {
story_id: "43_story_y".to_string(),
story_name: String::new(),
reason: "retry limit".to_string(),
timestamp_ms: 2000,
});
let all = buf.events_since(0);
assert_eq!(all.len(), 2);
let after_1000 = buf.events_since(1000);
assert_eq!(after_1000.len(), 1);
assert!(matches!(after_1000[0], StoredEvent::StoryBlocked { .. }));
}
#[test]
fn evicts_oldest_when_full() {
let buf = EventBuffer::new();
for i in 0..MAX_BUFFER_SIZE + 1 {
buf.push(StoredEvent::MergeFailure {
story_id: format!("{i}_story_x"),
story_name: String::new(),
reason: "x".to_string(),
timestamp_ms: i as u64,
});
}
assert_eq!(buf.events_since(0).len(), MAX_BUFFER_SIZE);
assert!(buf.events_since(0).iter().all(|e| e.timestamp_ms() > 0));
}
#[test]
fn timestamp_ms_accessor_for_all_variants() {
let variants = [
@@ -171,27 +89,4 @@ mod tests {
assert_eq!(variants[1].timestamp_ms(), 200);
assert_eq!(variants[2].timestamp_ms(), 300);
}
#[test]
fn events_since_filters_by_timestamp() {
let buf = EventBuffer::new();
for ts in [100u64, 200, 300] {
buf.push(StoredEvent::MergeFailure {
story_id: "x".to_string(),
story_name: String::new(),
reason: "r".to_string(),
timestamp_ms: ts,
});
}
// strictly greater than 100
let result = buf.events_since(100);
assert_eq!(result.len(), 2);
assert!(result.iter().all(|e| e.timestamp_ms() > 100));
}
#[test]
fn default_creates_empty_buffer() {
let buf = EventBuffer::default();
assert_eq!(buf.events_since(0).len(), 0);
}
}
server/src/service/events/io.rs
-80
View File
@@ -1,80 +0,0 @@
//! Events I/O wrappers — the ONLY place in `service/events/` that may perform
//! side effects such as reading the system clock or spawning async tasks.
use crate::io::watcher::WatcherEvent;
use tokio::sync::broadcast;
use super::buffer::{EventBuffer, StoredEvent};
/// Returns the current Unix timestamp in milliseconds.
pub(super) fn now_ms() -> u64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_millis() as u64)
.unwrap_or(0)
}
/// Spawn a background task that consumes [`WatcherEvent`] broadcasts and
/// stores relevant events in `buffer`.
///
/// Only [`WatcherEvent::WorkItem`] (with a known `from_stage`),
/// [`WatcherEvent::MergeFailure`], and [`WatcherEvent::StoryBlocked`]
/// variants are stored. All other variants are silently ignored.
pub fn subscribe_to_watcher(buffer: EventBuffer, mut rx: broadcast::Receiver<WatcherEvent>) {
tokio::spawn(async move {
loop {
match rx.recv().await {
Ok(WatcherEvent::WorkItem {
stage,
item_id,
from_stage,
..
}) => {
if let Some(from) = from_stage {
let story_name = lookup_story_name(&item_id);
buffer.push(StoredEvent::StageTransition {
story_id: item_id,
story_name,
from_stage: from,
to_stage: stage,
timestamp_ms: now_ms(),
});
}
}
Ok(WatcherEvent::MergeFailure { story_id, reason }) => {
let story_name = lookup_story_name(&story_id);
buffer.push(StoredEvent::MergeFailure {
story_id,
story_name,
reason,
timestamp_ms: now_ms(),
});
}
Ok(WatcherEvent::StoryBlocked { story_id, reason }) => {
let story_name = lookup_story_name(&story_id);
buffer.push(StoredEvent::StoryBlocked {
story_id,
story_name,
reason,
timestamp_ms: now_ms(),
});
}
Ok(_) => {}
Err(broadcast::error::RecvError::Lagged(n)) => {
crate::slog!("[events] Subscriber lagged, skipped {n} events");
}
Err(broadcast::error::RecvError::Closed) => {
crate::slog!("[events] Watcher channel closed; stopping event subscriber");
break;
}
}
}
});
}
/// Look up the human-readable story name from the CRDT, or empty when absent.
fn lookup_story_name(story_id: &str) -> String {
crate::crdt_state::read_item(story_id)
.map(|view| view.name().to_string())
.unwrap_or_default()
}
server/src/service/events/mod.rs
+8 -20
View File
@@ -1,28 +1,23 @@
//! Events service — public API for the events domain.
//!
//! This module re-exports the pure buffer types from `buffer.rs` and the
//! side-effectful watcher subscription from `io.rs`. HTTP handlers call
//! these exports instead of containing the logic inline.
//! Re-exports [`StoredEvent`], the wire format shared by the gateway's
//! WS-based event-push relay (`gateway_relay.rs` on the project side,
//! `http/gateway/websocket.rs`'s event-push handler on the gateway side).
//!
//! Conventions: `docs/architecture/service-modules.md`
/// Bounded in-memory event ring buffer for SSE streaming.
/// Pure event types — no side effects.
pub mod buffer;
pub(super) mod io;
pub use buffer::{EventBuffer, StoredEvent};
// Re-exported for tests (http::events uses it via `use super::*`).
#[allow(unused_imports)]
pub use buffer::MAX_BUFFER_SIZE;
pub use io::subscribe_to_watcher;
pub use buffer::StoredEvent;
// ── Error type ────────────────────────────────────────────────────────────────
/// Typed errors returned by `service::events` functions.
///
/// Events operations on the in-memory buffer are infallible; this enum
/// exists to satisfy the module convention and to accommodate future
/// error cases (e.g. persistence).
/// Events operations are currently infallible; this enum exists to satisfy
/// the module convention and to accommodate future error cases (e.g.
/// persistence).
#[allow(dead_code)]
#[derive(Debug)]
pub enum Error {
@@ -37,10 +32,3 @@ impl std::fmt::Display for Error {
}
}
}
// ── Public API ────────────────────────────────────────────────────────────────
/// Return all events in `buffer` recorded after `since_ms` milliseconds.
pub fn events_since(buffer: &EventBuffer, since_ms: u64) -> Vec<StoredEvent> {
buffer.events_since(since_ms)
}
server/src/service/gateway/io.rs
+1 -89
View File
@@ -7,7 +7,7 @@
use super::config::{GatewayConfig, ProjectEntry};
pub use reqwest::Client;
use serde_json::{Value, json};
use std::collections::{BTreeMap, HashMap};
use std::collections::BTreeMap;
use std::path::Path;
// ── Config I/O ───────────────────────────────────────────────────────────────
@@ -239,28 +239,6 @@ pub fn read_installed_manifest(
// ── MCP proxy I/O ───────────────────────────────────────────────────────────
/// Proxy a raw MCP request body to the given project URL.
pub async fn proxy_mcp_call(
client: &Client,
base_url: &str,
request_bytes: &[u8],
) -> Result<Vec<u8>, String> {
let mcp_url = format!("{}/mcp", base_url.trim_end_matches('/'));
let resp = client
.post(&mcp_url)
.header("Content-Type", "application/json")
.body(request_bytes.to_vec())
.send()
.await
.map_err(|e| format!("failed to reach {mcp_url}: {e}"))?;
resp.bytes()
.await
.map(|b| b.to_vec())
.map_err(|e| format!("failed to read response from {mcp_url}: {e}"))
}
/// Proxy an MCP `tools/call` request to the sled with `Accept: text/event-stream`
/// and return the raw response for streaming. No per-request timeout is applied
/// so long-running tool calls (e.g. `run_tests`, up to 1200 s) are not cut short.
@@ -563,72 +541,6 @@ pub fn spawn_gateway_broadcaster_forwarder(
});
}
/// Spawn a background task that polls events from all project servers.
pub fn spawn_gateway_notification_poller(
transport: std::sync::Arc<dyn crate::chat::ChatTransport>,
room_ids: Vec<String>,
project_urls: BTreeMap<String, String>,
poll_interval_secs: u64,
) {
tokio::spawn(async move {
let client = Client::builder()
.timeout(std::time::Duration::from_secs(10))
.build()
.unwrap_or_else(|_| Client::new());
let interval = std::time::Duration::from_secs(poll_interval_secs.max(1));
let mut last_ts: HashMap<String, u64> = project_urls
.keys()
.map(|name| (name.clone(), 0u64))
.collect();
loop {
for (project_name, base_url) in &project_urls {
let since = last_ts.get(project_name).copied().unwrap_or(0);
let url = format!("{base_url}/api/events?since={since}");
let response = match client.get(&url).send().await {
Ok(r) => r,
Err(e) => {
crate::slog!(
"[gateway-poller] {project_name}: unreachable ({e}); skipping"
);
continue;
}
};
let events: Vec<crate::service::events::StoredEvent> = match response.json().await {
Ok(v) => v,
Err(e) => {
crate::slog!(
"[gateway-poller] {project_name}: failed to parse events: {e}"
);
continue;
}
};
for event in &events {
let ts = event.timestamp_ms();
if ts > *last_ts.get(project_name).unwrap_or(&0) {
last_ts.insert(project_name.clone(), ts);
}
let (plain, html) = super::polling::format_gateway_event(project_name, event);
for room_id in &room_ids {
if let Err(e) = transport.send_message(room_id, &plain, &html).await {
crate::slog!(
"[gateway-poller] Failed to send notification to {room_id}: {e}"
);
}
}
}
}
tokio::time::sleep(interval).await;
}
});
}
// ── Gateway bot spawn ───────────────────────────────────────────────────────
/// Re-export type alias for the active project lock.
server/src/service/gateway/mod.rs
+15 -12
View File
@@ -24,7 +24,6 @@ pub use config::{GatewayConfig, ProjectEntry};
pub use identity::{IdentityCheck, check_identity};
pub use io::{
fetch_all_project_pipeline_statuses, probe_identity, spawn_gateway_broadcaster_forwarder,
spawn_gateway_notification_poller,
};
use crate::http::context::PermissionForward;
@@ -364,21 +363,25 @@ impl GatewayState {
}
}
/// Proxy an MCP request to the active project, preferring the live
/// sled-uplink WebSocket when available (story 899 AC 2) and falling
/// back to HTTP otherwise.
/// Proxy an MCP request to the active project over its live sled-uplink
/// WebSocket (story 899 AC 2).
///
/// The gateway is WS-only for MCP proxying (story 1180): when no live
/// uplink connection exists for the active project, this returns an
/// immediate, actionable error naming the sled rather than falling back
/// to HTTP. Callers (e.g. bot chat commands) surface this error straight
/// to the user instead of hanging on an unreachable HTTP endpoint.
///
/// Returns the raw response body bytes ready to be relayed to the caller.
pub async fn proxy_active_mcp(&self, bytes: &[u8]) -> Result<Vec<u8>, String> {
if let Some(conn) = self.active_sled_connection().await {
return proxy_mcp_via_ws(&conn, bytes).await;
let name = self.active_project.read().await.clone();
match self.active_sled_connection().await {
Some(conn) => proxy_mcp_via_ws(&conn, bytes).await,
None => Err(format!(
"sled '{name}' has no live WS uplink connection; \
ensure the sled is running and connected to this gateway"
)),
}
let url = self.active_url().await.map_err(|e| e.to_string())?;
crate::slog!(
"[gateway] MCP proxy: WS uplink unavailable, falling back to HTTP \
(deprecated, will be removed once all sleds are WS-only)"
);
crate::service::gateway::io::proxy_mcp_call(&self.client, &url, bytes).await
}
}