109 lines
4.3 KiB
Rust
109 lines
4.3 KiB
Rust
//! The connection side of the hive-wanted KV buckets: opening and watching a
|
|
//! hive's bucket. The bucket names and the stored types, and why absence is
|
|
//! not a deletion order, are `swarm_wire::wanted`.
|
|
//!
|
|
//! # Only the controller creates it
|
|
//!
|
|
//! Unlike [`crate::status::open_or_create`], where either end may legitimately
|
|
//! arrive first, the writer here is single and known. A hive opens read-only
|
|
//! and treats absence as "not published yet", so that stays quiet rather than
|
|
//! looking like an error a hive could fix.
|
|
|
|
#[cfg(feature = "kv")]
|
|
use crate::Error;
|
|
#[cfg(feature = "kv")]
|
|
use swarm_wire::wanted::bucket;
|
|
|
|
/// Open the wanted-state bucket for writing, creating it if nothing has yet.
|
|
///
|
|
/// **Controller-side only.** `history: 1` because a hive converges to the
|
|
/// current declaration and never asks what the previous one was. The
|
|
/// controller keeps no second copy either: the value published here is the
|
|
/// record, and it reads it back from this bucket when it needs it.
|
|
#[cfg(feature = "kv")]
|
|
pub async fn open_or_create(
|
|
client: &async_nats::Client,
|
|
hive: &str,
|
|
) -> Result<async_nats::jetstream::kv::Store, Error> {
|
|
let js = async_nats::jetstream::new(client.clone());
|
|
let bucket = bucket(hive);
|
|
let description = format!("Agent set the swarm controller declares for hive {hive}");
|
|
crate::kv::open_or_create(&js, &bucket, "wanted-state", description).await
|
|
}
|
|
|
|
/// Open the wanted-state bucket for reading, or `None` when it does not exist.
|
|
///
|
|
/// **Hive-side.** `None` is the ordinary pre-publication state, not a failure:
|
|
/// a hive holds no grant to create this bucket and must not treat its absence
|
|
/// as a reason to converge to an empty agent set — see `swarm_wire::wanted`.
|
|
///
|
|
/// # Errors
|
|
///
|
|
/// [`Error::OpenBucket`] for every failure other than the bucket's stream not
|
|
/// existing: a refused grant, a timeout, `JetStream` unavailable.
|
|
#[cfg(feature = "kv")]
|
|
pub async fn open_read_only(
|
|
client: &async_nats::Client,
|
|
hive: &str,
|
|
) -> Result<Option<async_nats::jetstream::kv::Store>, Error> {
|
|
let js = async_nats::jetstream::new(client.clone());
|
|
let bucket = bucket(hive);
|
|
match js.get_key_value(&bucket).await {
|
|
Ok(store) => Ok(Some(store)),
|
|
Err(e) if is_bucket_not_found(&e) => Ok(None),
|
|
Err(source) => Err(Error::OpenBucket { bucket, source }),
|
|
}
|
|
}
|
|
|
|
/// `get_key_value` reports a missing bucket as `GetBucket` wrapping the
|
|
/// stream lookup's `STREAM_NOT_FOUND`; every other kind is a real failure.
|
|
#[cfg(feature = "kv")]
|
|
pub(crate) fn is_bucket_not_found(e: &async_nats::jetstream::context::KeyValueError) -> bool {
|
|
use async_nats::jetstream::context::{GetStreamError, GetStreamErrorKind, KeyValueErrorKind};
|
|
|
|
e.kind() == KeyValueErrorKind::GetBucket
|
|
&& std::error::Error::source(e)
|
|
.and_then(|s| s.downcast_ref::<GetStreamError>())
|
|
.is_some_and(|s| {
|
|
matches!(
|
|
s.kind(),
|
|
GetStreamErrorKind::JetStream(js)
|
|
if js.error_code() == async_nats::jetstream::ErrorCode::STREAM_NOT_FOUND
|
|
)
|
|
})
|
|
}
|
|
|
|
/// Watch this hive's own declaration for changes.
|
|
///
|
|
/// **Hive-side.** `None` on the same terms as [`open_read_only`]: no bucket
|
|
/// yet. A watch is a `JetStream` *consumer*, so it needs a grant a plain `get`
|
|
/// does not; a hive missing `CONSUMER.CREATE` on its own stream gets
|
|
/// [`Error::WatchKey`] here while `get` keeps working.
|
|
///
|
|
/// Updates only, deliberately: the boot-time read already has the current
|
|
/// value, and a watch that replayed history would re-converge the whole
|
|
/// declaration on every reconnect for nothing.
|
|
///
|
|
/// # Errors
|
|
///
|
|
/// [`Error::OpenBucket`] as for [`open_read_only`], and [`Error::WatchKey`]
|
|
/// when the watch itself is refused or fails.
|
|
#[cfg(feature = "kv")]
|
|
pub async fn watch(
|
|
client: &async_nats::Client,
|
|
hive: &str,
|
|
) -> Result<Option<async_nats::jetstream::kv::Watch>, Error> {
|
|
let Some(store) = open_read_only(client, hive).await? else {
|
|
return Ok(None);
|
|
};
|
|
// The bucket holds exactly one key, named for the hive — same key
|
|
// `open_read_only`'s caller reads, so both paths address one declaration.
|
|
store
|
|
.watch(hive)
|
|
.await
|
|
.map(Some)
|
|
.map_err(|source| Error::WatchKey {
|
|
key: hive.to_owned(),
|
|
source,
|
|
})
|
|
}
|