zenkey-fleet 0.18.0

//! Producer-side bring-up discipline (RFC 04 §5, RFC 05 §2.1/§3, RFC 08
//! §6.1) — the MUSTs every producer owes the bus, as an API that makes the
//! wrong shapes unrepresentable rather than a checklist that trusts review.
//!
//! The conformance review found these uniformly unenforced, with the
//! in-repo dev tools modeling the violations; this module is the
//! enforcement seam, and [`crate::bus::serve`]/[`crate::tape::generate`] now ride it
//! or its rules. It lives beside — not inside — `serve.rs`, deliberately:
//! `serve` is the *explorer's* mock (answer anything, wildcards welcome),
//! while this is the *producer's* posture (concrete keys, ordered
//! presence), and folding the strict API into the permissive one would
//! give every mock a way to claim it is a producer.
//!
//! Four rules, four shapes:
//!
//! - **Order** ([`BringUp`]): a producer declares its queryables
//!   (`introspect`, `describe`, its procedures) **first**, and its `alive`
//!   liveliness token **last** — "alive ⇒ callable" (RFC 04 §5): callers
//!   attribute RPC silence against the roster, so a token that precedes
//!   the queryables manufactures false negatives. RFC 08 §6.1's bounded
//!   grace exists to tolerate declarations spawned concurrently at
//!   startup; this API sequences them instead, so there is no race for a
//!   checker to be graceful about. The token is only mintable by
//!   consuming the bring-up: alive-before-queryable does not typecheck.
//! - **Vocabulary** ([`ReservedError`]): the RFC 05 §3 reserved error
//!   names as an enum, so a responder cannot misspell `error/unsupported`
//!   — and RFC 08 §6.1's conditional-procedure rule (declare always,
//!   answer `error/unsupported`/`error/gated` when gated) is writable
//!   without a string literal.
//! - **Reply key** ([`Responder::reply`]): a procedure replies on its
//!   **own concrete key**, never by echoing `query.key_expr()` (RFC 05
//!   §2.1: consolidation keeps one reply *per reply key*, so a fleet
//!   echoing a shared wildcard selector consolidates down to one
//!   survivor). The responder holds its declared key and replies on it;
//!   the query's key is never consulted for the *reply*.
//! - **Target** ([`BringUp::serve_write`]): a `kind = "write"` procedure
//!   whose entry is not `fanout = "allowed"` answers only a query whose key
//!   expression **is** its own concrete key, and refuses every other with
//!   `error/fanout-forbidden` before the handler sees it (RFC 05 §2.1,
//!   v1.38). The builder and the registry refuse a broadcast write on the
//!   caller's side; a raw `get` has neither, and the ACL cannot stand in —
//!   deny is by inclusion (RFC 09 §3 fact 6), so `v1/*/@rpc/**` walks past
//!   any producer-scoped rule. The query's key *is* consulted here, and only
//!   here.

use crate::{Error, Result};
use zenoh::Session;
use zenoh::handlers::FifoChannelHandler;
use zenoh::liveliness::LivelinessToken;
use zenoh::query::{Query, Queryable};

/// The RFC 05 §3 reserved error vocabulary — the names every conforming
/// caller understands, as a closed enum. Producer-specific names live under
/// `error/<producer>/…` and are registered like subjects; these seven are the
/// convention's own.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum ReservedError {
    /// `error/invalid-args` — the request payload does not parse or fails
    /// validation.
    InvalidArgs,
    /// `error/unauthorized` — the caller may not invoke this procedure.
    Unauthorized,
    /// `error/not-found` — the named entity does not exist here.
    NotFound,
    /// `error/unsupported` — the capability is absent from this build
    /// (RFC 08 §6.1: a conditional procedure still declares, and answers
    /// with this — "producer present, capability not in this build →
    /// rebuild").
    Unsupported,
    /// `error/busy` — present and capable, but not now.
    Busy,
    /// `error/gated` — the capability is built in but disabled here by
    /// policy or configuration (RFC 05 §3: gated writes keep their gate at
    /// the server; RFC 08 §6.1: "capability built in, disabled here →
    /// reconfigure").
    Gated,
    /// `error/fanout-forbidden` — the query's key expression is not this
    /// procedure's own concrete key: a broadcast reached a write whose entry
    /// is not `fanout = "allowed"` (RFC 05 §2.1, v1.38). Sent by a
    /// [`Responder`] declared through [`BringUp::serve_write`], before the
    /// handler sees the query; the message is [`zenkey::ExactKeyError`]'s.
    FanoutForbidden,
}

