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

Compare commits

base: vinzenz:cb2c90f32ef7ccca74b674108e5e7f4108e2e80c
Branches Tags
vinzenz:main
..
compare: vinzenz:2989c5ccdb9a55fcf21aad3fc636abb1ba93c553
Branches Tags
vinzenz:main
11 changed files with 9 additions and 444 deletions
Download patch file Download diff file Expand all files Collapse all files

1
Cargo.lock generated
View file

@ -1646,7 +1646,6 @@ name = "hive-agent"
version = "0.1.0" version = "0.1.0"
dependencies = [ dependencies = [
"anyhow", "anyhow",
"async-nats",
"axum", "axum",
"chrono", "chrono",
"clap", "clap",

14
docs/swarm/README.md
View file

@ -389,20 +389,6 @@ how it gets there. A hive that hasn't been given the queue's address for its
agents sets none of the four and each agent logs that it has none; a half-set agents sets none of the four and each agent logs that it has none; a half-set
environment logs an error and the harness keeps serving. environment logs an error and the harness keeps serving.
What an agent does with that connection is publish its terminal. Every row its
own web UI renders also goes to `$SWARM.term.<hive>.<agent>`, one subject per
agent, so a swarm-level terminal can follow one agent without subscribing to
the swarm's whole traffic. The `<hive>` is the one the agent's client id names,
which is the same string the broker builds its grant from. Publishing only: an
agent talks about itself here and reads nothing. Rows aren't retained — a
subscriber that wasn't listening missed them, the same as on the agent's own
live stream.
A row too large for the queue's `max_payload` would be refused outright and
take the connection down with it, so the harness drops such a row's body before
sending and leaves a marker in its place; the summary, level and icon still
arrive. A row that's too large even without its body is logged and skipped.
### Swarm-wide forge webhooks ### Swarm-wide forge webhooks
At startup the controller registers two Forgejo hooks pointing at At startup the controller registers two Forgejo hooks pointing at

8
hive-agent/Cargo.toml
View file

@ -9,9 +9,6 @@ workspace = true
[dependencies] [dependencies]
anyhow.workspace = true anyhow.workspace = true
# Named directly only for the client type the terminal publisher holds; the
# connect and the credential handling live in `swarm-queue-client` below.
async-nats.workspace = true
axum.workspace = true axum.workspace = true
chrono.workspace = true chrono.workspace = true
reqwest.workspace = true reqwest.workspace = true
@ -34,9 +31,8 @@ rusqlite.workspace = true
schemars.workspace = true schemars.workspace = true
serde.workspace = true serde.workspace = true
serde_json.workspace = true serde_json.workspace = true
# Bare: `kv`/`notices` name buckets and streams this harness opens neither end # Bare: `kv`/`notices` name buckets and streams this harness opens neither
# of. The terminal publisher is a plain core-subject publish, so it needs the # end of. All it wants from the crate is `QueueConfig` and, next, a connect.
# connect and the payload limit and nothing from JetStream.
swarm-queue-client.workspace = true swarm-queue-client.workspace = true
tokio.workspace = true tokio.workspace = true
tokio-stream.workspace = true tokio-stream.workspace = true

4
hive-agent/src/main.rs
View file

@ -29,7 +29,6 @@ mod state_entry_watch;
mod stats; mod stats;
mod stream_enrich; mod stream_enrich;
mod swarm_queue; mod swarm_queue;
mod swarm_term;
mod term_msg; mod term_msg;
mod todo_server; mod todo_server;
mod todos; mod todos;
@ -506,9 +505,6 @@ async fn serve_main<S: Surface>(socket: &Path, poll_ms: u64) -> Result<()> {
harness_state::write_api_key_mode(login::using_api_key()); harness_state::write_api_key_mode(login::using_api_key());
let login_state = Arc::new(Mutex::new(initial)); let login_state = Arc::new(Mutex::new(initial));
let bus = Bus::new(); let bus = Bus::new();
// Subscribes before anything is emitted, so the swarm sees this agent's
// terminal from its first row rather than from whenever a reader attached.
swarm_term::spawn(bus.subscribe());
// Set by the web UI's `/api/cancel` on a successful SIGINT, read-and- // Set by the web UI's `/api/cancel` on a successful SIGINT, read-and-
// cleared by `handle_turn` before building the next wake prompt — see // cleared by `handle_turn` before building the next wake prompt — see
// `hive_sh4re::inbox::INTERRUPTED_HINT`. Shared between the web server task and // `hive_sh4re::inbox::INTERRUPTED_HINT`. Shared between the web server task and

53
hive-agent/src/swarm_queue.rs
View file

@ -17,8 +17,6 @@
use std::path::Path; use std::path::Path;
use std::sync::OnceLock; use std::sync::OnceLock;
use tokio::sync::OnceCell;
use swarm_queue_client::QueueConfig; use swarm_queue_client::QueueConfig;
/// Variable prefix for this agent's coordinates. Distinct from `HIVE_C0RE`'s /// Variable prefix for this agent's coordinates. Distinct from `HIVE_C0RE`'s
@ -29,11 +27,6 @@ const ENV_PREFIX: &str = "HIVE_AGENT";
/// one answer rather than re-deriving it per call. /// one answer rather than re-deriving it per call.
static CONFIG: OnceLock<Option<QueueConfig>> = OnceLock::new(); static CONFIG: OnceLock<Option<QueueConfig>> = OnceLock::new();
/// The one connection every publisher in this process shares. Separate from
/// [`CONFIG`] because resolving the coordinates is synchronous boot work and
/// connecting is not — see [`client`].
static CLIENT: OnceCell<Option<async_nats::Client>> = OnceCell::const_new();
/// The four variables the harness unit sets, before the client-id file is /// The four variables the harness unit sets, before the client-id file is
/// read. Collected into a struct so [`decide`] is pure over them and the /// read. Collected into a struct so [`decide`] is pure over them and the
/// process env is touched in exactly one place. /// process env is touched in exactly one place.
@ -165,52 +158,6 @@ pub fn init() {
let _ = CONFIG.set(resolved); let _ = CONFIG.set(resolved);
} }
/// What [`init`] resolved, or `None` when this agent has no queue.
///
/// Borrowed from the `OnceLock` rather than cloned: a caller wants the client
/// id to derive its subject from, and handing out an owned copy of a struct
/// holding a credential path invites it being stored somewhere with a
/// different lifetime than the one place that owns it.
///
/// `None` before [`init`] has run, which is the same answer as "no queue" and
/// deliberately not a panic — the ordering is a boot detail, and a harness
/// that reordered its boot should lose the queue, not die.
pub fn config() -> Option<&'static QueueConfig> {
CONFIG.get()?.as_ref()
}
/// The shared queue connection, made on first call and memoized for the rest
/// of the process.
///
/// One connection per process, not per publisher: the agent authenticates as
/// one client, so a second `connect` would be a second token mint and a second
/// live connection for the same identity rather than a second credential.
///
/// `None` covers both "no queue coordinates" and "configured but the connect
/// failed" — a caller does nothing differently between them, since either way
/// there is nothing to publish onto. Connecting is lazy so that an agent on a
/// hive with no queue pays nothing at boot.
pub async fn client() -> Option<async_nats::Client> {
CLIENT.get_or_init(connect_once).await.clone()
}
async fn connect_once() -> Option<async_nats::Client> {
let cfg = config()?;
match swarm_queue_client::connect(cfg.clone()).await {
Ok(client) => Some(client),
Err(e) => {
// `chain`, not `{:#}`: this is `swarm_queue_client::Error`, whose
// `Display` ignores the alternate flag, so `{:#}` renders the
// headline and drops the cause that says which half failed.
tracing::warn!(
error = %swarm_queue_client::chain(&e),
"swarm queue connect failed; this agent publishes nothing upward"
);
None
}
}
}
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use super::{QueueEnv, Resolution, decide, read_client_id}; use super::{QueueEnv, Resolution, decide, read_client_id};

311
hive-agent/src/swarm_term.rs
View file

@ -1,311 +0,0 @@
//! Publishing this agent's terminal rows onto the swarm queue.
//!
//! The per-agent web UI already classifies the event bus into [`TermMsg`]
//! rows; this offers the same rows upward so a swarm-level terminal can
//! render an agent without reaching into its hive. It publishes the value the
//! web UI is handed rather than a second model, so a new tool or a
//! reclassified event changes both surfaces at once.
//!
//! **Subscribes to the bus, not to a browser.** The SSE handler builds its
//! rows per connected client, so hanging this off it would mean an agent
//! whose terminal nobody has open publishes nothing. This is its own
//! subscriber with its own long-lived [`ClassifyCtx`] — a publisher that
//! restarted its correlation state per reader would lose the `tool_use` →
//! name mapping a `tool_result` needs.
//!
//! **A core subject, not `JetStream`.** A terminal row has no current value a
//! late reader can ask for, and nothing here is worth the durability of the
//! notices stream: a subscriber that was not listening missed the row, the
//! same way it would have missed it on the agent's own SSE stream. That also
//! keeps the agent's grant to a plain publish.
//!
//! **Best-effort in every direction.** The turn loop and the web UI must not
//! notice whether the queue exists, so the bus receiver is the only thing this
//! task blocks on; a failed publish is a log line and the next row is still
//! attempted.
use tokio::sync::broadcast;
use crate::events::BusEvent;
use crate::term_msg::{ClassifyCtx, TermMsg, classify};
/// Subject family carrying agent terminal rows, the swarm-wide agreement this
/// publisher holds up its end of. One leaf subject per agent, so a subscriber
/// can follow one agent without filtering the swarm's whole terminal traffic.
const SUBJECT_PREFIX: &str = "$SWARM.term";
/// What the queue's client id looks like either side of the hive's own name.
///
/// These mirror the auth-callout responder's `--hive-client-prefix` and
/// `--agent-client-suffix`, which is where the authoritative pair lives: the
/// responder parses the hive back out of the presented client id and grants
/// exactly `$SWARM.term.<that hive>.>`. The agent is not told those flags, so
/// it restates their defaults — a deployment that retunes either one has to
/// change them here too, and the symptom of not doing so is every publish
/// refused rather than a wrong subject being accepted.
const CLIENT_ID_PREFIX: &str = "hive-";
const CLIENT_ID_SUFFIX: &str = "-agent";
/// What a row that lost its body to the payload limit carries instead.
const DROPPED_BODY: &str = "[body dropped: over the queue's payload limit]";
/// How much room to leave under the announced limit for everything the
/// publish adds around the payload — subject, headers, protocol framing.
///
/// The server measures the payload alone, so this is not required for
/// correctness; it is here because the cost of being one byte over is not one
/// lost byte but a refused message and a dropped connection, and a kilobyte
/// out of eight megabytes buys that margin for nothing.
const HEADROOM: usize = 1024;
/// The hive a queue client id names.
///
/// The hive in the subject has to be the string the **responder** parses out
/// of this same client id, because that is what its grant is built from. The
/// harness knows a hive display name too, from a different source and with no
/// rule tying the two together — deriving the subject from that one instead
/// produces a publish the broker refuses, which surfaces as a terminal that is
/// simply empty at the swarm.
///
/// `None` for an id that does not have the expected shape: a publisher that
/// guessed at a subject would be asking for a grant it cannot have, so the
/// caller disables itself instead.
fn hive_from_client_id(client_id: &str) -> Option<&str> {
let hive = client_id
.strip_prefix(CLIENT_ID_PREFIX)?
.strip_suffix(CLIENT_ID_SUFFIX)?;
(!hive.is_empty()).then_some(hive)
}
/// Bring `msg` under `limit` serialized bytes, or report that it cannot be.
///
/// The body is the only field that carries arbitrary length — a diff, a whole
/// tool result — so it is the only one worth spending: replacing it keeps the
/// row's identity, level and summary, which is what makes the row readable at
/// all, and a reader sees that something was there rather than seeing nothing.
///
/// `None` means even the degraded row does not fit, so the caller reports it
/// instead of publishing. That matters more than it looks: an oversize publish
/// is not truncated by the server, it is refused and the connection is closed,
/// which costs the row *and* every row racing behind it through the reconnect.
fn fit(msg: TermMsg, limit: usize) -> Option<TermMsg> {
if serialized_len(&msg)? <= limit {
return Some(msg);
}
// Rebuilt rather than mutated in place: `body_format` describes the body,
// and a marker left tagged `Diff` renders as a broken diff downstream.
let degraded = TermMsg {
body: Some(DROPPED_BODY.to_owned()),
body_format: None,
..msg
};
(serialized_len(&degraded)? <= limit).then_some(degraded)
}
/// Serialized size of a row, or `None` if it does not serialize at all.
///
/// Measured by serializing rather than estimated from field lengths: the
/// payload is what the server measures, and JSON escaping makes the two differ
/// by an unbounded factor on exactly the rows that are already near the limit.
fn serialized_len(msg: &TermMsg) -> Option<usize> {
serde_json::to_vec(msg).ok().map(|v| v.len())
}
/// Start the publish task, if this agent has both queue coordinates and an
/// identity the subject can be derived from.
///
/// Returns without spawning in every other case — no queue, an unparseable
/// client id, no label — each of which is a legal state for an agent rather
/// than an error, and each logged once here rather than per row.
pub fn spawn(rx: broadcast::Receiver<BusEvent>) {
let Some(cfg) = crate::swarm_queue::config() else {
// `swarm_queue::init` already said why at boot; repeating it here
// would be the same fact logged twice.
return;
};
let Some(hive) = hive_from_client_id(&cfg.client_id) else {
tracing::warn!(
client_id = %cfg.client_id,
expected = format!("{CLIENT_ID_PREFIX}<hive>{CLIENT_ID_SUFFIX}"),
"queue client id does not name a hive; not publishing the terminal upward"
);
return;
};
let agent = crate::identity::label();
if agent.is_empty() {
tracing::warn!("this agent has no label; not publishing the terminal upward");
return;
}
let subject = format!("{SUBJECT_PREFIX}.{hive}.{agent}");
tokio::spawn(run(rx, subject));
}
async fn run(mut rx: broadcast::Receiver<BusEvent>, subject: String) {
let Some(client) = crate::swarm_queue::client().await else {
return;
};
tracing::info!(subject, "publishing the agent terminal to the swarm queue");
// One context for the task's whole life — see the module doc.
let mut ctx = ClassifyCtx::default();
loop {
let event = match rx.recv().await {
Ok(event) => event,
// The bus drops events for a subscriber that falls behind. The
// terminal is a tail rather than a log, so the answer is to say
// how many were missed and keep reading, exactly as the web UI's
// own subscriber does.
Err(broadcast::error::RecvError::Lagged(missed)) => {
tracing::warn!(missed, "swarm terminal: lagged, rows skipped");
continue;
}
// Every sender is gone, so the harness is shutting down.
Err(broadcast::error::RecvError::Closed) => return,
};
for msg in classify(&event.event, &mut ctx) {
publish(&client, &subject, msg).await;
}
}
}
/// Offer one row. Every failure is terminal for that row and for nothing else.
async fn publish(client: &async_nats::Client, subject: &str, msg: TermMsg) {
// An unconnected client does not fail a publish, it buffers it — and the
// limit read below is the library's pre-connect default until the server
// has announced its own, which is smaller than any deployment sets and
// would degrade rows that would have fit.
if let Err(e) = swarm_queue_client::ensure_connected(client) {
tracing::warn!(error = %swarm_queue_client::chain(&e), "swarm terminal: publish skipped");
return;
}
let limit = swarm_queue_client::max_payload(client).saturating_sub(HEADROOM);
let summary = msg.summary.clone();
let Some(msg) = fit(msg, limit) else {
tracing::warn!(
summary,
limit,
"swarm terminal: row does not fit even without its body, dropped"
);
return;
};
let payload = match serde_json::to_vec(&msg) {
Ok(payload) => payload,
Err(e) => {
tracing::warn!(error = %e, "swarm terminal: serialising failed, row dropped");
return;
}
};
if let Err(e) = client.publish(subject.to_owned(), payload.into()).await {
tracing::warn!(error = %e, "swarm terminal: publish failed, row dropped");
}
}
#[cfg(test)]
mod tests {
use super::{DROPPED_BODY, fit, hive_from_client_id};
use crate::term_msg::{BodyFormat, Level, TermMsg};
#[test]
fn a_client_id_names_the_hive_between_the_prefix_and_the_suffix() {
assert_eq!(hive_from_client_id("hive-alpha-agent"), Some("alpha"));
// A hive whose own name contains the suffix still resolves: only the
// trailing one is stripped, so the responder and this agree.
assert_eq!(
hive_from_client_id("hive-alpha-agent-agent"),
Some("alpha-agent")
);
}
/// Each way an id can fail to name a hive. A guessed subject would be
/// refused by the broker, and a refusal reaches an operator as an empty
/// terminal rather than as an error, so none of these may fall back.
#[test]
fn an_id_of_another_shape_names_no_hive() {
for id in [
// The hive's own id, not an agent's.
"hive-alpha",
// Missing the prefix the responder keys on.
"alpha-agent",
// Prefix and suffix but nothing between them: the subject would
// be `$SWARM.term..<agent>`, whose empty token matches no grant.
"hive--agent",
// Prefix and suffix overlapping with no hive at all.
"hive-agent",
"",
] {
assert_eq!(hive_from_client_id(id), None, "{id} must name no hive");
}
}
#[test]
fn a_row_that_already_fits_is_published_unchanged() {
let msg = TermMsg::new(Level::Info, "turn ok")
.icon("✅")
.body("a short body", Some(BodyFormat::Markdown));
let fitted = fit(msg, 4096).expect("a small row fits");
assert_eq!(fitted.body.as_deref(), Some("a short body"));
assert_eq!(fitted.body_format, Some(BodyFormat::Markdown));
}
/// The case the degrade exists for: a body larger than the limit costs the
/// body and nothing else, and what comes back is actually under the limit
/// rather than merely smaller.
#[test]
fn an_oversize_body_is_replaced_and_the_result_fits() {
let limit = 512;
let msg = TermMsg::new(Level::Info, "Edit(src/main.rs)")
.icon("🔧")
.body("x".repeat(limit * 4), Some(BodyFormat::Diff))
.coalesce("tool-1");
let fitted = fit(msg, limit).expect("dropping the body brings this under the limit");
assert_eq!(fitted.body.as_deref(), Some(DROPPED_BODY));
// The tag describes a body that is no longer there; left set, a reader
// renders the marker as a diff.
assert_eq!(fitted.body_format, None);
// The fields that make the row readable survive.
assert_eq!(fitted.summary, "Edit(src/main.rs)");
assert_eq!(fitted.icon.as_deref(), Some("🔧"));
assert_eq!(fitted.coalesce_key.as_deref(), Some("tool-1"));
assert!(
serde_json::to_vec(&fitted).expect("serialises").len() <= limit,
"the degraded row must be under the limit, not merely smaller"
);
}
/// A row whose summary alone exceeds the limit cannot be degraded into
/// one, and publishing it anyway would cost the connection rather than the
/// row. Reported by the caller, never sent.
#[test]
fn a_row_too_large_even_without_its_body_is_refused() {
let limit = 256;
let msg = TermMsg::new(Level::Warn, "s".repeat(limit * 4)).body("x".repeat(limit), None);
assert!(fit(msg, limit).is_none());
}
/// Control for the test above: the same oversize summary with a body that
/// would fit still refuses, so the refusal is about total size and not
/// about the body having been present.
#[test]
fn the_refusal_is_about_size_rather_than_the_body_being_present() {
let limit = 256;
let msg = TermMsg::new(Level::Warn, "s".repeat(limit * 4));
assert!(fit(msg, limit).is_none());
}
/// JSON escaping is why the size is measured by serializing: a body of
/// quotes serializes to twice its own length, so a row that fits by
/// character count can still be refused on the wire.
#[test]
fn the_limit_is_measured_on_the_serialized_bytes() {
let body = "\"".repeat(200);
let msg = TermMsg::new(Level::Info, "quotes").body(body.clone(), None);
let limit = body.len() + 64;
// Well under the limit as characters, over it once escaped.
assert!(
serde_json::to_vec(&msg).expect("serialises").len() > limit,
"this fixture must be oversize only after escaping"
);
let fitted = fit(msg, limit).expect("dropping the body fits");
assert_eq!(fitted.body.as_deref(), Some(DROPPED_BODY));
}
}

2
hive-c0re/src/swarm_agent_status.rs
View file

@ -35,7 +35,7 @@ const FAILURES_BEFORE_BANNER: u32 = 3;
/// ///
/// Shares its connect gate with [`crate::swarm_status::spawn`] — both read /// Shares its connect gate with [`crate::swarm_status::spawn`] — both read
/// `crate::swarm_queue::client()`, which connects once per process — so a /// `crate::swarm_queue::client()`, which connects once per process — so a
/// hive with no queue coordinates pays for this decision once, not twice. /// hive with no queue configured pays for this decision once, not twice.
/// ///
/// Takes no `Coordinator` handle, unlike its sibling: this task only reads /// Takes no `Coordinator` handle, unlike its sibling: this task only reads
/// agent state that already exists on disk / in the container runtime /// agent state that already exists on disk / in the container runtime

4
hive-c0re/src/swarm_notices.rs
View file

@ -13,8 +13,8 @@
//! [`swarm_queue_client::notices`] for the stream this publishes into. //! [`swarm_queue_client::notices`] for the stream this publishes into.
//! //!
//! **Best-effort, never fatal to the caller.** A hive with no queue //! **Best-effort, never fatal to the caller.** A hive with no queue
//! coordinates is a silent no-op (the ordinary case). A hive that cannot //! configured is a silent no-op (the ordinary case). A hive whose queue
//! reach the queue loses the swarm's visibility of the notice, not the //! is unreachable loses the swarm's visibility of the notice, not the
//! host's — `warn!` fires on every failed attempt regardless, and the //! host's — `warn!` fires on every failed attempt regardless, and the
//! dashboard banners only after [`FAILURES_BEFORE_BANNER`] consecutive //! dashboard banners only after [`FAILURES_BEFORE_BANNER`] consecutive
//! misses, the same debounce shape [`crate::swarm_status`] uses and for //! misses, the same debounce shape [`crate::swarm_status`] uses and for

18
nix/host-modules/swarm-nats.nix
View file

@ -677,10 +677,10 @@ in
# The most a single publish may be, in bytes, before the # The most a single publish may be, in bytes, before the
# server answers `-ERR 'Maximum Payload Violation'` and # server answers `-ERR 'Maximum Payload Violation'` and
# closes the connection — not a truncation, a dropped row # closes the connection — not a truncation, a dropped row
# and a reconnect. This queue carries agent terminal rows # and a reconnect. This queue is about to carry agent
# published whole rather than split, so upstream's own # terminal rows published whole rather than split, so
# default is sized to lose one of those rather than merely # upstream's own default is sized to lose one of those
# shorten it. Bounded from above by # rather than merely shorten it. Bounded from above by
# `max_pending`: nats-server refuses to start once this # `max_pending`: nats-server refuses to start once this
# exceeds it, and widening that ceiling instead costs # exceeds it, and widening that ceiling instead costs
# memory per connection, so this stays comfortably under it. # memory per connection, so this stays comfortably under it.
@ -783,16 +783,6 @@ in
"--hive-client-prefix ${lib.escapeShellArg autheliaCfg.hiveClientPrefix}" "--hive-client-prefix ${lib.escapeShellArg autheliaCfg.hiveClientPrefix}"
"--agent-client-suffix ${lib.escapeShellArg autheliaCfg.agentClientSuffix}" "--agent-client-suffix ${lib.escapeShellArg autheliaCfg.agentClientSuffix}"
"--reader-client ${lib.escapeShellArg controllerCfg.queueClientId}" "--reader-client ${lib.escapeShellArg controllerCfg.queueClientId}"
# What an agent may publish to, `{hive}` standing for the
# hive its client id names. Without it the responder has no
# agent grant to hand out and refuses every agent at CONNECT,
# so the terminal stream below depends on this line existing.
#
# `$$`, not `$`: systemd substitutes `$NAME` in `ExecStart`
# whether or not it is quoted, so a single dollar reaches the
# responder as the empty expansion of an unset `SWARM` and the
# grant silently becomes `.term.{hive}.>`.
"--agent-publish-subject ${lib.escapeShellArg "\$\$SWARM.term.{hive}.>"}"
]; ];
# Every credential arrives by `LoadCredential` and is named # Every credential arrives by `LoadCredential` and is named
# on the command line only as a **path** — `argv` is # on the command line only as a **path** — `argv` is

