Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
02d458cdf5 | ||
|
|
90c72d9131 | ||
|
|
5e0cb5e0f5 |
4 changed files with 177 additions and 8 deletions
|
|
@ -67,6 +67,43 @@ Per-vhost timeouts + body-size limits live in the location blocks:
|
||||||
|
|
||||||
SSH for forge stays direct on `cfg.sshPort` — separate listener protocol, not HTTP-over-nginx.
|
SSH for forge stays direct on `cfg.sshPort` — separate listener protocol, not HTTP-over-nginx.
|
||||||
|
|
||||||
|
## Per-agent unix-socket upstream (#784)
|
||||||
|
|
||||||
|
Sub-agent `/agent/<name>/` upstreams flip from TCP loopback to a
|
||||||
|
unix-domain socket as each agent opts in. The mechanism:
|
||||||
|
|
||||||
|
1. **Agent side** (`hyperhive.web.useUnixSocket = true` in
|
||||||
|
`agent.nix`, #815). Sets `HIVE_WEB_SOCKET=/run/hive-agent/<name>/web.sock`
|
||||||
|
on the harness service env; `web_ui::serve` binds a `UnixListener`
|
||||||
|
at that path instead of TCP.
|
||||||
|
2. **Host side**. `hive-c0re` bind-mounts the per-agent subdir
|
||||||
|
(`/run/hive-agent/<name>/`) into the agent's container (#813). Dir
|
||||||
|
bind, not file bind — file bind-mounts don't survive the
|
||||||
|
harness's `unlink + bind(2)` cycle on socket replace. Per-agent
|
||||||
|
subdir keeps each agent's container blind to siblings'
|
||||||
|
sockets (mara on #800).
|
||||||
|
3. **Marker gate**. After successful `bind_unix`, the harness drops
|
||||||
|
`<dir>/.bound` next to the socket. c0re's `agent_sockets::write`
|
||||||
|
filters its JSON map by marker presence — only agents whose
|
||||||
|
harness has actually bound the socket appear there (#784 atlas
|
||||||
|
gate). Without this filter, the gateway would `proxy_pass` to a
|
||||||
|
non-existent socket for every sub-agent that hasn't opted in yet.
|
||||||
|
4. **Gateway side** (#829). Reads `agent-sockets.json` at
|
||||||
|
request-handling time and routes `/agent/<name>/` to
|
||||||
|
`http://unix:/run/hive-agent/<name>/web.sock:/`. Whole
|
||||||
|
`/run/hive-agent/` is bind-mounted read-only into the gateway
|
||||||
|
container so it can reach every published socket.
|
||||||
|
|
||||||
|
c0re re-fires `agent_sockets::write` every 10s so newly-bound
|
||||||
|
markers get picked up without needing a container-start hook in
|
||||||
|
every lifecycle path. `write()` is idempotent: steady-state cost is
|
||||||
|
one stat per agent per tick.
|
||||||
|
|
||||||
|
Transition: agents that haven't flipped `useUnixSocket = true` still
|
||||||
|
appear in `agent-ports.json` (the legacy TCP map) and the gateway
|
||||||
|
falls back to TCP for them. Step 4 of #784 will drop the TCP map +
|
||||||
|
the harness's TCP bind once every agent's flipped.
|
||||||
|
|
||||||
## Sequencing history
|
## Sequencing history
|
||||||
|
|
||||||
- #15 v0 (per-agent routing, #740) — first sub-app behind the gateway, JSON port table from c0re.
|
- #15 v0 (per-agent routing, #740) — first sub-app behind the gateway, JSON port table from c0re.
|
||||||
|
|
@ -74,6 +111,7 @@ SSH for forge stays direct on `cfg.sshPort` — separate listener protocol, not
|
||||||
- #749 / #754 — forge to sub-domain (mara: sub-domain over sub-path).
|
- #749 / #754 — forge to sub-domain (mara: sub-domain over sub-path).
|
||||||
- #747 / #764 — matrix sub-domain vhost + `.well-known` delegation.
|
- #747 / #764 — matrix sub-domain vhost + `.well-known` delegation.
|
||||||
- #772 / #775 — fluffychat hops from `<hive>/matrix/` to `matrix.<hive>/`.
|
- #772 / #775 — fluffychat hops from `<hive>/matrix/` to `matrix.<hive>/`.
|
||||||
|
- #784 / #800 / #813 / #815 / #822 / #829 — sub-agent UI flips to unix-domain socket upstream, opt-in per agent.
|
||||||
|
|
||||||
Next-up tracked separately: #14 (container netns isolation), TLS (#594).
|
Next-up tracked separately: #14 (container netns isolation), TLS (#594).
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -195,6 +195,24 @@ fn bind_unix(path: &Path) -> Result<tokio::net::UnixListener> {
|
||||||
use std::os::unix::fs::PermissionsExt;
|
use std::os::unix::fs::PermissionsExt;
|
||||||
std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o660))
|
std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o660))
|
||||||
.with_context(|| format!("set perms on {}", path.display()))?;
|
.with_context(|| format!("set perms on {}", path.display()))?;
|
||||||
|
// Drop a `.bound` marker next to the socket so c0re's
|
||||||
|
// `agent_sockets::write` can filter the JSON map to only include
|
||||||
|
// agents whose harness has actually opted in to (and bound) the
|
||||||
|
// unix socket. Without this gating, gateway would `proxy_pass`
|
||||||
|
// to a non-existent socket for any sub-agent that hasn't flipped
|
||||||
|
// `hyperhive.web.useUnixSocket = true` yet — atlas's concern on
|
||||||
|
// PR #813. Best-effort: a failed write isn't fatal (the harness
|
||||||
|
// still binds + serves on the socket), it just means the
|
||||||
|
// gateway side keeps using the TCP upstream for one more sync.
|
||||||
|
if let Some(parent) = path.parent() {
|
||||||
|
let marker = parent.join(".bound");
|
||||||
|
if let Err(e) = std::fs::write(&marker, b"") {
|
||||||
|
tracing::warn!(
|
||||||
|
marker = %marker.display(), error = %e,
|
||||||
|
"failed to write .bound marker — gateway may keep TCP upstream"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
||||||
Ok(listener)
|
Ok(listener)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -75,6 +75,16 @@ pub const AGENT_SOCKET_DIR: &str = "/run/hive-agent";
|
||||||
/// degree of freedom for callers to get wrong.
|
/// degree of freedom for callers to get wrong.
|
||||||
pub const SOCKET_FILENAME: &str = "web.sock";
|
pub const SOCKET_FILENAME: &str = "web.sock";
|
||||||
|
|
||||||
|
/// Marker file the harness drops next to the socket after a
|
||||||
|
/// successful `bind_unix`. Presence = "this agent has opted in to
|
||||||
|
/// `hyperhive.web.useUnixSocket = true` and its harness has bound
|
||||||
|
/// the socket"; absence = "the harness is still on TCP, don't
|
||||||
|
/// publish the unix upstream for this agent yet". Atlas's gate on
|
||||||
|
/// PR #813 — without it, the gateway would `proxy_pass` to a
|
||||||
|
/// non-existent socket for every sub-agent that hasn't flipped the
|
||||||
|
/// option yet.
|
||||||
|
pub const READY_MARKER: &str = ".bound";
|
||||||
|
|
||||||
#[must_use]
|
#[must_use]
|
||||||
pub fn host_sockets_path() -> PathBuf {
|
pub fn host_sockets_path() -> PathBuf {
|
||||||
PathBuf::from(HOST_SOCKETS_PATH)
|
PathBuf::from(HOST_SOCKETS_PATH)
|
||||||
|
|
@ -103,18 +113,46 @@ pub fn socket_path_for(name: &str) -> PathBuf {
|
||||||
/// (manager UI is routed via the c0re dashboard upstream, not via
|
/// (manager UI is routed via the c0re dashboard upstream, not via
|
||||||
/// `/agent/<name>/`).
|
/// `/agent/<name>/`).
|
||||||
///
|
///
|
||||||
|
/// Also filters by `READY_MARKER` presence: only agents whose
|
||||||
|
/// harness has actually bound the unix socket (and dropped the
|
||||||
|
/// marker) appear in the map. Atlas's gate on #813 — without this,
|
||||||
|
/// the gateway would `proxy_pass` to a non-existent socket for
|
||||||
|
/// every sub-agent that hasn't yet flipped
|
||||||
|
/// `hyperhive.web.useUnixSocket = true`.
|
||||||
|
///
|
||||||
/// `BTreeMap` keeps the JSON output sorted by key so a re-emit
|
/// `BTreeMap` keeps the JSON output sorted by key so a re-emit
|
||||||
/// without churn produces byte-identical output — same idempotency
|
/// without churn produces byte-identical output — same idempotency
|
||||||
/// shape `agent_ports::write` relies on.
|
/// shape `agent_ports::write` relies on.
|
||||||
#[must_use]
|
#[must_use]
|
||||||
pub fn build_map(names: &[String]) -> BTreeMap<String, PathBuf> {
|
pub fn build_map(names: &[String]) -> BTreeMap<String, PathBuf> {
|
||||||
|
build_map_with(names, |name| ready_marker_for(name).exists())
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Body of `build_map` with the ready-check parameterised. Tests
|
||||||
|
/// pass a predicate they control (no real filesystem access).
|
||||||
|
/// Production callers go through `build_map` which wires the
|
||||||
|
/// predicate to the on-disk `.bound` marker check.
|
||||||
|
fn build_map_with<F>(names: &[String], is_ready: F) -> BTreeMap<String, PathBuf>
|
||||||
|
where
|
||||||
|
F: Fn(&str) -> bool,
|
||||||
|
{
|
||||||
names
|
names
|
||||||
.iter()
|
.iter()
|
||||||
.filter(|n| n.as_str() != MANAGER_NAME)
|
.filter(|n| n.as_str() != MANAGER_NAME)
|
||||||
|
.filter(|n| is_ready(n))
|
||||||
.map(|n| (n.clone(), socket_path_for(n)))
|
.map(|n| (n.clone(), socket_path_for(n)))
|
||||||
.collect()
|
.collect()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Path to the per-agent `.bound` marker file the harness writes
|
||||||
|
/// after a successful `bind_unix`. Lives next to `web.sock` in the
|
||||||
|
/// per-agent subdir so it's covered by the same bind-mount and same
|
||||||
|
/// per-agent isolation as the socket itself.
|
||||||
|
#[must_use]
|
||||||
|
pub fn ready_marker_for(name: &str) -> PathBuf {
|
||||||
|
agent_dir_for(name).join(READY_MARKER)
|
||||||
|
}
|
||||||
|
|
||||||
/// Render the map as pretty-printed JSON. Pretty so a human peek at
|
/// Render the map as pretty-printed JSON. Pretty so a human peek at
|
||||||
/// `cat /var/lib/hyperhive/agent-sockets.json` shows one row per agent
|
/// `cat /var/lib/hyperhive/agent-sockets.json` shows one row per agent
|
||||||
/// — keeps the file readable without a separate jq step (mirrors
|
/// — keeps the file readable without a separate jq step (mirrors
|
||||||
|
|
@ -166,6 +204,40 @@ pub fn write(names: &[String]) -> Result<()> {
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Spawn the marker poll task. Periodically re-runs `write` so the
|
||||||
|
/// JSON map picks up newly-bound sockets (an agent flipping
|
||||||
|
/// `hyperhive.web.useUnixSocket = true`, rebuilding, then having its
|
||||||
|
/// harness drop a fresh `.bound` marker) without needing an explicit
|
||||||
|
/// hook on container start. `write` is idempotent (skips the rename
|
||||||
|
/// when content unchanged) so the steady-state cost is one directory
|
||||||
|
/// stat per agent per poll interval.
|
||||||
|
///
|
||||||
|
/// Mirrors the spawn-loop shape used by `crash_watch`,
|
||||||
|
/// `reminder_scheduler`, etc. — the existing background-task
|
||||||
|
/// convention in `main.rs`.
|
||||||
|
pub fn spawn_poll() {
|
||||||
|
tokio::spawn(async move {
|
||||||
|
let mut interval = tokio::time::interval(std::time::Duration::from_secs(10));
|
||||||
|
// First tick fires immediately; that's fine — meta::sync_agents
|
||||||
|
// also writes on boot, this just catches up the window before
|
||||||
|
// the next agent restart.
|
||||||
|
loop {
|
||||||
|
interval.tick().await;
|
||||||
|
match crate::lifecycle::agents_for_meta_listing().await {
|
||||||
|
Ok(agents) => {
|
||||||
|
let names: Vec<String> = agents.into_iter().map(|a| a.name).collect();
|
||||||
|
if let Err(e) = write(&names) {
|
||||||
|
tracing::debug!(error = ?e, "agent_sockets poll write failed");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
Err(e) => {
|
||||||
|
tracing::debug!(error = ?e, "agent_sockets poll: failed to list agents");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
use super::*;
|
use super::*;
|
||||||
|
|
@ -199,12 +271,13 @@ mod tests {
|
||||||
// exercises the filter path — a literal `"hm1nd"` would pass
|
// exercises the filter path — a literal `"hm1nd"` would pass
|
||||||
// trivially if the constant ever changed and the filter
|
// trivially if the constant ever changed and the filter
|
||||||
// silently became a no-op (same pattern as #748 fix on
|
// silently became a no-op (same pattern as #748 fix on
|
||||||
// agent_ports::build_map).
|
// agent_ports::build_map). All-ready predicate bypasses the
|
||||||
|
// marker check so we exercise the manager filter in isolation.
|
||||||
let names: Vec<String> = ["iris", MANAGER_NAME, "argus"]
|
let names: Vec<String> = ["iris", MANAGER_NAME, "argus"]
|
||||||
.iter()
|
.iter()
|
||||||
.map(|s| (*s).to_owned())
|
.map(|s| (*s).to_owned())
|
||||||
.collect();
|
.collect();
|
||||||
let map = build_map(&names);
|
let map = build_map_with(&names, |_| true);
|
||||||
assert!(!map.contains_key(MANAGER_NAME));
|
assert!(!map.contains_key(MANAGER_NAME));
|
||||||
assert!(map.contains_key("iris"));
|
assert!(map.contains_key("iris"));
|
||||||
assert!(map.contains_key("argus"));
|
assert!(map.contains_key("argus"));
|
||||||
|
|
@ -216,13 +289,13 @@ mod tests {
|
||||||
// (build_map for the bulk write, socket_path_for for one-off
|
// (build_map for the bulk write, socket_path_for for one-off
|
||||||
// lookups) without divergence.
|
// lookups) without divergence.
|
||||||
let names = vec!["iris".to_owned()];
|
let names = vec!["iris".to_owned()];
|
||||||
let map = build_map(&names);
|
let map = build_map_with(&names, |_| true);
|
||||||
assert_eq!(map.get("iris"), Some(&socket_path_for("iris")));
|
assert_eq!(map.get("iris"), Some(&socket_path_for("iris")));
|
||||||
}
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn build_map_handles_empty_input() {
|
fn build_map_handles_empty_input() {
|
||||||
let map = build_map(&[]);
|
let map = build_map_with::<fn(&str) -> bool>(&[], |_| true);
|
||||||
assert!(map.is_empty());
|
assert!(map.is_empty());
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -233,10 +306,41 @@ mod tests {
|
||||||
// bug doesn't surface as a corrupt JSON doc (two `"iris":`
|
// bug doesn't surface as a corrupt JSON doc (two `"iris":`
|
||||||
// keys). Mirrors agent_ports test.
|
// keys). Mirrors agent_ports test.
|
||||||
let names = vec!["iris".to_owned(), "iris".to_owned()];
|
let names = vec!["iris".to_owned(), "iris".to_owned()];
|
||||||
let map = build_map(&names);
|
let map = build_map_with(&names, |_| true);
|
||||||
assert_eq!(map.len(), 1);
|
assert_eq!(map.len(), 1);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn build_map_filters_by_ready_predicate() {
|
||||||
|
// The new gate: only ready agents (with `.bound` marker) get
|
||||||
|
// published. Pin the behaviour so a future refactor that
|
||||||
|
// drops the filter surfaces here, not as a 502-spew in the
|
||||||
|
// gateway.
|
||||||
|
let names: Vec<String> = ["iris", "argus", "atlas"]
|
||||||
|
.iter()
|
||||||
|
.map(|s| (*s).to_owned())
|
||||||
|
.collect();
|
||||||
|
// Pretend only `atlas` has flipped + bound (mara on PR #813:
|
||||||
|
// "agents can only access their own sockets" — the gate
|
||||||
|
// makes sure only opted-in agents get a UDS upstream).
|
||||||
|
let map = build_map_with(&names, |name| name == "atlas");
|
||||||
|
assert!(map.contains_key("atlas"));
|
||||||
|
assert!(!map.contains_key("iris"));
|
||||||
|
assert!(!map.contains_key("argus"));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn ready_marker_path_is_sibling_of_socket() {
|
||||||
|
// Marker lives in the same per-agent subdir as the socket so
|
||||||
|
// the same bind-mount covers both; harness writes both inside
|
||||||
|
// the container, host (and gateway via shared bind-mount)
|
||||||
|
// sees both at the deterministic path.
|
||||||
|
let marker = ready_marker_for("iris");
|
||||||
|
let socket = socket_path_for("iris");
|
||||||
|
assert_eq!(marker.parent(), socket.parent());
|
||||||
|
assert_eq!(marker, Path::new("/run/hive-agent/iris/.bound"));
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn render_is_pretty_and_sorted() {
|
fn render_is_pretty_and_sorted() {
|
||||||
let mut map = BTreeMap::new();
|
let mut map = BTreeMap::new();
|
||||||
|
|
|
||||||
|
|
@ -12,9 +12,9 @@ use hive_sh4re::{HostRequest, HostResponse};
|
||||||
// explicit (any new daemon entry point reads off the next add).
|
// explicit (any new daemon entry point reads off the next add).
|
||||||
use hive_c0re::coordinator::Coordinator;
|
use hive_c0re::coordinator::Coordinator;
|
||||||
use hive_c0re::{
|
use hive_c0re::{
|
||||||
auto_update, broker, client, crash_watch, dashboard, dashboard_events, events_vacuum, forge,
|
agent_sockets, auto_update, broker, client, crash_watch, dashboard, dashboard_events,
|
||||||
manager_server, matrix, migrate, rebuild_queue, reminder_scheduler, scheduled_prompts_worker,
|
events_vacuum, forge, manager_server, matrix, migrate, rebuild_queue, reminder_scheduler,
|
||||||
server, stats_vacuum,
|
scheduled_prompts_worker, server, stats_vacuum,
|
||||||
};
|
};
|
||||||
|
|
||||||
#[derive(Parser)]
|
#[derive(Parser)]
|
||||||
|
|
@ -253,6 +253,15 @@ async fn cmd_serve(
|
||||||
// when a previously-running container goes away without an
|
// when a previously-running container goes away without an
|
||||||
// operator-initiated transient state.
|
// operator-initiated transient state.
|
||||||
crash_watch::spawn(coord.clone());
|
crash_watch::spawn(coord.clone());
|
||||||
|
// Agent-sockets marker poll: re-fires `agent_sockets::write`
|
||||||
|
// every 10s so the JSON picks up newly-bound `.bound` markers
|
||||||
|
// (a sub-agent flipping `hyperhive.web.useUnixSocket = true`,
|
||||||
|
// rebuilding, then having its harness bind the socket) without
|
||||||
|
// needing an explicit hook on each container start. write() is
|
||||||
|
// idempotent so steady-state cost is one stat per agent per
|
||||||
|
// tick. closes atlas's #813 concern that agents in the JSON
|
||||||
|
// would 502 the gateway until they actually opt in.
|
||||||
|
agent_sockets::spawn_poll();
|
||||||
// Reminder scheduler: drains due reminders + handles
|
// Reminder scheduler: drains due reminders + handles
|
||||||
// file_path payload persistence. See reminder_scheduler.rs.
|
// file_path payload persistence. See reminder_scheduler.rs.
|
||||||
reminder_scheduler::spawn(coord.clone());
|
reminder_scheduler::spawn(coord.clone());
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue