dora-message 1.0.0

pub use crate::common::{DataMessage, LogLevel, LogMessage, SharedMemoryId, Timestamped};
use crate::{
    DataflowId, current_crate_version,
    id::{DataId, NodeId},
    metadata::Metadata,
    versions_compatible,
};

#[allow(clippy::large_enum_variant)]
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
#[non_exhaustive]
pub enum DaemonRequest {
    Register(NodeRegisterRequest),
    Subscribe,
    SendMessage {
        output_id: DataId,
        metadata: Metadata,
        data: Option<DataMessage>,
    },
    OutputSent {
        output_id: DataId,
        metadata: Metadata,
    },
    CloseOutputs(Vec<DataId>),
    /// Signals that the node is finished sending outputs.
    OutputsDone,
    NextEvent,
    EventStreamDropped,
    NodeConfig {
        node_id: NodeId,
    },
    /// Store an opaque value in the daemon's dataflow-scoped extension table.
    ///
    /// dora attaches no meaning to `namespace`, `key` or `value`: this is the
    /// out-of-band channel for transports that live outside the tree (see
    /// `docs/extensions.md`). The daemon tracks who stored a key and who has
    /// read it, so it can notify readers on removal and reclaim the entry when
    /// the dataflow ends.
    ExtensionStore {
        namespace: String,
        key: String,
        value: Vec<u8>,
    },
    /// Read an opaque value back. `remove: true` drops it in the same round
    /// trip, so a consume-once handoff needs one request rather than two.
    ExtensionLoad {
        namespace: String,
        key: String,
        remove: bool,
    },
    /// Drop an opaque value. Every node that stored or loaded the key is sent
    /// [`NodeEvent::ExtensionDropped`] so it can release whatever the value
    /// referred to.
    ExtensionDrop {
        namespace: String,
        key: String,
    },
    /// An opaque request from a node to the extension registered under
    /// `namespace` on its daemon. dora never interprets either field.
    ///
    /// Complements the extension table (`ExtensionStore`/`ExtensionLoad`/
    /// `ExtensionDrop`), which brokers descriptor *lifetime*. This one
    /// carries a call the extension's daemon half must service — the
    /// tensor-pool uses it for cross-machine registration and pool writes.
    ExtensionRequest {
        namespace: String,
        #[serde(with = "crate::bulk_bytes::vec")]
        payload: Vec<u8>,
    },
}

impl DaemonRequest {
    /// Bulk bytes this request will contribute to its encoding, for
    /// [`crate::encode_presized`].
    ///
    /// Matched exhaustively on purpose: a new payload-carrying variant that
    /// forgets to report its size would silently fall back to growing the
    /// buffer from empty, which is the cost `encode_presized` exists to avoid
    /// and which no test would catch.
    pub fn encode_size_hint(&self) -> usize {
        match self {
            DaemonRequest::SendMessage { data, .. } => data.as_ref().map_or(0, DataMessage::len),
            DaemonRequest::Register(_)
            | DaemonRequest::Subscribe
            | DaemonRequest::OutputSent { .. }
            | DaemonRequest::CloseOutputs(_)
            | DaemonRequest::OutputsDone
            | DaemonRequest::NextEvent
            | DaemonRequest::EventStreamDropped
            | DaemonRequest::NodeConfig { .. } => 0,
            // The stored value dominates; the namespace and key are short.
            DaemonRequest::ExtensionStore { value, .. } => value.len(),
            DaemonRequest::ExtensionLoad { .. } | DaemonRequest::ExtensionDrop { .. } => 0,
            DaemonRequest::ExtensionRequest { payload, .. } => payload.len(),
        }
    }

    pub fn expects_tcp_binary_reply(&self) -> bool {
        #[allow(clippy::match_like_matches_macro)]
        match self {
            DaemonRequest::SendMessage { .. }
            | DaemonRequest::OutputSent { .. }
            | DaemonRequest::NodeConfig { .. } => false,
            DaemonRequest::Register(NodeRegisterRequest { .. })
            | DaemonRequest::Subscribe
            | DaemonRequest::CloseOutputs(_)
            | DaemonRequest::OutputsDone
            | DaemonRequest::NextEvent
            | DaemonRequest::EventStreamDropped
            | DaemonRequest::ExtensionRequest { .. }
            | DaemonRequest::ExtensionStore { .. }
            | DaemonRequest::ExtensionLoad { .. }
            | DaemonRequest::ExtensionDrop { .. } => true,
        }
    }