18
nix/module-eval.nix
View file

@ -773,24 +773,6 @@ let
name = "the queue's payload ceiling is set, not inherited from the server's default"; name = "the queue's payload ceiling is set, not inherited from the server's default";
ok = natsOldPath.containers.swarm-nats.config.services.nats.settings.max_payload == 8388608; ok = natsOldPath.containers.swarm-nats.config.services.nats.settings.max_payload == 8388608;
} }
{
# Reads the RENDERED unit text, not the module's source, because the
# failure this defends against renders perfectly: systemd substitutes
# `$NAME` in `ExecStart` regardless of quoting, so a single dollar
# here hands the responder `.term.{hive}.>` — a grant that parses, is
# accepted, and matches nothing an agent ever publishes to. Asserting
# the doubled dollar is the only way to tell the two apart before
# deploy. The flag's presence is asserted separately so that dropping
# the grant entirely fails as its own arm rather than as an escaping
# complaint.
name = "the responder grants agents their hive's terminal subject, and the dollar survives systemd";
ok =
let
exec =
natsOldPath.containers.swarm-nats.config.systemd.services.swarm-nats-auth.serviceConfig.ExecStart;
in
lib.hasInfix "--agent-publish-subject " exec && lib.hasInfix "$$SWARM.term.{hive}.>" exec;
}
{ {
# Reads the host's tmpfiles rules, not the options: the socket directory # Reads the host's tmpfiles rules, not the options: the socket directory
# nginx and the container share is created there, so a rename that # nginx and the container share is created there, so a rename that

20
swarm-queue-client/src/lib.rs
View file

@ -634,26 +634,6 @@ pub fn ensure_connected(client: &async_nats::Client) -> Result<(), Error> {
Ok(()) Ok(())
} }
/// The largest payload this connection's server will accept, as the server
/// itself announced it at CONNECT.
///
/// Read it from here rather than from a constant: the limit is set in the
/// queue's own NixOS module, so a client that hardcoded it would be asserting
/// a number it does not own and would keep asserting it after the operator
/// raised it. A publisher that exceeds the limit does not get a shortened
/// message — the server answers `Maximum Payload Violation` and closes the
/// connection, costing the row and a reconnect — so a publisher of
/// variable-size rows must size them against this before handing them over.
///
/// ⚠️ The value is per connection. Before the first successful CONNECT the
/// client reports the library's own default, which is smaller than any
/// deployment sets; call this after [`ensure_connected`], and the answer is
/// the one this server actually announced.
#[must_use]
pub fn max_payload(client: &async_nats::Client) -> usize {
client.server_info().max_payload
}
/// Connect to the swarm queue, presenting a token on each connection attempt /// Connect to the swarm queue, presenting a token on each connection attempt
/// and minting a fresh one only when the cached one is near expiry. /// and minting a fresh one only when the cached one is near expiry.
/// ///