impl ReservedError {
    /// Every reserved name, for iteration (renderers, doctors).
    pub const ALL: [ReservedError; 7] = [
        ReservedError::InvalidArgs,
        ReservedError::Unauthorized,
        ReservedError::NotFound,
        ReservedError::Unsupported,
        ReservedError::Busy,
        ReservedError::Gated,
        ReservedError::FanoutForbidden,
    ];

    /// The wire name, namespaced like a key (RFC 05 §3).
    pub fn name(self) -> &'static str {
        match self {
            ReservedError::InvalidArgs => "error/invalid-args",
            ReservedError::Unauthorized => "error/unauthorized",
            ReservedError::NotFound => "error/not-found",
            ReservedError::Unsupported => "error/unsupported",
            ReservedError::Busy => "error/busy",
            ReservedError::Gated => "error/gated",
            ReservedError::FanoutForbidden => "error/fanout-forbidden",
        }
    }

    /// Read a wire error name back into the vocabulary: `Some` for one of
    /// the seven, `None` for anything else — a producer's own
    /// `error/<producer>/…` name (RFC 08 §2, v1.40) as much as a stranger's.
    /// What a judge that *receives* an envelope branches on (#222), so the
    /// spelling it compares against is this enum's and nobody's copy.
    pub fn parse(name: &str) -> Option<ReservedError> {
        ReservedError::ALL.into_iter().find(|e| e.name() == name)
    }

    /// The RFC 05 §3 error envelope, `{ "error": <name>, "message": … }`,
    /// serialized as JSON (the RFC shows the envelope as JSON for
    /// readability; a deployment whose payload default differs encodes the
    /// same two fields itself).
    pub fn envelope(self, message: &str) -> Vec<u8> {
        serde_json::to_vec(&serde_json::json!({
            "error": self.name(),
            "message": message,
        }))
        .expect("two string fields serialize")
    }
}

/// One declared `@rpc` queryable, bound to its own concrete key — the only
/// key it will ever reply on, and (for a write) the only key it will answer.
pub struct Responder {
    key: String,
    /// The same key, parsed once, for [`zenkey::require_exact`].
    own: zenoh::key_expr::KeyExpr<'static>,
    /// Whether this responder answers only its own concrete key
    /// ([`BringUp::serve_write`]) or any query that intersects it
    /// ([`BringUp::serve`]).
    exact: bool,
    queryable: Queryable<FifoChannelHandler<Query>>,
}

impl std::fmt::Debug for Responder {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Responder")
            .field("key", &self.key)
            .field("exact", &self.exact)
            .finish_non_exhaustive()
    }
}

impl Responder {
    /// The concrete key this responder was declared on — and replies on.
    pub fn key(&self) -> &str {
        &self.key
    }

    /// The next query, or `None` once the queryable is gone.
    ///
    /// On a responder declared through [`BringUp::serve_write`], a query
    /// whose key expression is not this responder's own concrete key never
    /// comes out of here: it is answered `error/fanout-forbidden` on the
    /// spot, with [`zenkey::ExactKeyError`]'s text as the message, and the
    /// wait continues (RFC 05 §2.1, v1.38). A refusal that could not be sent
    /// is dropped with the query — the caller was a broadcast, and there is
    /// nobody to report to.
    pub async fn next(&self) -> Option<Query> {
        loop {
            let query = self.queryable.recv_async().await.ok()?;
            if !self.exact {
                return Some(query);
            }
            match zenkey::require_exact(query.key_expr(), &self.own) {
                Ok(()) => return Some(query),
                Err(e) => {
                    let _ = self
                        .reply_err(&query, ReservedError::FanoutForbidden, &e.to_string())
                        .await;
                }
            }
        }
    }

