mara/hyperhive
mara/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Activity

remove hive and agent apis nothing calls: build-log routes, /api/compact, host.sock list, mcp.sock wake

Browse source
This commit is contained in:
atlas 2026-11-13 02:48:37 +01:00
commit 36c7781245
7 changed files with 115 additions and 801 deletions
Download patch file Download diff file Expand all files Collapse all files

8
hive-c0re/src/coordinator.rs
View file

@ -25,9 +25,7 @@ pub struct Coordinator {
pub broker: Arc<Broker>,
/// Full build-log capture. `lifecycle::run` `start()`s a row per attempt,
/// pipes every stdout/stderr line into it, and `finish()`es it on
/// child exit. Dashboard reads it via `list_recent_for_agent` /
/// `get_full` for the per-card chip + side-panel viewer. See
/// `build_logs.rs` for retention.
/// child exit. See `build_logs.rs` for retention.
pub build_logs: Arc<crate::build_logs::BuildLogs>,
/// Store path of the hyperhive flake source (no fragment). Used to
/// identify the running revision (`current_flake_rev`).
@ -1061,7 +1059,7 @@ mod open_tests {
let coord = open(&db);
let row = coord.build_logs.get_full(id).expect("get").expect("Some");
assert_eq!(row.header.status.as_deref(), Some("fail"));
assert!(row.header.finished_at.is_some());
assert_eq!(row.status.as_deref(), Some("fail"));
assert!(row.finished_at.is_some());
}
}

365
hive-c0re/src/dashboard/build_logs.rs
View file

