92 lines
4.1 KiB
Rust
92 lines
4.1 KiB
Rust
//! The hive-notices stream: its name, subject shape, and how a hive
|
|
//! opens it to publish.
|
|
//!
|
|
//! Same reason [`crate::status`] exists rather than a bare `const` on
|
|
//! whichever side happens to need one first: a hive that publishes and a
|
|
//! swarm-level reader that eventually consumes live in different crates,
|
|
//! and a literal name repeated across both is an agreement nothing
|
|
//! checks.
|
|
//!
|
|
//! **This is a stream, not a bucket, and that is a real design choice —
|
|
//! not the same shape as [`crate::status`] wearing a different name.**
|
|
//! [`crate::status`]'s hive-status snapshot has a current value: a
|
|
//! reconnecting reader can always ask "what does this hive say *now*"
|
|
//! and get the true answer, so a KV bucket (last value per key) is the
|
|
//! right shape. A lifecycle notice ("container crashed at 04:12") has no
|
|
//! such steady state — miss the message and there is nothing left to
|
|
//! read later that would tell you it happened. That needs durable
|
|
//! delivery (a JetStream stream a consumer acks against), which is what
|
|
//! this module opens instead.
|
|
//!
|
|
//! Feature-gated (`notices`) for the same reason [`crate::status`] is
|
|
//! gated behind `kv`: the crate's other consumers (the auth-callout
|
|
//! responder, a hive that only publishes status) should not compile
|
|
//! against a stream shape they never touch.
|
|
|
|
use crate::Error;
|
|
|
|
/// The stream a hive publishes lifecycle notices into.
|
|
///
|
|
/// A constant and not an option, matching [`crate::status::BUCKET`]:
|
|
/// reader and writer must name the same stream, and letting either side
|
|
/// pick its own name is how two deployments end up disagreeing about
|
|
/// which stream a notice actually landed in.
|
|
pub const STREAM: &str = "hive-notices";
|
|
|
|
/// Build the subject a given hive's notices publish to.
|
|
///
|
|
/// Every hive's notices land under `STREAM`, one subject per hive:
|
|
/// `hive-notices.<hiveName>`.
|
|
///
|
|
/// Not one subject per notice *kind* — a consumer that wants a specific
|
|
/// hive's notices subscribes to `subject(hive)`; one that wants the whole
|
|
/// swarm's subscribes to `{STREAM}.>`. The kind travels inside the message
|
|
/// payload instead, so adding a new notice kind is never a subject-design
|
|
/// change. Deliberately reuses `STREAM` rather than a second `PREFIX`
|
|
/// constant with the same value — one name for one fact, same reasoning
|
|
/// the module doc above gives for a shared `const` over a repeated literal.
|
|
#[must_use]
|
|
pub fn subject(hive: &str) -> String {
|
|
format!("{STREAM}.{hive}")
|
|
}
|
|
|
|
/// Open the notices stream, creating it if nothing has yet.
|
|
///
|
|
/// **Retention is time-bounded (30 days), not unbounded.** A notice
|
|
/// this old has long since been superseded by whatever the hive is
|
|
/// doing now — keeping it forever buys nothing but disk, the same
|
|
/// argument `hive-forge`'s own bash-task retention makes elsewhere in
|
|
/// this workspace.
|
|
///
|
|
/// Creating rather than requiring a provisioning step is the same call
|
|
/// [`crate::status::open_or_create`] makes and for the same reason: a
|
|
/// hive and a swarm-level consumer come up in no particular order, and
|
|
/// a stream that must pre-exist turns "deployed in the wrong order"
|
|
/// into a permanent, silent absence of data.
|
|
pub async fn open_or_create(
|
|
client: &async_nats::Client,
|
|
) -> Result<async_nats::jetstream::stream::Stream, Error> {
|
|
let js = async_nats::jetstream::new(client.clone());
|
|
match js.get_stream(STREAM).await {
|
|
Ok(stream) => Ok(stream),
|
|
Err(e) => {
|
|
tracing::info!(
|
|
stream = STREAM,
|
|
reason = %e,
|
|
"notices stream not available, creating it"
|
|
);
|
|
js.create_stream(async_nats::jetstream::stream::Config {
|
|
name: STREAM.to_owned(),
|
|
description: Some("Lifecycle notices offered by each hive".to_owned()),
|
|
subjects: vec![format!("{STREAM}.>")],
|
|
max_age: std::time::Duration::from_hours(30 * 24),
|
|
..Default::default()
|
|
})
|
|
.await
|
|
.map_err(|source| Error::CreateStream {
|
|
stream: STREAM,
|
|
source,
|
|
})
|
|
}
|
|
}
|
|
}
|