    /// The queries as a [`Stream`](futures_core::Stream) (#343).
    ///
    /// Borrows, which is the point: `reply` and `reply_err` take `&self`, so
    /// a query pulled from this stream can still be answered while the stream
    /// is held — the same receive/answer split [`next`](Self::next) has, and
    /// the same refusal of a non-exact query on a write responder.
    pub fn stream(&self) -> impl futures_core::Stream<Item = Query> + '_ {
        futures_util::stream::unfold((), move |()| async move {
            self.next().await.map(|query| (query, ()))
        })
    }

    /// Reply a value on this responder's **own concrete key** (RFC 05
    /// §2.1). `query.key_expr()` is deliberately never consulted: echoing
    /// the query's selector puts a fleet's replies on one shared wildcard
    /// key, and default consolidation keeps one survivor per reply key.
    pub async fn reply(
        &self,
        query: &Query,
        payload: Vec<u8>,
        encoding: Option<&str>,
    ) -> Result<()> {
        self.reply_stamped(query, payload, encoding, None).await
    }

    /// [`reply`](Self::reply), with an explicit HLC when the producer mints
    /// one (`session.new_timestamp()`) — the reply-side twin of
    /// [`crate::Publication::send_stamped`].
    ///
    /// A deployment's `timestamping.enabled` stamps *publications* on the
    /// way past; a query reply is stamped only when the responder stamps
    /// it. That stamp is what gives a traced call (#215) its HLC reference:
    /// without it every effect's `hlc_delta_ms` is absent, because the
    /// caller's own session mints no HLC to measure from.
    pub async fn reply_stamped(
        &self,
        query: &Query,
        payload: Vec<u8>,
        encoding: Option<&str>,
        timestamp: Option<zenoh::time::Timestamp>,
    ) -> Result<()> {
        let reply = query.reply(self.key.clone(), payload);
        let reply = match encoding {
            Some(e) => reply.encoding(e),
            None => reply,
        };
        let reply = match timestamp {
            Some(ts) => reply.timestamp(ts),
            None => reply,
        };
        reply.await.map_err(|e| Error::bus("reply", &self.key, e))
    }

    /// Refuse on Zenoh's reply-error channel with a [`ReservedError`]
    /// envelope (RFC 05 §3: a value reply always means success; a failure
    /// always rides `reply_err` — never a success payload carrying
    /// `ok: false`).
    pub async fn reply_err(
        &self,
        query: &Query,
        error: ReservedError,
        message: &str,
    ) -> Result<()> {
        query
            .reply_err(error.envelope(message))
            .encoding("application/json")
            .await
            .map_err(|e| Error::bus("reply_err", &self.key, e))
    }

    /// Undeclare, acknowledged.
    pub async fn undeclare(self) -> Result<()> {
        self.queryable
            .undeclare()
            .await
            .map_err(|e| Error::bus("undeclare", &self.key, e))
    }
}

/// The ordered bring-up (RFC 04 §5): queryables first, `alive` last — as
/// one API in which the wrong order is unrepresentable.
///
/// ```no_run
/// # async fn demo(session: zenoh::Session) -> zenkey_fleet::Result<()> {
/// let mut up = zenkey_fleet::bus::producer::BringUp::new(&session);
/// up.serve("v1/h-3fa9c2d41b7e/@rpc/sysinfo/introspect").await?;
/// up.serve("v1/h-3fa9c2d41b7e/@rpc/sysinfo/describe").await?;
/// // The token is mintable only by consuming the bring-up: every
/// // queryable above is declared (awaited, not spawned) before it.
/// let live = up.alive("v1/h-3fa9c2d41b7e/state/sysinfo/alive").await?;
/// # let _ = live; Ok(()) }
/// ```
///
/// Keys are written base-relative here because a *producer's* session is
/// namespaced (the deployment base is session config, RFC 03 §1.1); an
/// un-namespaced explorer passes full wire keys, which is equally fine —
/// the API is a string seam either way, concreteness enforced.
#[derive(Debug)]
pub struct BringUp<'a> {
    session: &'a Session,
    responders: Vec<Responder>,
}

impl<'a> BringUp<'a> {
    /// Start a bring-up: nothing is declared yet, and no `alive` token can
    /// exist before [`alive`](Self::alive) consumes this value.
    pub fn new(session: &'a Session) -> Self {
        BringUp {
            session,
            responders: Vec::new(),
        }
    }

    /// Declare one `@rpc` queryable on the producer's **own concrete key**,
    /// awaited before return (no spawn race, so RFC 08 §6.1's bounded
    /// grace has nothing to tolerate). Refused:
    ///
    /// - a wildcard key — a producer serves its own concrete keys, and a
    ///   concrete declared key is what makes [`Responder::reply`]'s
    ///   reply-key discipline meaningful (RFC 05 §2.1);
    /// - `complete` is not a parameter at all: `@rpc` queryables are
    ///   **never** declared complete (RFC 05 §2.1 — one complete queryable
    ///   short-circuits `BestMatching` callers to a single reply).
    pub async fn serve(&mut self, key: &str) -> Result<&Responder> {
        self.declare(key, false).await
    }