@ -1,365 +0,0 @@
//! Build-log endpoints for the dashboard.
//!
//! Header lists (all-agents + per-agent), the full row by id, a `text/plain`
//! download, and an SSE stream that delivers incremental stdout/stderr while
//! a build runs (closing once it finishes or the row is vacuum-reaped).
use std::convert::Infallible;
use axum::{
extract::{Path as AxumPath, State},
http::StatusCode,
response::{
IntoResponse, Response,
sse::{Event, KeepAlive, Sse},
},
};
use serde::{Deserialize, Serialize};
use tokio_stream::Stream;
use tokio_stream::wrappers::ReceiverStream;
use utoipa::IntoParams;
use super::{AppState, Ident, error_response};
use crate::build_logs::{BuildLogFull, BuildLogHeader};
#[derive(Deserialize, IntoParams)]
pub(super) struct BuildLogsAllQuery {
/// Max rows to return. Capped at 100. Default 30.
#[serde(default)]
limit: Option<usize>,
}
/// Most-recent build log headers across
/// all agents, newest first.
///
/// Same JSON shape as the per-agent endpoint.
#[utoipa::path(
get,
path = "/api/build-logs",
params(BuildLogsAllQuery),
responses(
(status = 200, description = "recent build log headers, newest first", body = Vec<BuildLogHeader>),
(status = 500, description = "sqlite read failed"),
),
tag = "build_logs"
)]
pub(super) async fn get_build_logs_all(
State(state): State<AppState>,
axum::extract::Query(q): axum::extract::Query<BuildLogsAllQuery>,
) -> Response {
let limit = q.limit.unwrap_or(30);
match state.coord.build_logs.list_recent_all(limit) {
Ok(rows) => axum::Json(rows).into_response(),
Err(e) => error_response(&format!("build-logs all: {e:#}")),
}
}
#[derive(Deserialize, IntoParams)]
pub(super) struct BuildLogsQuery {
/// Maximum number of rows to return. Capped server-side at 50
/// (see `build_logs::list_recent_for_agent`). Default 10.
#[serde(default)]
limit: Option<usize>,
}
/// Most-recent build log
/// headers for one agent, newest first.
///
/// Returns `Vec<BuildLogHeader>` (JSON). Backs the per-agent log chip
/// in the agent card and the side-panel header list.
#[utoipa::path(
get,
path = "/api/build-logs/{agent}",
params(
("agent" = String, Path, description = "agent name"),
BuildLogsQuery,
),
responses(
(status = 200, description = "recent build log headers for the agent, newest first", body = Vec<BuildLogHeader>),
(status = 400, description = "bad agent name"),
(status = 500, description = "sqlite read failed"),
),
tag = "build_logs"
)]
pub(super) async fn get_build_logs_agent(
State(state): State<AppState>,
AxumPath(name): AxumPath<String>,
axum::extract::Query(q): axum::extract::Query<BuildLogsQuery>,
) -> Response {
let name = match Ident::parse(&name) {
Ok(n) => n,
Err(reason) => {
return (StatusCode::BAD_REQUEST, format!("bad agent name: {reason}")).into_response();
}
};
let limit = q.limit.unwrap_or(10);
match state
.coord
.build_logs
.list_recent_for_agent(name.as_str(), limit)
{
Ok(rows) => axum::Json(rows).into_response(),
Err(e) => error_response(&format!("build-logs {name}: {e:#}")),
}
}
/// Full build log row (stdout +
/// stderr concatenated) by id.
///
/// Returns `BuildLogFull` (JSON), or HTTP 404 when the id doesn't
/// exist (vacuum-reaped, or the operator passed a stale id from a
/// refresh race).
#[utoipa::path(
get,
path = "/api/build-logs/id/{id}",
params(("id" = i64, Path, description = "build log row id")),
responses(
(status = 200, description = "full build log row", body = BuildLogFull),
(status = 404, description = "no such build log row"),
(status = 500, description = "sqlite read failed"),
),
tag = "build_logs"
)]
pub(super) async fn get_build_log_full(
State(state): State<AppState>,
AxumPath(id): AxumPath<i64>,
) -> Response {
match state.coord.build_logs.get_full(id) {
Ok(Some(log)) => axum::Json(log).into_response(),
Ok(None) => (StatusCode::NOT_FOUND, format!("build log #{id} not found")).into_response(),
Err(e) => error_response(&format!("build-log {id}: {e:#}")),
}
}
/// The build log for a **queue node**,
/// resolved node id → log-row id → full log.
///
/// Same `BuildLogFull` JSON (`stdout` / `stderr` + header) as
/// `get_build_log_full`; HTTP 404 when the node has no linked log (the
/// client gates the request on the wire node's `build_log_id`, but a vacuum
/// race can still 404). This is the on-demand live-log-panel fetch,
/// distinct from the `build_log_id` on the wire — that id is for
/// deep-linking to the BUILD L0GS tab's full history view, not for
/// fetching the log content itself.
#[utoipa::path(
get,
path = "/api/build-log/{node_id}",
params(("node_id" = u64, Path, description = "job-queue node id")),
responses(
(status = 200, description = "full build log row for the node's linked log", body = BuildLogFull),
(status = 404, description = "node has no linked build log, or the log row is gone"),
(status = 500, description = "sqlite read failed"),
),
tag = "build_logs"
)]
pub(super) async fn get_build_log_for_node(
State(state): State<AppState>,
AxumPath(node_id): AxumPath<u64>,
) -> Response {
match state.coord.build_logs.id_for_node(node_id) {
Some(log_id) => get_build_log_full(State(state), AxumPath(log_id)).await,
None => (
StatusCode::NOT_FOUND,
format!("node #{node_id} has no build log"),
)
.into_response(),
}
}
/// The node's build log as `text/plain`
/// for download (delegates to `get_build_log_raw` after resolving the node id).
#[utoipa::path(
get,
path = "/api/build-log/{node_id}/raw",
params(("node_id" = u64, Path, description = "job-queue node id")),
responses(
(status = 200, description = "build log text for download", body = String, content_type = "text/plain"),
(status = 404, description = "node has no linked build log, or the log row is gone"),
(status = 500, description = "sqlite read failed"),
),
tag = "build_logs"
)]
pub(super) async fn get_build_log_raw_for_node(
State(state): State<AppState>,
AxumPath(node_id): AxumPath<u64>,
) -> Response {
match state.coord.build_logs.id_for_node(node_id) {
Some(log_id) => get_build_log_raw(State(state), AxumPath(log_id)).await,
None => (
StatusCode::NOT_FOUND,
format!("node #{node_id} has no build log"),
)
.into_response(),
}
}
/// JSON frame sent on the `/api/build-logs/id/{id}/stream` SSE channel.
/// `stdout_append` / `stderr_append` carry only the new bytes since the
/// last frame; `done = true` means the build finished and the stream
/// will close after this frame.
#[derive(Serialize)]
struct BuildLogFrame {
stdout_append: String,
stderr_append: String,
#[serde(skip_serializing_if = "Option::is_none")]
status: Option<String>,
done: bool,
}
/// SSE stream that delivers
/// incremental stdout/stderr as a build runs.
///
/// The client connects when it opens a running-build panel; the
/// stream closes automatically once the build finishes (or the row
/// disappears due to a vacuum). Each frame is a JSON-serialised
/// `BuildLogFrame`. The first frame always carries the full
/// accumulated log so far (cursors start at 0); subsequent frames
/// carry only new bytes. `done: true` on the final frame signals the
/// browser to close the `EventSource`.
#[utoipa::path(
get,
path = "/api/build-logs/id/{id}/stream",
params(("id" = i64, Path, description = "build log row id")),
responses(
(status = 200, description = "server-sent event stream; each event's \
`data` is a JSON-serialised `BuildLogFrame` \
(stdout_append/stderr_append/status/done)",
body = String, content_type = "text/event-stream"),
),
tag = "build_logs"
)]
pub(super) async fn get_build_log_stream(
State(state): State<AppState>,
AxumPath(id): AxumPath<i64>,
) -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
let (tx, rx) = tokio::sync::mpsc::channel::<Result<Event, Infallible>>(32);
let logs = state.coord.build_logs.clone();
tokio::spawn(async move {
let mut notify_rx = logs.subscribe_notifications();
let mut stdout_cursor = 0usize;
let mut stderr_cursor = 0usize;
// ── initial snapshot ──────────────────────────────────────────
match logs.get_progress(id, stdout_cursor, stderr_cursor) {
Ok(Some(prog)) => {
stdout_cursor += prog.stdout_append.len();
stderr_cursor += prog.stderr_append.len();
let done = prog.finished_at.is_some();
if let Ok(json) = serde_json::to_string(&BuildLogFrame {
stdout_append: prog.stdout_append,
stderr_append: prog.stderr_append,
status: prog.status,
done,
}) {
let _ = tx.send(Ok(Event::default().data(json))).await;
}
if done {
return;
}
}
Ok(None) => {
// Row missing — send a single error event and exit.
let _ = tx
.send(Ok(Event::default()
.event("error")
.data(format!("build log #{id} not found"))))
.await;
return;
}
Err(e) => {
let _ = tx
.send(Ok(Event::default()
.event("error")
.data(format!("build log #{id}: {e:#}"))))
.await;
return;
}
}
// ── live delta loop ───────────────────────────────────────────
loop {
match notify_rx.recv().await {
// Notification for a different build — ignore and wait
// for the next one.
Ok(notif_id) if notif_id != id => {}
Ok(_) => {
match logs.get_progress(id, stdout_cursor, stderr_cursor) {
Ok(Some(prog)) => {
stdout_cursor += prog.stdout_append.len();
stderr_cursor += prog.stderr_append.len();
let done = prog.finished_at.is_some();
if let Ok(json) = serde_json::to_string(&BuildLogFrame {
stdout_append: prog.stdout_append,
stderr_append: prog.stderr_append,
status: prog.status,
done,
}) && tx.send(Ok(Event::default().data(json))).await.is_err()
{
return; // browser disconnected
}
if done {
return;
}
}
Ok(None) | Err(_) => return, // vacuum reaped row / channel closed
}
}
Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => {}
Err(tokio::sync::broadcast::error::RecvError::Closed) => return,
}
}
});
Sse::new(ReceiverStream::new(rx)).keep_alive(KeepAlive::default())
}
/// Full log as `text/plain` for
/// download.
///
/// Stdout and stderr are concatenated with a `--- stderr ---`
/// separator (same layout the JS side-panel renders). The
/// `Content-Disposition` header triggers a browser download with a
/// descriptive filename so the operator can save and share the log.
#[utoipa::path(
get,
path = "/api/build-logs/id/{id}/raw",
params(("id" = i64, Path, description = "build log row id")),
responses(
(status = 200, description = "build log text for download", body = String, content_type = "text/plain"),
(status = 404, description = "no such build log row"),
(status = 500, description = "sqlite read failed"),
),
tag = "build_logs"
)]
pub(super) async fn get_build_log_raw(
State(state): State<AppState>,
AxumPath(id): AxumPath<i64>,
) -> Response {
match state.coord.build_logs.get_full(id) {
Ok(Some(log)) => {
let mut text = log.stdout;
if !log.stderr.is_empty() {
text.push_str("\n--- stderr ---\n");
text.push_str(&log.stderr);
}
(
StatusCode::OK,
[
("content-type", "text/plain; charset=utf-8".to_string()),
(
"content-disposition",
format!(
"attachment; filename=\"build-log-{}-{}.txt\"",
log.header.agent, id
),
),
],
text,
)
.into_response()
}
Ok(None) => (StatusCode::NOT_FOUND, format!("build log #{id} not found")).into_response(),
Err(e) => error_response(&format!("build-log {id}: {e:#}")),
}
}

