1f214f9f01
hive-c0re's admin socket and hive-agent's in-agent socket (mode 0666) read the request with an unbounded read_line: no length cap and no deadline, so a peer that never sends a newline grows a buffer and pins a task forever. The bounded reader (16 MiB cap) moves from hive-c0re's socket_server into hive-sock-client, the one crate both daemons already share or can take, and gains optional deadlines for the first byte and for the rest of the line. Both sockets now use it and answer an over-long or stalled line with the same error reply the bounded sockets give, then close. The admin socket bounds only the started line (clients may idle between requests); the in-agent socket is single-shot and bounds silence too. The per-agent sockets keep their behaviour (no deadline).
680 lines
25 KiB
Rust
680 lines
25 KiB
Rust
//! In-agent socket server (loose-ends v2 + harness-local reminders +
|
|
//! self-service compact). Binds the harness-owned `HIVE_AGENT_SOCKET` and
|
|
//! serves the `hive-agent-sock` protocol to the in-container producers —
|
|
//! matrix / bash daemons and forge-notify are the built-in ones, but any
|
|
//! user-configured MCP server can dial the same socket and push its own
|
|
//! todos — and to `hive-agent-mcp`'s `remind`/`get_loose_ends`/
|
|
//! `cancel_loose_end`/`compact` tool impls. Todo ops hit the harness-local
|
|
//! [`Todos`] store; a new-or-changed upsert fires an in-process [`Notify`]
|
|
//! so the serve loop drives a turn. Reminder ops hit the harness-local
|
|
//! [`Reminders`] store (`None` when the store failed to open — every
|
|
//! reminder op then returns `Response::Err`); a reminder *firing* is a
|
|
//! separate path (`reminder_timer`), not driven through this socket.
|
|
//! `Request::Compact` is the odd one out — it doesn't touch any store, just
|
|
//! the harness's [`Bus`] (gate-checked context usage, then the same
|
|
//! deferred `compact_pending` flag the operator dashboard's `/compact`
|
|
//! button sets). No hive-c0re round-trip, no broker long-poll, no marker
|
|
//! files.
|
|
//!
|
|
//! One request/response line per connection, matching the producers'
|
|
//! existing best-effort JSON-line clients (they just change which socket
|
|
//! they dial, not the payload).
|
|
|
|
use std::os::unix::fs::PermissionsExt;
|
|
use std::path::{Path, PathBuf};
|
|
use std::sync::Arc;
|
|
|
|
use anyhow::{Context, Result};
|
|
use hive_agent_sock::{Request, Response};
|
|
use hive_sh4re::inbox::{LooseEnd, saturating_age};
|
|
use hive_sock_client::{LineDeadlines, REQUEST_LINE_DEADLINE, RequestLine, read_request_line};
|
|
use tokio::io::{AsyncWriteExt, BufReader};
|
|
use tokio::net::{UnixListener, UnixStream};
|
|
use tokio::sync::Notify;
|
|
|
|
use crate::events::Bus;
|
|
use crate::reminders::{Reminder, Reminders};
|
|
use crate::todos::{Todo, Todos};
|
|
|
|
/// Context-usage floor for `Request::Compact`: below this fraction of the
|
|
/// effective context window, an agent-initiated compact is refused (nothing
|
|
/// meaningful to reclaim yet, and it'd just burn a compaction pass on a
|
|
/// near-empty session). Same number a human operator would eyeball off the
|
|
/// dashboard's `ctx·Nk` badge before hitting the `/compact` button.
|
|
const COMPACT_MIN_USAGE_FRACTION: f64 = 0.66;
|
|
|
|
/// Resolve the in-agent socket path from `HIVE_AGENT_SOCKET`. `None` when
|
|
/// unset/empty (e.g. a standalone dev run) — the server then stays off.
|
|
fn socket_path() -> Option<PathBuf> {
|
|
std::env::var_os("HIVE_AGENT_SOCKET")
|
|
.filter(|v| !v.is_empty())
|
|
.map(PathBuf::from)
|
|
}
|
|
|
|
/// Dial this same in-agent socket from elsewhere IN THIS PROCESS — the
|
|
/// harness's own `web_ui` endpoints (`api_todos`, `/api/stats` reminder
|
|
/// rollup) and `Surface::post_turn_counts` all need read access to the
|
|
/// `Todos`/`Reminders` stores this module owns behind an `Arc` on a
|
|
/// different spawned task, and a loopback dial is simpler than threading
|
|
/// those `Arc`s through every caller. One-shot, best-effort (no retry,
|
|
/// unlike the broker client): a connect failure means the socket server
|
|
/// task itself isn't up, which a retry within one call wouldn't fix.
|
|
pub(crate) async fn dial(req: &Request) -> Option<Response> {
|
|
let path = socket_path()?;
|
|
if !path.exists() {
|
|
return None;
|
|
}
|
|
tokio::time::timeout(
|
|
std::time::Duration::from_secs(3),
|
|
hive_sock_client::request::<_, Response>(&path, req, hive_sock_client::Retry::None),
|
|
)
|
|
.await
|
|
.ok()?
|
|
.ok()
|
|
}
|
|
|
|
/// Run the in-agent socket server: bind + accept loop, one request/response
|
|
/// line per connection. A no-op (returns `Ok`) when `HIVE_AGENT_SOCKET` is
|
|
/// unset, so a standalone harness without producers just skips it.
|
|
///
|
|
/// # Errors
|
|
///
|
|
/// Returns an error if the socket path is set but can't be bound.
|
|
pub async fn run(
|
|
store: Arc<Todos>,
|
|
wake: Arc<Notify>,
|
|
reminders: Option<Arc<Reminders>>,
|
|
bus: Bus,
|
|
) -> Result<()> {
|
|
let Some(path) = socket_path() else {
|
|
tracing::info!("HIVE_AGENT_SOCKET unset — in-agent todo socket disabled");
|
|
return Ok(());
|
|
};
|
|
let listener = bind(&path)?;
|
|
tracing::info!(socket = %path.display(), "in-agent todo socket listening");
|
|
loop {
|
|
match listener.accept().await {
|
|
Ok((stream, _)) => {
|
|
let store = store.clone();
|
|
let wake = wake.clone();
|
|
let reminders = reminders.clone();
|
|
let bus = bus.clone();
|
|
tokio::spawn(async move {
|
|
if let Err(e) =
|
|
handle_conn(stream, &store, &wake, reminders.as_deref(), &bus).await
|
|
{
|
|
tracing::warn!(error = ?e, "in-agent todo connection failed");
|
|
}
|
|
});
|
|
}
|
|
Err(e) => tracing::warn!(error = ?e, "in-agent todo accept failed"),
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Bind a `UnixListener` at `path`, creating the parent dir and unlinking a
|
|
/// stale socket left by a prior boot (which would otherwise block `bind`
|
|
/// with `EADDRINUSE`).
|
|
///
|
|
/// The mode is set **here, at creation** — not corrected afterwards by
|
|
/// whoever notices. `connect(2)` on a unix socket requires *write*
|
|
/// permission, and `bind` leaves `0777 & ~umask` (typically `0755`), which
|
|
/// silently excludes every uid but the harness's own. `hive-c0re` pushes
|
|
/// todos in over this socket from the host, so that default locks it out.
|
|
/// Same two lines the sibling `web.sock` has carried all along
|
|
/// (`web_ui::serve`); access control is the containing directory's job, not
|
|
/// the socket's.
|
|
fn bind(path: &Path) -> Result<UnixListener> {
|
|
if let Some(parent) = path.parent() {
|
|
std::fs::create_dir_all(parent)
|
|
.with_context(|| format!("create in-agent socket dir {}", parent.display()))?;
|
|
}
|
|
if path.exists() {
|
|
let _ = std::fs::remove_file(path);
|
|
}
|
|
let listener = UnixListener::bind(path)
|
|
.with_context(|| format!("bind in-agent socket {}", path.display()))?;
|
|
std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o666))
|
|
.with_context(|| format!("set perms on in-agent socket {}", path.display()))?;
|
|
Ok(listener)
|
|
}
|
|
|
|
/// Handle one connection: read a single JSON request line, apply it to the
|
|
/// store, write the JSON response line back.
|
|
async fn handle_conn(
|
|
stream: UnixStream,
|
|
store: &Todos,
|
|
wake: &Notify,
|
|
reminders: Option<&Reminders>,
|
|
bus: &Bus,
|
|
) -> Result<()> {
|
|
serve_one(stream, REQUEST_DEADLINES, |req| {
|
|
dispatch(req, store, wake, reminders, bus)
|
|
})
|
|
.await
|
|
}
|
|
|
|
/// One request line, one response line. A single-shot connection, so silence
|
|
/// is bounded too: the socket is mode 0666 and any local process can open it.
|
|
const REQUEST_DEADLINES: LineDeadlines = LineDeadlines {
|
|
idle: Some(REQUEST_LINE_DEADLINE),
|
|
line: Some(REQUEST_LINE_DEADLINE),
|
|
};
|
|
|
|
/// Read the request line under `deadlines`, answer it with `apply`, and write
|
|
/// the response. An over-long or stalled line gets a `Response::Err`.
|
|
async fn serve_one(
|
|
stream: UnixStream,
|
|
deadlines: LineDeadlines,
|
|
apply: impl FnOnce(Request) -> Response,
|
|
) -> Result<()> {
|
|
let (read, mut write) = stream.into_split();
|
|
let mut reader = BufReader::new(read);
|
|
let mut line = Vec::new();
|
|
let outcome = read_request_line(&mut reader, &mut line, deadlines).await?;
|
|
if outcome == RequestLine::Eof {
|
|
return Ok(());
|
|
}
|
|
let resp = if let Some(message) = outcome.refusal() {
|
|
Response::Err { message }
|
|
} else {
|
|
match serde_json::from_slice::<Request>(line.trim_ascii()) {
|
|
Ok(req) => apply(req),
|
|
Err(e) => Response::Err {
|
|
message: format!("bad request: {e}"),
|
|
},
|
|
}
|
|
};
|
|
let mut out = serde_json::to_string(&resp)?;
|
|
out.push('\n');
|
|
write.write_all(out.as_bytes()).await?;
|
|
write.flush().await?;
|
|
Ok(())
|
|
}
|
|
|
|
/// Apply one request to the store, firing `wake` on a new/changed upsert so
|
|
/// the serve loop runs a turn. Each arm calls a small named handler function
|
|
/// directly — no sub-match/`unreachable!()` indirection per family (that
|
|
/// pattern got reviewed out of the todo family in the diagnostic-logging
|
|
/// follow-up PR; kept the reminder family consistent with it here rather
|
|
/// than reintroducing it). `reminders` is `None` when that store failed to
|
|
/// open at boot, in which case every reminder op returns an `Err` — each
|
|
/// handler checks for its own `None` case.
|
|
fn dispatch(
|
|
req: Request,
|
|
store: &Todos,
|
|
wake: &Notify,
|
|
reminders: Option<&Reminders>,
|
|
bus: &Bus,
|
|
) -> Response {
|
|
match req {
|
|
Request::UpsertTodo {
|
|
subsystem,
|
|
key,
|
|
summary,
|
|
source,
|
|
reopen_if_acked,
|
|
} => upsert_todo(
|
|
store,
|
|
wake,
|
|
&subsystem,
|
|
key.as_deref(),
|
|
&summary,
|
|
source.as_deref(),
|
|
reopen_if_acked,
|
|
),
|
|
Request::ClearTodo {
|
|
subsystem,
|
|
key,
|
|
all,
|
|
} => clear_todo(store, &subsystem, key.as_deref(), all),
|
|
Request::ListTodos { subsystem } => list_todos(store, subsystem.as_deref()),
|
|
Request::MarkTodoDone { id } => mark_todo_done(store, id),
|
|
Request::MarkTodosDone { ids } => mark_todos_done(store, &ids),
|
|
Request::StoreReminder {
|
|
message,
|
|
timing,
|
|
file_path,
|
|
} => store_reminder(reminders, &message, &timing, file_path.as_deref()),
|
|
Request::ListReminders => list_reminders(reminders),
|
|
Request::CancelReminder { id } => cancel_reminder(reminders, id),
|
|
Request::CountPendingReminders => count_pending_reminders(reminders),
|
|
Request::ReminderRollup { since_secs } => reminder_rollup(reminders, since_secs),
|
|
Request::Compact { wake_prompt } => compact(bus, wake_prompt),
|
|
}
|
|
}
|
|
|
|
/// `StoreReminder` handler: persists a reminder due at `timing`.
|
|
fn store_reminder(
|
|
reminders: Option<&Reminders>,
|
|
message: &str,
|
|
timing: &hive_sh4re::inbox::ReminderTiming,
|
|
file_path: Option<&str>,
|
|
) -> Response {
|
|
let Some(r) = reminders else {
|
|
return no_reminders_store();
|
|
};
|
|
match crate::reminder_timer::store(r, message, timing, file_path) {
|
|
Ok(_id) => Response::Ok,
|
|
Err(message) => Response::Err { message },
|
|
}
|
|
}
|
|
|
|
/// `ListReminders` handler: this agent's own pending reminders.
|
|
fn list_reminders(reminders: Option<&Reminders>) -> Response {
|
|
let Some(r) = reminders else {
|
|
return no_reminders_store();
|
|
};
|
|
match r.list_pending() {
|
|
Ok(rows) => Response::LooseEnds {
|
|
loose_ends: rows.into_iter().map(reminder_to_loose_end).collect(),
|
|
},
|
|
Err(e) => err(&e),
|
|
}
|
|
}
|
|
|
|
/// `CancelReminder` handler: drops one pending reminder by id.
|
|
fn cancel_reminder(reminders: Option<&Reminders>, id: i64) -> Response {
|
|
let Some(r) = reminders else {
|
|
return no_reminders_store();
|
|
};
|
|
match r.cancel(id) {
|
|
Ok(count) => Response::Acked {
|
|
count: u64::try_from(count).unwrap_or(0),
|
|
},
|
|
Err(e) => err(&e),
|
|
}
|
|
}
|
|
|
|
/// `CountPendingReminders` handler.
|
|
fn count_pending_reminders(reminders: Option<&Reminders>) -> Response {
|
|
let Some(r) = reminders else {
|
|
return no_reminders_store();
|
|
};
|
|
match r.count_pending() {
|
|
Ok(count) => Response::PendingRemindersCount { count },
|
|
Err(e) => err(&e),
|
|
}
|
|
}
|
|
|
|
/// `ReminderRollup` handler: scheduled/delivered/pending counts over a
|
|
/// trailing `since_secs` window (`0` = all time).
|
|
fn reminder_rollup(reminders: Option<&Reminders>, since_secs: u64) -> Response {
|
|
let Some(r) = reminders else {
|
|
return no_reminders_store();
|
|
};
|
|
let since_secs = i64::try_from(since_secs).unwrap_or(i64::MAX);
|
|
match r.rollup(since_secs) {
|
|
Ok(stats) => Response::ReminderRollup { stats },
|
|
Err(e) => err(&e),
|
|
}
|
|
}
|
|
|
|
/// `UpsertTodo` handler: writes/refreshes a todo row, logs the outcome, and
|
|
/// fires `wake` on a new-or-changed upsert so the serve loop runs a turn.
|
|
fn upsert_todo(
|
|
store: &Todos,
|
|
wake: &Notify,
|
|
subsystem: &str,
|
|
key: Option<&str>,
|
|
summary: &str,
|
|
source: Option<&str>,
|
|
reopen_if_acked: bool,
|
|
) -> Response {
|
|
match store.upsert(subsystem, key, summary, source, reopen_if_acked) {
|
|
Ok((id, changed)) => {
|
|
tracing::debug!(%subsystem, ?key, id, changed, "todo upsert");
|
|
if changed {
|
|
wake.notify_one();
|
|
}
|
|
Response::Ok
|
|
}
|
|
Err(e) => err(&e),
|
|
}
|
|
}
|
|
|
|
/// `ClearTodo` handler: drops one keyed todo, or every todo in `subsystem`
|
|
/// when `all` is set.
|
|
fn clear_todo(store: &Todos, subsystem: &str, key: Option<&str>, all: bool) -> Response {
|
|
let result = if all {
|
|
store.clear_subsystem(subsystem)
|
|
} else {
|
|
store.clear(subsystem, key)
|
|
};
|
|
match result {
|
|
Ok(count) => {
|
|
tracing::debug!(%subsystem, ?key, all, count, "todo clear");
|
|
Response::Acked {
|
|
count: u64::try_from(count).unwrap_or(0),
|
|
}
|
|
}
|
|
Err(e) => err(&e),
|
|
}
|
|
}
|
|
|
|
/// `ListTodos` handler: read-only, so no debug logging — not relevant to
|
|
/// diagnosing wake behaviour.
|
|
fn list_todos(store: &Todos, subsystem: Option<&str>) -> Response {
|
|
match store.list(subsystem) {
|
|
Ok(todos) => Response::LooseEnds {
|
|
loose_ends: todos.into_iter().map(to_loose_end).collect(),
|
|
},
|
|
Err(e) => err(&e),
|
|
}
|
|
}
|
|
|
|
/// `MarkTodoDone` handler: marks a single todo done by id.
|
|
fn mark_todo_done(store: &Todos, id: i64) -> Response {
|
|
match store.mark_done(id) {
|
|
Ok(count) => {
|
|
tracing::debug!(id, count, "todo mark-done");
|
|
Response::Acked {
|
|
count: u64::try_from(count).unwrap_or(0),
|
|
}
|
|
}
|
|
Err(e) => err(&e),
|
|
}
|
|
}
|
|
|
|
/// `MarkTodosDone` handler: bulk-acks an explicit list of todo ids — the
|
|
/// escape hatch for a backlog too large to triage one-by-one (see
|
|
/// `Todos::mark_done_many`'s doc for why it's ids, not a threshold).
|
|
fn mark_todos_done(store: &Todos, ids: &[i64]) -> Response {
|
|
match store.mark_done_many(ids) {
|
|
Ok(count) => {
|
|
tracing::debug!(n_ids = ids.len(), count, "todo bulk mark-done");
|
|
Response::Acked {
|
|
count: u64::try_from(count).unwrap_or(0),
|
|
}
|
|
}
|
|
Err(e) => err(&e),
|
|
}
|
|
}
|
|
|
|
/// `Request::Compact` handler: gate on context usage, then queue the same
|
|
/// deferred `compact_pending` flag the operator's `/compact` button sets.
|
|
/// Mirrors `hive-agent::web_ui::actions::post_compact` but reachable from
|
|
/// the agent's own MCP tool instead of the dashboard, and refuses below
|
|
/// [`COMPACT_MIN_USAGE_FRACTION`] instead of always honouring the request —
|
|
/// an agent can call this speculatively, a human clicking the dashboard
|
|
/// button already made the judgment call. `wake_prompt`, when set, is
|
|
/// forwarded to `Bus::request_compact` so the turn loop drives a synthetic
|
|
/// follow-up turn once the compaction actually finishes.
|
|
fn compact(bus: &Bus, wake_prompt: Option<String>) -> Response {
|
|
let Some(usage) = bus.last_ctx_usage() else {
|
|
return Response::Err {
|
|
message: "compact refused: no completed turn yet — nothing to compact".to_owned(),
|
|
};
|
|
};
|
|
let model = bus.model();
|
|
let window = bus.effective_context_window(&model);
|
|
if window == 0 {
|
|
return Response::Err {
|
|
message: "compact refused: effective context window is unknown (0)".to_owned(),
|
|
};
|
|
}
|
|
// Token counts stay well under 2^53 in practice, so the f64 conversion
|
|
// is exact; this is a threshold ratio, not a byte-count display, but the
|
|
// same "cosmetic precision loss" reasoning as `hivectl::quota::human_bytes` applies.
|
|
#[allow(clippy::cast_precision_loss)]
|
|
let fraction = usage.context_tokens() as f64 / window as f64;
|
|
if fraction < COMPACT_MIN_USAGE_FRACTION {
|
|
return Response::Err {
|
|
message: format!(
|
|
"compact refused: context usage is {:.0}% of the {window}-token window, \
|
|
below the {:.0}% floor — not worth compacting yet",
|
|
fraction * 100.0,
|
|
COMPACT_MIN_USAGE_FRACTION * 100.0
|
|
),
|
|
};
|
|
}
|
|
let will_wake = wake_prompt.is_some();
|
|
bus.request_compact(wake_prompt);
|
|
bus.emit(crate::events::LiveEvent::Note {
|
|
text: if will_wake {
|
|
"agent: self-requested /compact (with wake prompt) — running at the end of the \
|
|
current turn"
|
|
.into()
|
|
} else {
|
|
"agent: self-requested /compact — running at the end of the current turn".into()
|
|
},
|
|
});
|
|
Response::Ok
|
|
}
|
|
|
|
/// Shared "reminders db unavailable" response for every reminder op when
|
|
/// the store failed to open at boot (see `main.rs`'s best-effort open).
|
|
fn no_reminders_store() -> Response {
|
|
Response::Err {
|
|
message: "reminders store unavailable on this agent".to_owned(),
|
|
}
|
|
}
|
|
|
|
/// Map a stored [`Reminder`] to a [`LooseEnd::Reminder`], deriving
|
|
/// `age_seconds` from `created_at` (mirrors the old c0re rendering —
|
|
/// "age" is how long the reminder has been *scheduled*, not how soon
|
|
/// it's due).
|
|
fn reminder_to_loose_end(r: Reminder) -> LooseEnd {
|
|
let now = chrono::Utc::now().timestamp();
|
|
LooseEnd::Reminder {
|
|
id: r.id,
|
|
owner: crate::identity::label(),
|
|
message: r.message,
|
|
due_at: r.due_at,
|
|
age_seconds: saturating_age(now, r.created_at.timestamp()),
|
|
}
|
|
}
|
|
|
|
/// Format a store error into an `Err` response.
|
|
fn err(e: &anyhow::Error) -> Response {
|
|
Response::Err {
|
|
message: format!("{e:#}"),
|
|
}
|
|
}
|
|
|
|
/// Map a stored [`Todo`] to a [`LooseEnd::Todo`], deriving `age_seconds`
|
|
/// from `updated_at` via [`saturating_age`].
|
|
fn to_loose_end(t: Todo) -> LooseEnd {
|
|
let now = chrono::Utc::now().timestamp();
|
|
LooseEnd::Todo {
|
|
id: t.id,
|
|
subsystem: t.subsystem,
|
|
subsystem_key: t.subsystem_key,
|
|
summary: t.summary,
|
|
source: t.source,
|
|
age_seconds: saturating_age(now, t.updated_at.timestamp()),
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
use chrono::{DateTime, Duration, Utc};
|
|
use hive_sock_client::REQUEST_LINE_MAX_BYTES;
|
|
use tokio::io::AsyncBufReadExt;
|
|
|
|
/// `saturating_age`'s own arms are exact (`hive-sh4re::inbox`); what
|
|
/// is only checkable here is *which* stored timestamp each projection
|
|
/// feeds it, and both read the clock internally — so these assert a
|
|
/// range wide enough that a slow runner can't fail them.
|
|
const SLACK: u64 = 600;
|
|
|
|
/// Not asserted anywhere below: `LooseEnd::Reminder.owner`. It comes
|
|
/// from `identity::label()`, which reads a process-global env var
|
|
/// that `identity`'s own tests set and restore — in the same test
|
|
/// binary, in parallel. Any value assertion here would be a race, not
|
|
/// a check.
|
|
fn reminder(created_at: DateTime<Utc>, due_at: DateTime<Utc>) -> Reminder {
|
|
Reminder {
|
|
id: 7,
|
|
message: "check the gate".to_owned(),
|
|
file_path: None,
|
|
due_at,
|
|
created_at,
|
|
}
|
|
}
|
|
|
|
fn todo(updated_at: DateTime<Utc>) -> Todo {
|
|
Todo {
|
|
id: 11,
|
|
subsystem: "matrix".to_owned(),
|
|
subsystem_key: Some("!room:x".to_owned()),
|
|
summary: "3 unread".to_owned(),
|
|
source: Some("#ops".to_owned()),
|
|
updated_at,
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn a_reminder_ages_from_created_at_not_from_when_it_is_due() {
|
|
// Scheduled 600s ago, due an hour out. Reading the age off
|
|
// `due_at` would clamp to 0, so the two fields are separable.
|
|
let r = reminder(
|
|
Utc::now() - Duration::seconds(600),
|
|
Utc::now() + Duration::seconds(3600),
|
|
);
|
|
let due = r.due_at;
|
|
match reminder_to_loose_end(r) {
|
|
LooseEnd::Reminder {
|
|
id,
|
|
message,
|
|
due_at,
|
|
age_seconds,
|
|
..
|
|
} => {
|
|
assert_eq!(id, 7);
|
|
assert_eq!(message, "check the gate");
|
|
assert_eq!(due_at, due, "due_at passes through untouched");
|
|
assert!(
|
|
(600..600 + SLACK).contains(&age_seconds),
|
|
"age {age_seconds} should track created_at, not due_at"
|
|
);
|
|
}
|
|
other => panic!("expected a Reminder loose end, got {other:?}"),
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn a_todo_ages_from_updated_at_and_carries_its_dedup_key() {
|
|
let t = todo(Utc::now() - Duration::seconds(600));
|
|
match to_loose_end(t) {
|
|
LooseEnd::Todo {
|
|
id,
|
|
subsystem,
|
|
subsystem_key,
|
|
summary,
|
|
source,
|
|
age_seconds,
|
|
} => {
|
|
assert_eq!(id, 11);
|
|
assert_eq!(subsystem, "matrix");
|
|
// Distinct values: the two `Option<String>` fields are
|
|
// adjacent and same-typed, so a swap has to be visible.
|
|
assert_eq!(subsystem_key.as_deref(), Some("!room:x"));
|
|
assert_eq!(source.as_deref(), Some("#ops"));
|
|
assert_eq!(summary, "3 unread");
|
|
assert!((600..600 + SLACK).contains(&age_seconds));
|
|
}
|
|
other => panic!("expected a Todo loose end, got {other:?}"),
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn a_timestamp_in_the_future_reads_as_zero_through_both_projections() {
|
|
// Not a hypothetical: these stores are written by in-container
|
|
// producers, and `updated_at` crossing `now` is exactly what a
|
|
// clock sync during a turn produces.
|
|
let ahead = Utc::now() + Duration::seconds(3600);
|
|
let behind = Utc::now() - Duration::seconds(600);
|
|
|
|
let (future_r, past_r) = (age_of_reminder(ahead), age_of_reminder(behind));
|
|
let (future_t, past_t) = (age_of_todo(ahead), age_of_todo(behind));
|
|
|
|
assert_eq!(future_r, 0);
|
|
assert_eq!(future_t, 0);
|
|
// Controls: same call, past timestamp — so the zeros above are
|
|
// the clamp firing, not the projections always emitting 0.
|
|
assert!(past_r >= 600, "reminder control read {past_r}");
|
|
assert!(past_t >= 600, "todo control read {past_t}");
|
|
}
|
|
|
|
fn age_of_reminder(created_at: DateTime<Utc>) -> u64 {
|
|
match reminder_to_loose_end(reminder(created_at, Utc::now())) {
|
|
LooseEnd::Reminder { age_seconds, .. } => age_seconds,
|
|
other => panic!("expected a Reminder loose end, got {other:?}"),
|
|
}
|
|
}
|
|
|
|
fn age_of_todo(updated_at: DateTime<Utc>) -> u64 {
|
|
match to_loose_end(todo(updated_at)) {
|
|
LooseEnd::Todo { age_seconds, .. } => age_seconds,
|
|
other => panic!("expected a Todo loose end, got {other:?}"),
|
|
}
|
|
}
|
|
|
|
/// Send `input` (then keep the write half open), return the one response line.
|
|
async fn exchange(input: Vec<u8>, deadlines: LineDeadlines) -> (Response, Result<()>) {
|
|
let (client, server) = UnixStream::pair().expect("socketpair");
|
|
let handler = tokio::spawn(serve_one(server, deadlines, |_| Response::Ok));
|
|
let (read, mut write) = client.into_split();
|
|
let _writer = tokio::spawn(async move {
|
|
let _ = write.write_all(&input).await;
|
|
std::future::pending::<()>().await;
|
|
});
|
|
let mut reply = String::new();
|
|
BufReader::new(read)
|
|
.read_line(&mut reply)
|
|
.await
|
|
.expect("reply");
|
|
let resp = serde_json::from_str(&reply).expect("response json");
|
|
(resp, handler.await.expect("join"))
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn a_valid_request_is_applied() {
|
|
let (resp, done) = exchange(b"{\"cmd\":\"compact\"}\n".to_vec(), REQUEST_DEADLINES).await;
|
|
assert!(matches!(resp, Response::Ok), "{resp:?}");
|
|
done.expect("served");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn an_over_long_line_gets_an_error_not_an_unbounded_buffer() {
|
|
let (resp, done) =
|
|
exchange(vec![b'x'; REQUEST_LINE_MAX_BYTES + 1], REQUEST_DEADLINES).await;
|
|
assert!(
|
|
matches!(&resp, Response::Err { message } if message.starts_with("request too long")),
|
|
"{resp:?}"
|
|
);
|
|
done.expect("served");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn a_sender_that_stalls_mid_line_times_out() {
|
|
let short = std::time::Duration::from_millis(50);
|
|
let deadlines = LineDeadlines {
|
|
idle: Some(short),
|
|
line: Some(short),
|
|
};
|
|
let (resp, done) = exchange(b"{\"cmd\"".to_vec(), deadlines).await;
|
|
assert!(
|
|
matches!(&resp, Response::Err { message } if message.starts_with("request timed out")),
|
|
"{resp:?}"
|
|
);
|
|
done.expect("served");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn a_sender_that_never_writes_times_out() {
|
|
let short = std::time::Duration::from_millis(50);
|
|
let deadlines = LineDeadlines {
|
|
idle: Some(short),
|
|
line: Some(short),
|
|
};
|
|
let (resp, done) = exchange(Vec::new(), deadlines).await;
|
|
assert!(
|
|
matches!(&resp, Response::Err { message } if message.starts_with("request timed out")),
|
|
"{resp:?}"
|
|
);
|
|
done.expect("served");
|
|
}
|
|
}
|