    /// Declare a `kind = "write"` procedure whose entry is not
    /// `fanout = "allowed"`: everything [`serve`](Self::serve) does, and the
    /// responder answers **only** a query whose key expression is its own
    /// concrete key — any other is refused `error/fanout-forbidden` inside
    /// [`Responder::next`], before the handler sees it (RFC 05 §2.1, v1.38).
    ///
    /// Reads keep [`serve`](Self::serve): a fan-in read (`introspect`,
    /// `describe`, a per-host `range`) is *supposed* to answer a wildcard
    /// query, and refusing it would break every fleet-wide `get`.
    pub async fn serve_write(&mut self, key: &str) -> Result<&Responder> {
        self.declare(key, true).await
    }

    async fn declare(&mut self, key: &str, exact: bool) -> Result<&Responder> {
        let parsed = zenoh::key_expr::KeyExpr::try_from(key.to_string())
            .map_err(|e| Error::bus("declare queryable", key, e))?;
        if parsed.is_wild() {
            // The caller handed us the key; nothing was declared.
            return Err(Error::unaskable(
                key.to_string(),
                "a producer serves its own concrete key, never a wildcard \
                 (RFC 05 §2.1 — replies are attributed by their concrete reply \
                 key)",
            ));
        }
        let queryable = crate::bus::teardown::declared(
            "declare queryable",
            key,
            self.session
                .declare_queryable(parsed.clone())
                .complete(false),
        )
        .await?;
        self.responders.push(Responder {
            key: key.to_string(),
            own: parsed,
            exact,
            queryable,
        });
        Ok(self.responders.last().expect("just pushed"))
    }

    /// Declare the `alive` liveliness token — consuming the bring-up, so
    /// the token structurally cannot precede the queryables ("alive ⇒
    /// callable", RFC 04 §5).
    pub async fn alive(self, alive_key: &str) -> Result<LiveProducer> {
        let token = crate::bus::teardown::declared(
            "declare alive token",
            alive_key,
            self.session
                .liveliness()
                .declare_token(alive_key.to_string()),
        )
        .await?;
        Ok(LiveProducer {
            token: Some(token),
            responders: self.responders,
        })
    }

    /// Finish **without** presence: hand back the declared responders and
    /// never mint a token. This is the mock/synthetic path ([`crate::
    /// generate`]'s impersonated producers, RFC 13 §5) — a tool that
    /// answers for a producer must not also claim its presence. A real
    /// producer wants [`alive`](Self::alive).
    pub fn without_alive(self) -> Vec<Responder> {
        self.responders
    }
}

/// A producer that is up: queryables declared, then `alive` — held while
/// serving.
#[derive(Debug)]
pub struct LiveProducer {
    token: Option<LivelinessToken>,
    /// The declared responders, in bring-up order — drive them
    /// ([`Responder::next`]) to serve.
    pub responders: Vec<Responder>,
}

impl LiveProducer {
    /// Retire in the reverse of bring-up: retract `alive` **first** (the
    /// roster must stop attributing silence to a producer that is going
    /// away), then undeclare the queryables, acknowledged.
    ///
    /// A token that will not retract still bails before the queryables, and
    /// deliberately: "alive ⇒ callable" (RFC 04 §5) is a claim that outlives
    /// this call, and stripping callability while presence stands would
    /// manufacture exactly the false negative the ordering exists to prevent.
    /// The queryables themselves drain (#346): one that will not undeclare
    /// must not leave the rest declared, and the failures are reported
    /// together — the `bus::teardown` shape, shared with the monitor and the
    /// replayer.
    pub async fn retire(mut self) -> Result<()> {
        if let Some(token) = self.token.take() {
            token
                .undeclare()
                .await
                .map_err(|e| Error::bus("retract alive token", "", e))?;
        }
        let declared: Vec<(String, Responder)> = self
            .responders
            .drain(..)
            .map(|r| (r.key.clone(), r))
            .collect();
        crate::bus::teardown::drain_undeclare(declared, Responder::undeclare).await
    }
}