16
hive-c0re/src/dashboard/mod.rs
View file

@ -35,7 +35,6 @@ use crate::lifecycle;
),
tags(
(name = "health", description = "hive-wide liveness/readiness probes"),
(name = "build_logs", description = "build log headers, full rows, and raw text downloads"),
(name = "lifecycle_ops", description = "agent container lifecycle: rebuild/restart/start/stop/pause/limits"),
(name = "misc_api", description = "mark-all-read, compose, spawn-request"),
@ -46,7 +45,6 @@ use crate::lifecycle;
)]
struct ApiDoc;
mod build_logs;
// The single validated identifier type — homed in `hive-host-sock` (the crate
// owning agent-path facts) so every dashboard path-param validates through the
// same type used to build agent paths. Re-exported so submodules + the socket
@ -104,12 +102,6 @@ fn app(coord: Arc<Coordinator>) -> axum::Router {
.routes(routes!(health::get_health_ready))
.routes(routes!(state_snapshot::api_state))
.routes(routes!(misc_api::post_mark_all_read))
.routes(routes!(build_logs::get_build_logs_all))
.routes(routes!(build_logs::get_build_log_for_node))
.routes(routes!(build_logs::get_build_log_raw_for_node))
.routes(routes!(build_logs::get_build_logs_agent))
.routes(routes!(build_logs::get_build_log_full))
.routes(routes!(build_logs::get_build_log_raw))
.routes(routes!(rebuild_queue::post_rebuild_queue_cancel))
.routes(routes!(lifecycle_ops::post_destroy))
.routes(routes!(lifecycle_ops::post_kill))
@ -119,7 +111,6 @@ fn app(coord: Arc<Coordinator>) -> axum::Router {
.routes(routes!(lifecycle_ops::post_pause))
.routes(routes!(lifecycle_ops::post_resume))
.routes(routes!(tombstones::post_purge_tombstone))
.routes(routes!(build_logs::get_build_log_stream))
.routes(routes!(state_snapshot::dashboard_stream))
.routes(routes!(state_snapshot::jobq_graph))
.routes(routes!(state_snapshot::jobq_rollup))
@ -351,12 +342,6 @@ mod router_build_probe {
.routes(routes!(health::get_health_ready))
.routes(routes!(state_snapshot::api_state))
.routes(routes!(misc_api::post_mark_all_read))
.routes(routes!(build_logs::get_build_logs_all))
.routes(routes!(build_logs::get_build_log_for_node))
.routes(routes!(build_logs::get_build_log_raw_for_node))
.routes(routes!(build_logs::get_build_logs_agent))
.routes(routes!(build_logs::get_build_log_full))
.routes(routes!(build_logs::get_build_log_raw))
.routes(routes!(rebuild_queue::post_rebuild_queue_cancel))
.routes(routes!(lifecycle_ops::post_destroy))
.routes(routes!(lifecycle_ops::post_kill))
@ -366,7 +351,6 @@ mod router_build_probe {
.routes(routes!(lifecycle_ops::post_pause))
.routes(routes!(lifecycle_ops::post_resume))
.routes(routes!(tombstones::post_purge_tombstone))
.routes(routes!(build_logs::get_build_log_stream))
.routes(routes!(state_snapshot::dashboard_stream))
.routes(routes!(state_snapshot::jobq_graph))
.routes(routes!(state_snapshot::jobq_rollup));

3
hive-c0re/src/lifecycle/mod.rs
View file

@ -882,8 +882,7 @@ async fn priv_run(kind: &str, name: &str) -> Result<()> {
}
/// Like `priv_run` but stamps `node_id` onto the build-log row it opens, so
/// the dashboard can find the log from the queue node (and link to
/// `/api/build-logs/id/{id}/stream`).
/// the log can be found from the queue node.
///
/// This used to be a `Fn(i64)` callback that handed the row id *back* to the
/// queue, which then held it in a side map. The row carries the link itself

13
hive-c0re/src/server.rs
View file

@ -162,7 +162,6 @@ async fn dispatch(req: &HostRequest, coord: Arc<Coordinator>) -> HostResponse {
HostRequest::QueueNodes { ids } => {
HostResponse::nodes(coord.job_queue.node_subtrees(ids))
}
HostRequest::List => HostResponse::list(lifecycle::list().await?),
// The agents root is ours and not world-traversable, so this
// question is only answerable on this side of the socket —
// see the request's doc comment for why the client asks.
@ -745,6 +744,18 @@ mod tests {
use super::*;
/// A verb this daemon no longer serves fails to parse (answered with
/// `parse error` by the connection loop) while a served verb parses.
#[test]
fn a_removed_verb_is_a_parse_error_and_a_served_one_parses() {
let err =
serde_json::from_slice::<HostRequest>(br#"{"cmd":"list"}"#).expect_err("list is gone");
assert!(err.to_string().contains("unknown variant `list`"), "{err}");
assert!(
serde_json::from_slice::<HostRequest>(br#"{"cmd":"subscribe_agent_status"}"#).is_ok()
);
}
/// A real `Coordinator` over a throwaway sqlite dir; `handle_with` takes
/// an `Arc<Coordinator>` and `open` touches nothing outside the db path.
fn coordinator() -> (tempfile::TempDir, Arc<Coordinator>) {

34
hive-c0re/src/socket_server/mod.rs
View file

@ -9,7 +9,6 @@ use std::sync::Arc;
use anyhow::{Context, Result};
use hive_core_agent_sock::{Request, Response};
use hive_sh4re::inbox::Message;
use hive_sock_client::{LineDeadlines, RequestLine, read_request_line};
use tokio::io::{AsyncWriteExt, BufReader};
use tokio::net::{UnixListener, UnixStream};
@ -209,7 +208,6 @@ async fn dispatch(
handle_recv(coord, agent, *wait_seconds).await
}
hive_core_agent_sock::Request::Status => handle_status(coord, agent),
hive_core_agent_sock::Request::Wake { from, body } => handle_wake(coord, agent, from, body),
hive_core_agent_sock::Request::Recent { limit } => handle_recent(coord, agent, *limit),
hive_core_agent_sock::Request::SetStatus { text } => handle_set_status(coord, text),
hive_core_agent_sock::Request::GetAgentMeta { name } => {
@ -278,28 +276,6 @@ async fn handle_recv(
}
}
/// `Wake` — inject a wake into `agent`'s own inbox. Persisted through
/// the sqlite broker like any other message so it appears in message history
/// for post-mortem.
fn handle_wake(
coord: &Arc<Coordinator>,
agent: &str,
from: &str,
body: &str,
) -> hive_core_agent_sock::Response {
match coord.broker.send(&Message {
from: hive_sh4re::manager::trusted_sender(from),
to: agent.to_owned(),
body: body.to_owned(),
in_reply_to: None,
}) {
Ok(()) => hive_core_agent_sock::Response::Ok,
Err(e) => hive_core_agent_sock::Response::Err {
message: format!("{e:#}"),
},
}
}
/// `SetStatus` — validate the status text, then trigger a dashboard
/// rescan. The harness has already written the status file to its own
/// `state/` dir (it runs as the agent user), so this only refreshes the
@ -461,6 +437,16 @@ mod tests {
(dir, Arc::new(coord))
}
/// A verb this daemon no longer serves fails to parse (the connection loop
/// answers that with `parse error`, see `serve`) while a served verb parses.
#[test]
fn a_removed_verb_is_a_parse_error_and_a_served_one_parses() {
let removed = br#"{"cmd":"wake","from":"x","body":"y"}"#;
let err = serde_json::from_slice::<Request>(removed).expect_err("wake is gone");
assert!(err.to_string().contains("unknown variant `wake`"), "{err}");
assert!(serde_json::from_slice::<Request>(br#"{"cmd":"status"}"#).is_ok());
}
#[tokio::test]
async fn serve_answers_an_over_long_line_with_an_error_and_closes() {
let (_dir, coord) = coordinator();

281
hive-c0re/src/stores/build_logs.rs
View file

@ -8,10 +8,8 @@ use std::sync::{Arc, Mutex, OnceLock};
use anyhow::{Context, Result};
use chrono::Utc;
use rusqlite::{Connection, OptionalExtension, params};
use rusqlite::{Connection, params};
use serde::Serialize;
use tokio::sync::broadcast;
use utoipa::ToSchema;
use crate::db::Migration;
@ -103,65 +101,23 @@ pub enum BuildStatus {
Fail,
}
/// Header-only row returned by `list_recent_for_agent`. Carries the
/// metadata the dashboard's agent-card chip needs (status + age +
/// id-to-open) without the multi-MB stdout/stderr payload.
#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct BuildLogHeader {
pub id: i64,
/// A stored row with its captured text, as read back by tests.
#[cfg(test)]
#[derive(Debug, Clone)]
pub struct BuildLogFull {
pub agent: String,
pub kind: String,
pub cmdline: String,
pub started_at: i64,
pub finished_at: Option<i64>,
pub status: Option<String>,
/// Elapsed seconds from `started_at` to `finished_at`. `None` while
/// the build is still in progress.
pub runtime_secs: Option<i64>,
}
/// Full row with stdout/stderr text inlined. Returned by `get_full`,
/// backs the side-panel viewer's payload.
#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct BuildLogFull {
#[serde(flatten)]
pub header: BuildLogHeader,
pub finished_at: Option<i64>,
pub stdout: String,
pub stderr: String,
}
/// Incremental text returned by `get_progress`. Carries only the new
/// bytes since the caller's last cursor positions so the SSE stream
/// handler can send deltas without re-transmitting the full log.
#[derive(Debug, Clone)]
pub struct BuildLogProgress {
/// New stdout bytes beyond `stdout_cursor`.
pub stdout_append: String,
/// New stderr bytes beyond `stderr_cursor`.
pub stderr_append: String,
/// `Some(unix_ts)` once the build is finished.
pub finished_at: Option<i64>,
/// Terminal status string (`"ok"` / `"fail"`) once finished.
pub status: Option<String>,
}
/// Channel capacity for per-build append notifications. 64 slots is
/// plenty — the consumer reads fast relative to line-append rate and
/// any lag means "read now, you have new content" rather than a lost
/// data line.
const NOTIFY_CAP: usize = 64;
/// Sqlite-backed build-log store. `Arc<BuildLogs>`-friendly: all
/// methods take `&self`, internal `Mutex<Connection>` serializes
/// access.
pub struct BuildLogs {
conn: Mutex<Connection>,
/// Broadcast channel that fires with the `id` of the row that just
/// had a line appended or was finished. The SSE stream handler
/// subscribes once per open panel and drives delta reads from this.
/// `send()` is non-async and silently drops frames when there are
/// no subscribers — safe to call from sync append/finish paths.
notify_tx: broadcast::Sender<i64>,
}
impl BuildLogs {
@ -171,10 +127,8 @@ impl BuildLogs {
conn.execute_batch(SCHEMA)
.context("apply build_logs schema")?;
crate::db::apply_versioned_migrations(&conn, "build_logs", MIGRATIONS)?;
let (notify_tx, _) = broadcast::channel(NOTIFY_CAP);
Ok(Self {
conn: Mutex::new(conn),
notify_tx,
})
}
@ -202,13 +156,6 @@ impl BuildLogs {
Ok(u64::try_from(closed).unwrap_or(0))
}
/// Subscribe to per-build append/finish notifications. Each emitted
/// value is the `id` of the row that changed. The SSE stream handler
/// calls this once and filters for its target id.
pub fn subscribe_notifications(&self) -> broadcast::Receiver<i64> {
self.notify_tx.subscribe()
}
/// Open a row for a new build attempt. Returns the assigned id
/// — the caller threads it through `append_stdout` / `append_stderr`
/// while the child runs and into `finish` once it exits.
@ -282,14 +229,6 @@ impl BuildLogs {
"build_logs: append failed (dropping line)"
);
}
drop(conn);
// Notify SSE stream subscribers — non-blocking, no-op when no
// subscribers are watching (e.g. no panel is open). Lagged
// receivers (channel full) automatically drop frames; the SSE
// handler re-reads the full delta on the next notification it
// does receive, so no content is lost, only an intermediate
// wake-up is coalesced.
let _ = self.notify_tx.send(id);
}
/// Finalize a build attempt. Sets `finished_at` to now and
@ -307,133 +246,30 @@ impl BuildLogs {
"build_logs: finish failed"
);
}
drop(conn);
// Final notification so the SSE stream handler sees the
// finished_at and status, closes the connection cleanly.
let _ = self.notify_tx.send(id);
}
/// Return incremental log content beyond the given byte cursors.
/// Used by the SSE stream handler to compute deltas between polls.
///
/// `stdout_cursor` / `stderr_cursor` are byte offsets into the
/// stored `stdout` / `stderr` columns from the previous read.
/// Slicing is safe because cursors are always derived from prior
/// `String::len()` values (valid UTF-8 boundaries).
///
/// Returns `None` when the row no longer exists (vacuum reap during
/// a long-open panel).
pub fn get_progress(
&self,
id: i64,
stdout_cursor: usize,
stderr_cursor: usize,
) -> Result<Option<BuildLogProgress>> {
let conn = self.conn.lock().unwrap();
let mut stmt = conn.prepare(
"SELECT stdout, stderr, finished_at, status \
FROM build_logs WHERE id = ?1",
)?;
let row = stmt
.query_row(params![id], |r| {
Ok((
r.get::<_, String>(0)?,
r.get::<_, String>(1)?,
r.get::<_, Option<i64>>(2)?,
r.get::<_, Option<String>>(3)?,
))
})
.optional()?;
match row {
None => Ok(None),
Some((stdout, stderr, finished_at, status)) => {
let stdout_append = stdout.get(stdout_cursor..).unwrap_or("").to_string();
let stderr_append = stderr.get(stderr_cursor..).unwrap_or("").to_string();
Ok(Some(BuildLogProgress {
stdout_append,
stderr_append,
finished_at,
status,
}))
}
}
}
/// Return the most recent `limit` rows for `agent`, newest first.
/// Headers only (no stdout/stderr blobs) — keeps `/api/state`
/// payloads light. Limit is hard-clamped to 50 to bound worst-case
/// payload regardless of caller input.
pub fn list_recent_for_agent(&self, agent: &str, limit: usize) -> Result<Vec<BuildLogHeader>> {
let limit = limit.min(50);
let conn = self.conn.lock().unwrap();
let mut stmt = conn.prepare(
"SELECT id, agent, kind, cmdline, started_at, finished_at, status
FROM build_logs
WHERE agent = ?1
ORDER BY started_at DESC
LIMIT ?2",
)?;
let rows = stmt.query_map(
params![agent, i64::try_from(limit).unwrap_or(50)],
row_to_header,
)?;
let mut out = Vec::new();
for r in rows {
out.push(r?);
}
Ok(out)
}
/// Return the most recent `limit` rows across all agents, newest first.
/// Same header-only shape as `list_recent_for_agent`. Limit clamped to 100.
pub fn list_recent_all(&self, limit: usize) -> Result<Vec<BuildLogHeader>> {
let limit = limit.min(100);
let conn = self.conn.lock().unwrap();
let mut stmt = conn.prepare(
"SELECT id, agent, kind, cmdline, started_at, finished_at, status
FROM build_logs
ORDER BY started_at DESC
LIMIT ?1",
)?;
let rows = stmt.query_map(params![i64::try_from(limit).unwrap_or(100)], row_to_header)?;
let mut out = Vec::new();
for r in rows {
out.push(r?);
}
Ok(out)
}
/// Fetch a single full row (with stdout/stderr text) by id.
/// Returns `None` when the id doesn't exist (vacuum sweep already
/// reaped it, or the operator passed a stale id from a refresh
/// race).
/// Read one row back, text included. `None` when the id doesn't exist
/// (already vacuumed).
#[cfg(test)]
pub fn get_full(&self, id: i64) -> Result<Option<BuildLogFull>> {
use rusqlite::OptionalExtension;
let conn = self.conn.lock().unwrap();
let mut stmt = conn.prepare(
"SELECT id, agent, kind, cmdline, started_at, finished_at, status, stdout, stderr
FROM build_logs
WHERE id = ?1",
)?;
let row = stmt
.query_row(params![id], |r| {
let started_at: i64 = r.get(4)?;
let finished_at: Option<i64> = r.get(5)?;
let runtime_secs = compute_runtime_secs(started_at, finished_at);
Ok(BuildLogFull {
header: BuildLogHeader {
id: r.get(0)?,
agent: r.get(1)?,
kind: r.get(2)?,
cmdline: r.get(3)?,
started_at,
finished_at,
status: r.get(6)?,
runtime_secs,
},
stdout: r.get(7)?,
stderr: r.get(8)?,
})
})
let row = conn
.query_row(
"SELECT agent, kind, finished_at, status, stdout, stderr \
FROM build_logs WHERE id = ?1",
params![id],
|r| {
Ok(BuildLogFull {
agent: r.get(0)?,
kind: r.get(1)?,
finished_at: r.get(2)?,
status: r.get(3)?,
stdout: r.get(4)?,
stderr: r.get(5)?,
})
},
)
.optional()?;
Ok(row)
}
@ -494,26 +330,6 @@ pub fn spawn_vacuum(coord: &Arc<crate::coordinator::Coordinator>) {
});
}
fn compute_runtime_secs(started_at: i64, finished_at: Option<i64>) -> Option<i64> {
finished_at.map(|f| f - started_at)
}
fn row_to_header(r: &rusqlite::Row) -> rusqlite::Result<BuildLogHeader> {
let started_at: i64 = r.get(4)?;
let finished_at: Option<i64> = r.get(5)?;
let runtime_secs = compute_runtime_secs(started_at, finished_at);
Ok(BuildLogHeader {
id: r.get(0)?,
agent: r.get(1)?,
kind: r.get(2)?,
cmdline: r.get(3)?,
started_at,
finished_at,
status: r.get(6)?,
runtime_secs,
})
}
#[cfg(test)]
mod tests {
use super::*;
@ -536,10 +352,10 @@ mod tests {
db.finish(id, BuildStatus::Fail);
let full = db.get_full(id).expect("get").expect("Some");
assert_eq!(full.header.agent, "alice");
assert_eq!(full.header.kind, "prebuild");
assert_eq!(full.header.status.as_deref(), Some("fail"));
assert!(full.header.finished_at.is_some());
assert_eq!(full.agent, "alice");
assert_eq!(full.kind, "prebuild");
assert_eq!(full.status.as_deref(), Some("fail"));
assert!(full.finished_at.is_some());
assert!(full.stdout.contains("/nix/store/abc.drv"));
assert!(full.stderr.contains("error: line 12"));
assert!(full.stderr.contains("module.nix:5"));
@ -565,46 +381,15 @@ mod tests {
assert_eq!(db.reap_orphans().expect("reap"), 1);
let row = db.get_full(orphan).expect("get").expect("Some");
assert_eq!(row.header.status.as_deref(), Some("fail"));
assert!(row.header.finished_at.is_some());
assert_eq!(row.status.as_deref(), Some("fail"));
assert!(row.finished_at.is_some());
assert!(row.stdout.contains("building"));
assert!(row.stderr.ends_with(&format!("{ORPHAN_NOTE}\n")));
let ok = db.get_full(done).expect("get").expect("Some");
assert_eq!(ok.header.status.as_deref(), Some("ok"));
assert_eq!(ok.status.as_deref(), Some("ok"));
assert_eq!(db.reap_orphans().expect("reap again"), 0);
}
#[test]
fn list_recent_orders_newest_first_and_clamps() {
let (_d, db) = tmpdb();
// Three attempts, two for alice + one for bob. Without sleeping
// sqlite's `INTEGER` started_at ties at 1-sec resolution, so we
// assert id-ordering (autoincrement) is the tiebreaker — list
// sorts by started_at DESC but the ORDER BY still produces the
// last-inserted row first when timestamps match.
let id_a1 = db.start("alice", "run", "cmd one", None).expect("start");
let _id_b = db.start("bob", "run", "cmd two", None).expect("start");
let id_a2 = db.start("alice", "run", "cmd three", None).expect("start");
db.finish(id_a1, BuildStatus::Ok);
let alice_rows = db.list_recent_for_agent("alice", 10).expect("list");
assert_eq!(alice_rows.len(), 2);
// Without distinct started_at values both rows share `now`,
// but list_recent already orders by `started_at DESC` then
// sqlite's natural insertion-order tiebreak. We rely only on
// both IDs being present + correct count + agent isolation.
let ids: std::collections::HashSet<i64> = alice_rows.iter().map(|h| h.id).collect();
assert!(ids.contains(&id_a1));
assert!(ids.contains(&id_a2));
let bob_rows = db.list_recent_for_agent("bob", 10).expect("list");
assert_eq!(bob_rows.len(), 1);
// Limit clamp at 50.
let huge = db.list_recent_for_agent("alice", 999_999).expect("list");
assert!(huge.len() <= 50);
}
#[test]
fn get_full_returns_none_for_missing_id() {
let (_d, db) = tmpdb();