diff --git a/Cargo.lock b/Cargo.lock index aa828666..6b2e9f51 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1646,7 +1646,6 @@ name = "hive-agent" version = "0.1.0" dependencies = [ "anyhow", - "async-nats", "axum", "chrono", "clap", diff --git a/docs/swarm/README.md b/docs/swarm/README.md index c09e404e..e67bd379 100644 --- a/docs/swarm/README.md +++ b/docs/swarm/README.md @@ -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 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..`, one subject per -agent, so a swarm-level terminal can follow one agent without subscribing to -the swarm's whole traffic. The `` 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 At startup the controller registers two Forgejo hooks pointing at diff --git a/hive-agent/Cargo.toml b/hive-agent/Cargo.toml index 14478708..205fba07 100644 --- a/hive-agent/Cargo.toml +++ b/hive-agent/Cargo.toml @@ -9,9 +9,6 @@ workspace = true [dependencies] 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 chrono.workspace = true reqwest.workspace = true @@ -34,9 +31,8 @@ rusqlite.workspace = true schemars.workspace = true serde.workspace = true serde_json.workspace = true -# Bare: `kv`/`notices` name buckets and streams this harness opens neither end -# of. The terminal publisher is a plain core-subject publish, so it needs the -# connect and the payload limit and nothing from JetStream. +# Bare: `kv`/`notices` name buckets and streams this harness opens neither +# end of. All it wants from the crate is `QueueConfig` and, next, a connect. swarm-queue-client.workspace = true tokio.workspace = true tokio-stream.workspace = true diff --git a/hive-agent/src/main.rs b/hive-agent/src/main.rs index 55f99144..28d371b6 100644 --- a/hive-agent/src/main.rs +++ b/hive-agent/src/main.rs @@ -29,7 +29,6 @@ mod state_entry_watch; mod stats; mod stream_enrich; mod swarm_queue; -mod swarm_term; mod term_msg; mod todo_server; mod todos; @@ -506,9 +505,6 @@ async fn serve_main(socket: &Path, poll_ms: u64) -> Result<()> { harness_state::write_api_key_mode(login::using_api_key()); let login_state = Arc::new(Mutex::new(initial)); 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- // cleared by `handle_turn` before building the next wake prompt — see // `hive_sh4re::inbox::INTERRUPTED_HINT`. Shared between the web server task and diff --git a/hive-agent/src/swarm_queue.rs b/hive-agent/src/swarm_queue.rs index c5d6250a..ab93fac7 100644 --- a/hive-agent/src/swarm_queue.rs +++ b/hive-agent/src/swarm_queue.rs @@ -17,8 +17,6 @@ use std::path::Path; use std::sync::OnceLock; -use tokio::sync::OnceCell; - use swarm_queue_client::QueueConfig; /// 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. static CONFIG: OnceLock> = 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> = OnceCell::const_new(); - /// 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 /// process env is touched in exactly one place. @@ -165,52 +158,6 @@ pub fn init() { 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 { - CLIENT.get_or_init(connect_once).await.clone() -} - -async fn connect_once() -> Option { - 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)] mod tests { use super::{QueueEnv, Resolution, decide, read_client_id}; diff --git a/hive-agent/src/swarm_term.rs b/hive-agent/src/swarm_term.rs deleted file mode 100644 index 187eabaa..00000000 --- a/hive-agent/src/swarm_term.rs +++ /dev/null @@ -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..>`. 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 { - 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(°raded)? <= 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 { - 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) { - 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}{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, 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..`, 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)); - } -} diff --git a/hive-c0re/src/swarm_agent_status.rs b/hive-c0re/src/swarm_agent_status.rs index 54dd1777..2514927c 100644 --- a/hive-c0re/src/swarm_agent_status.rs +++ b/hive-c0re/src/swarm_agent_status.rs @@ -35,7 +35,7 @@ const FAILURES_BEFORE_BANNER: u32 = 3; /// /// Shares its connect gate with [`crate::swarm_status::spawn`] — both read /// `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 /// agent state that already exists on disk / in the container runtime diff --git a/hive-c0re/src/swarm_notices.rs b/hive-c0re/src/swarm_notices.rs index 914fcd70..d2b9ceb4 100644 --- a/hive-c0re/src/swarm_notices.rs +++ b/hive-c0re/src/swarm_notices.rs @@ -13,8 +13,8 @@ //! [`swarm_queue_client::notices`] for the stream this publishes into. //! //! **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 -//! reach the queue loses the swarm's visibility of the notice, not the +//! configured is a silent no-op (the ordinary case). A hive whose queue +//! is unreachable loses the swarm's visibility of the notice, not the //! host's — `warn!` fires on every failed attempt regardless, and the //! dashboard banners only after [`FAILURES_BEFORE_BANNER`] consecutive //! misses, the same debounce shape [`crate::swarm_status`] uses and for diff --git a/nix/host-modules/swarm-nats.nix b/nix/host-modules/swarm-nats.nix index 4541fef2..341ebb6a 100644 --- a/nix/host-modules/swarm-nats.nix +++ b/nix/host-modules/swarm-nats.nix @@ -677,10 +677,10 @@ in # The most a single publish may be, in bytes, before the # server answers `-ERR 'Maximum Payload Violation'` and # closes the connection — not a truncation, a dropped row - # and a reconnect. This queue carries agent terminal rows - # published whole rather than split, so upstream's own - # default is sized to lose one of those rather than merely - # shorten it. Bounded from above by + # and a reconnect. This queue is about to carry agent + # terminal rows published whole rather than split, so + # upstream's own default is sized to lose one of those + # rather than merely shorten it. Bounded from above by # `max_pending`: nats-server refuses to start once this # exceeds it, and widening that ceiling instead costs # memory per connection, so this stays comfortably under it. @@ -783,16 +783,6 @@ in "--hive-client-prefix ${lib.escapeShellArg autheliaCfg.hiveClientPrefix}" "--agent-client-suffix ${lib.escapeShellArg autheliaCfg.agentClientSuffix}" "--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 # on the command line only as a **path** — `argv` is diff --git a/nix/module-eval.nix b/nix/module-eval.nix index a754f806..4cc04261 100644 --- a/nix/module-eval.nix +++ b/nix/module-eval.nix @@ -773,24 +773,6 @@ let 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; } - { - # 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 # nginx and the container share is created there, so a rename that diff --git a/swarm-queue-client/src/lib.rs b/swarm-queue-client/src/lib.rs index 0d353be3..809304ff 100644 --- a/swarm-queue-client/src/lib.rs +++ b/swarm-queue-client/src/lib.rs @@ -634,26 +634,6 @@ pub fn ensure_connected(client: &async_nats::Client) -> Result<(), Error> { 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 /// and minting a fresh one only when the cached one is near expiry. ///