    pub fn expects_tcp_json_reply(&self) -> bool {
        #[allow(clippy::match_like_matches_macro)]
        match self {
            DaemonRequest::NodeConfig { .. } => true,
            DaemonRequest::Register(NodeRegisterRequest { .. })
            | DaemonRequest::Subscribe
            | DaemonRequest::CloseOutputs(_)
            | DaemonRequest::OutputsDone
            | DaemonRequest::NextEvent
            | DaemonRequest::SendMessage { .. }
            | DaemonRequest::OutputSent { .. }
            | DaemonRequest::EventStreamDropped
            | DaemonRequest::ExtensionRequest { .. }
            | DaemonRequest::ExtensionStore { .. }
            | DaemonRequest::ExtensionLoad { .. }
            | DaemonRequest::ExtensionDrop { .. } => false,
        }
    }
}

#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct NodeRegisterRequest {
    pub dataflow_id: DataflowId,
    pub node_id: NodeId,
    dora_version: semver::Version,
    /// The metadata wire-format version this node was built against
    /// ([`Metadata::CURRENT_VERSION`]).
    ///
    /// The `dora_version` semver check above is too coarse to catch a
    /// message-format break within a release series: two builds that both
    /// report `1.0.0-rc` are "compatible" by semver even when the payload
    /// layout differs (as it did across #2366). Carrying the format version
    /// explicitly lets the daemon reject an incompatible peer at register with
    /// a clear message, instead of the node desyncing mid-stream into a
    /// cryptic bad-enum-discriminant deserialization error (#2742).
    ///
    /// This covers *layout* drift within one encoding. It cannot cover a change
    /// of the encoding itself: the register frame is encoded the same way as
    /// everything else, so a peer from before the bincode→postcard move fails
    /// while decoding the frame that carries this field, never reaching the
    /// check. Such changes are gated by the release notes instead.
    metadata_version: u16,
}

impl NodeRegisterRequest {
    pub fn new(dataflow_id: DataflowId, node_id: NodeId) -> Self {
        Self {
            dataflow_id,
            node_id,
            dora_version: semver::Version::parse(env!("CARGO_PKG_VERSION")).unwrap(),
            metadata_version: Metadata::CURRENT_VERSION,
        }
    }

    pub fn check_version(&self) -> Result<(), String> {
        let crate_version = current_crate_version();
        let specified_version = &self.dora_version;

        if !versions_compatible(&crate_version, specified_version)? {
            return Err(format!(
                "version mismatch: message format v{} is not compatible \
                with expected message format v{crate_version}",
                self.dora_version
            ));
        }

        // Even when the semver check passes, the payload wire format can still
        // differ within a release series (#2366 dropped a `Metadata` field
        // without changing the version). Reject that here so the failure is a
        // legible register-time error rather than a mid-stream desync (#2742).
        if self.metadata_version != Metadata::CURRENT_VERSION {
            return Err(format!(
                "message wire-format mismatch: node speaks metadata format v{} \
                but this daemon speaks v{}. The node and daemon were built from \
                dora revisions with incompatible message formats; rebuild both \
                from the same revision.",
                self.metadata_version,
                Metadata::CURRENT_VERSION
            ));
        }

        Ok(())
    }
}

#[derive(Debug, serde::Deserialize, serde::Serialize)]
pub enum DynamicNodeEvent {
    NodeConfig { node_id: NodeId },
}

#[cfg(test)]
mod register_version_tests {
    use super::*;

    fn request() -> NodeRegisterRequest {
        NodeRegisterRequest::new(uuid::Uuid::nil(), NodeId::from("test-node".to_string()))
    }

    #[test]
    fn new_stamps_current_metadata_version() {
        assert_eq!(request().metadata_version, Metadata::CURRENT_VERSION);
    }

    #[test]
    fn check_version_accepts_a_matching_request() {
        // A request built by this same crate version passes both the semver
        // and the wire-format gate.
        request().check_version().unwrap();
    }

    #[test]
    fn check_version_rejects_a_wire_format_mismatch() {
        // A peer built from a dora revision with a different metadata wire
        // format — same semver, incompatible bytes. This is the #2366 / #2742
        // shape: it must be rejected at register with a legible message rather
        // than desyncing mid-stream into a cryptic deserialization error.
        let mut req = request();
        req.metadata_version = Metadata::CURRENT_VERSION.wrapping_add(1);
        let err = req
            .check_version()
            .expect_err("mismatched metadata wire version must be rejected");
        assert!(
            err.contains("wire-format") && err.contains(&Metadata::CURRENT_VERSION.to_string()),
            "error should name the wire-format mismatch and the expected version, got: {err}"
        );
    }
}