refactor(swarm-queue-client): typed errors for the bucket and the guard
Finishes the anyhow removal for the parts this branch adds: the status bucket's open-or-create and the connected-client precondition. Two variants, one of them behind the `kv` feature because the error type it wraps does not exist without it — the error enum respects the same gate the module does. NotConnected is deliberately distinct from Connect: one is a connect that was attempted and refused, the other is a request made before any connection exists. The first is a deployment problem and the second is a caller-ordering one, which is the whole reason a caller wants an enum rather than a string. The controller's `store` now returns the queue client's error rather than an anyhow one: `OnceCell::get_or_try_init` takes its error type from the closure, so widening there would mean converting inside the closure for no gain. `view` `?`s it and anyhow converts at that boundary — the library keeps a typed error, the binary keeps anyhow, and no call site pays for the split.
This commit is contained in:
parent
c028b2ecfc
commit
3273971328
3 changed files with 33 additions and 6 deletions
|
|
@ -155,7 +155,16 @@ impl StatusReader {
|
||||||
///
|
///
|
||||||
/// Whichever side arrives first creates it, and both sides ask for the
|
/// Whichever side arrives first creates it, and both sides ask for the
|
||||||
/// same shape, so this is a race with one outcome.
|
/// same shape, so this is a race with one outcome.
|
||||||
async fn store(&self) -> Result<&async_nats::jetstream::kv::Store> {
|
///
|
||||||
|
/// Returns the queue client's own error rather than an `anyhow::Error`:
|
||||||
|
/// `OnceCell::get_or_try_init` takes its error type from the closure,
|
||||||
|
/// so widening here would mean converting *inside* the closure for no
|
||||||
|
/// gain. `view` below `?`s it and anyhow converts there — which is the
|
||||||
|
/// whole point of the library keeping a typed error while the binary
|
||||||
|
/// keeps anyhow.
|
||||||
|
async fn store(
|
||||||
|
&self,
|
||||||
|
) -> std::result::Result<&async_nats::jetstream::kv::Store, swarm_queue_client::Error> {
|
||||||
self.store
|
self.store
|
||||||
.get_or_try_init(|| swarm_queue_client::status::open_or_create(&self.client))
|
.get_or_try_init(|| swarm_queue_client::status::open_or_create(&self.client))
|
||||||
.await
|
.await
|
||||||
|
|
|
||||||
|
|
@ -84,6 +84,21 @@ pub enum Error {
|
||||||
#[source]
|
#[source]
|
||||||
source: async_nats::ConnectError,
|
source: async_nats::ConnectError,
|
||||||
},
|
},
|
||||||
|
|
||||||
|
/// Distinct from [`Error::Connect`] on purpose: that one is a connect
|
||||||
|
/// that was attempted and refused, this one is a request made before
|
||||||
|
/// any connection exists. The second is a caller-ordering problem and
|
||||||
|
/// the first is a deployment one.
|
||||||
|
#[error("not connected to the swarm queue (client state: {0:?})")]
|
||||||
|
NotConnected(async_nats::connection::State),
|
||||||
|
|
||||||
|
#[cfg(feature = "kv")]
|
||||||
|
#[error("creating the {bucket} bucket")]
|
||||||
|
CreateBucket {
|
||||||
|
bucket: &'static str,
|
||||||
|
#[source]
|
||||||
|
source: async_nats::jetstream::context::CreateKeyValueError,
|
||||||
|
},
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Render an error and its source chain on one line.
|
/// Render an error and its source chain on one line.
|
||||||
|
|
@ -242,10 +257,10 @@ async fn mint_token(http: &reqwest::Client, cfg: &QueueConfig) -> Result<String,
|
||||||
///
|
///
|
||||||
/// Naming the state is also the better error: "not connected" is actionable,
|
/// Naming the state is also the better error: "not connected" is actionable,
|
||||||
/// a timeout is not.
|
/// a timeout is not.
|
||||||
pub fn ensure_connected(client: &async_nats::Client) -> Result<()> {
|
pub fn ensure_connected(client: &async_nats::Client) -> Result<(), Error> {
|
||||||
let state = client.connection_state();
|
let state = client.connection_state();
|
||||||
if state != async_nats::connection::State::Connected {
|
if state != async_nats::connection::State::Connected {
|
||||||
bail!("not connected to the swarm queue (client state: {state:?})");
|
return Err(Error::NotConnected(state));
|
||||||
}
|
}
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -19,7 +19,7 @@
|
||||||
//! responder, still pulls neither `jetstream` nor `kv`: it speaks the
|
//! responder, still pulls neither `jetstream` nor `kv`: it speaks the
|
||||||
//! connect and nothing else.
|
//! connect and nothing else.
|
||||||
|
|
||||||
use anyhow::{Context, Result};
|
use crate::Error;
|
||||||
|
|
||||||
/// The KV bucket hives publish their status snapshots into, one key per
|
/// The KV bucket hives publish their status snapshots into, one key per
|
||||||
/// hive keyed by `hiveName`.
|
/// hive keyed by `hiveName`.
|
||||||
|
|
@ -41,7 +41,7 @@ pub const BUCKET: &str = "hive-status";
|
||||||
/// into a permanent, silent absence of data.
|
/// into a permanent, silent absence of data.
|
||||||
pub async fn open_or_create(
|
pub async fn open_or_create(
|
||||||
client: &async_nats::Client,
|
client: &async_nats::Client,
|
||||||
) -> Result<async_nats::jetstream::kv::Store> {
|
) -> Result<async_nats::jetstream::kv::Store, Error> {
|
||||||
let js = async_nats::jetstream::new(client.clone());
|
let js = async_nats::jetstream::new(client.clone());
|
||||||
match js.get_key_value(BUCKET).await {
|
match js.get_key_value(BUCKET).await {
|
||||||
Ok(store) => Ok(store),
|
Ok(store) => Ok(store),
|
||||||
|
|
@ -58,7 +58,10 @@ pub async fn open_or_create(
|
||||||
..Default::default()
|
..Default::default()
|
||||||
})
|
})
|
||||||
.await
|
.await
|
||||||
.with_context(|| format!("creating the {BUCKET} bucket"))
|
.map_err(|source| Error::CreateBucket {
|
||||||
|
bucket: BUCKET,
|
||||||
|
source,
|
||||||
|
})
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue