dora-node-api 1.0.0

use crate::{
    DaemonCommunicationWrapper, EventStream, NodeError, NodeResult,
    daemon_connection::{DaemonChannel, IntegrationTestingEvents},
    integration_testing::{
        TestingCommunication, TestingInput, TestingOptions, TestingOutput,
        take_testing_communication,
    },
};

use self::{arrow_utils::ipc_encode, control_channel::ControlChannel};
use aligned_vec::{AVec, ConstAlign};
use arrow::array::{Array, ArrayData};
use colored::Colorize;
use dora_arrow_convert::{DoraArray, IntoArrow};
use dora_core::{
    config::{DataId, NodeId, NodeRunConfig},
    descriptor::Descriptor,
    topics::{DORA_DAEMON_LOCAL_LISTEN_PORT_DEFAULT, DORA_DAEMON_LOCAL_LISTEN_PORT_ENV, LOCALHOST},
    types::TypeRegistry,
    uhlc,
};
use dora_message::{
    DataflowId,
    daemon_to_node::{DaemonCommunication, DaemonReply, NodeConfig, OutputRouting},
    metadata::{
        FIN, FLUSH, FRAMING, FRAMING_ARROW_IPC, Metadata, MetadataParameters, Parameter,
        SCHEMA_HASH, SEGMENT_ID, SEQ, SESSION_ID,
    },
    node_to_daemon::{DaemonRequest, DataMessage, Timestamped},
};
use eyre::WrapErr;
use is_terminal::IsTerminal;

use std::{
    collections::{BTreeMap, BTreeSet, HashMap},
    path::PathBuf,
    sync::{
        Arc, Mutex,
        atomic::{AtomicBool, Ordering},
    },
    time::{Duration, Instant},
};
#[cfg(feature = "tracing")]
use tokio::runtime::Handle;

#[cfg(feature = "tracing")]
use dora_tracing::{OtelGuard, TracingBuilder};
use tracing::{debug, error, info, warn};

pub mod arrow_utils;
mod control_channel;

/// Runtime type checking mode, controlled by `DORA_RUNTIME_TYPE_CHECK` env var.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum RuntimeTypeCheck {
    /// No runtime type checking (default).
    Off,
    /// Log warnings on type mismatches.
    Warn,
    /// Return errors on type mismatches.
    Error,
}

impl RuntimeTypeCheck {
    fn from_env() -> Self {
        Self::from_value(std::env::var("DORA_RUNTIME_TYPE_CHECK").ok().as_deref())
    }

    /// Parse the `DORA_RUNTIME_TYPE_CHECK` value (`None` when the var is unset).
    fn from_value(value: Option<&str>) -> Self {
        match value {
            Some("error") => Self::Error,
            Some("1" | "warn" | "true" | "on") => Self::Warn,
            // Accept the natural "disable" spellings without a warning: a user
            // who sets `=0`/`=false`/`=off` to turn the feature off means to
            // disable it, not to type an unrecognized value.
            Some("" | "0" | "false" | "off") | None => Self::Off,
            Some(other) => {
                tracing::warn!(
                    "unknown DORA_RUNTIME_TYPE_CHECK value \"{other}\", \
                     expected \"warn\" or \"error\"; disabling runtime type check"
                );
                Self::Off
            }
        }
    }
}

#[cfg(test)]
mod runtime_type_check_tests {
    use super::RuntimeTypeCheck;

    #[test]
    fn parses_enable_spellings() {
        for v in ["1", "warn", "true", "on"] {
            assert_eq!(
                RuntimeTypeCheck::from_value(Some(v)),
                RuntimeTypeCheck::Warn
            );
        }
        assert_eq!(
            RuntimeTypeCheck::from_value(Some("error")),
            RuntimeTypeCheck::Error
        );
    }

    #[test]
    fn disable_spellings_and_unset_are_off() {
        for v in ["", "0", "false", "off"] {
            assert_eq!(RuntimeTypeCheck::from_value(Some(v)), RuntimeTypeCheck::Off);
        }
        assert_eq!(RuntimeTypeCheck::from_value(None), RuntimeTypeCheck::Off);
    }

    #[test]
    fn unknown_value_falls_back_to_off() {
        assert_eq!(
            RuntimeTypeCheck::from_value(Some("maybe")),
            RuntimeTypeCheck::Off
        );
    }
}

/// The data size threshold at which we start using shared memory.
///
/// Shared memory works by sharing memory pages. This means that the smallest
/// memory region that can be shared is one memory page, which is typically
/// 4KiB.
///
/// Using shared memory for messages smaller than the page size still requires
/// sharing a full page, so we have some memory overhead. We also have some
/// performance overhead because setting up a shared segment is not free. For
/// small messages it is cheaper to copy them into a heap-buffered publish.
///
/// On the zenoh data plane this threshold selects *how* an output is
/// published: payloads at or above it go through zenoh shared memory
/// (zero-copy for local subscribers), while smaller payloads are published via
/// zenoh with a heap-buffered `put`. A large payload that did not get a
/// shared-memory buffer takes the reliable daemon path instead of the zenoh
/// one, because a fragmented express publish would be silently dropped
/// (dora-rs/dora#2366). See [`DoraNode::zero_copy_threshold`] for the runtime
/// value (overridable via `DORA_ZERO_COPY_THRESHOLD`).
pub const ZERO_COPY_THRESHOLD: usize = 4096;

/// How many large outbound sends are traced hop-by-hop
/// (dora-rs/dora#2742 diagnostic; see [`DoraNode::send_output_sample`]).
///
/// The Windows nightly wedge happens on a node's *first* large output, so a
/// handful of traced sends is enough to name the blocked call while keeping a
/// healthy run's logs clean.
const LARGE_SEND_DIAG_LIMIT: u32 = 3;

/// How often a starting node re-publishes its startup route-probe markers.
///
/// See [`StartupHandshake`]. Markers stop per output as soon as that output's
/// required acks have arrived (or the grace boundary froze it on the daemon
/// path), so this rate only applies while the handshake is in flight.
const ZENOH_STARTUP_MARKER_INTERVAL: Duration = Duration::from_millis(5);

/// The whole budget the startup handshake gets, measured from the daemon's
/// "all nodes ready" barrier: `init` waits this long for the handshake, and
/// whatever is still un-acked at the end is frozen on the daemon path for the
/// run (see [`wait_for_grace`] for why the freeze is unconditional).
///
/// The window starts at the barrier, not at spawn: by then every static
/// consumer has declared its subscribers and ack publishers, so a consumer
/// that is merely slow to *start* (a Python node importing heavy libraries for
/// a minute) cannot burn it. The healthy case completes in one marker→ack
/// round-trip (single-digit milliseconds); half a second of 5 ms markers with
/// no ack means the route is broken, not slow. Nothing fails at the boundary —
/// a frozen output just keeps riding the lossless daemon path, trading the
/// fast path for ordering. This is the single knob for that trade: raising it
/// gives slow-to-establish routes more chance at direct zenoh, at the cost of
/// delaying every node whose routes are genuinely broken.
///
/// For a **dynamic or restarted** producer the barrier releases immediately —
/// its consumers have long been running — so this window instead starts a few
/// milliseconds after the zenoh session opens, while the peer links it needs
/// may still be dialing. Such a producer is therefore the most likely to end
/// up frozen on the daemon path; if that shows up as a measurable regression,
/// this constant (not a late upgrade) is the thing to raise.
const ZENOH_STARTUP_GRACE: Duration = Duration::from_millis(500);

/// Poll interval for the post-barrier grace wait.
const ZENOH_STARTUP_GRACE_POLL_INTERVAL: Duration = Duration::from_millis(2);

/// A declared direct-zenoh data publisher plus its startup-handshake state.
struct DirectOutput {
    publisher: zenoh::pubsub::Publisher<'static>,
    /// `false` until the startup handshake proves this output's routes — every
    /// required consumer acked one of its markers (see [`StartupHandshake`]).
    /// Settled by [`wait_for_grace`] before `init` returns and immutable from
    /// then on, so every send of a given output takes the same path: direct
    /// zenoh when `true`, the reliable daemon path when `false`.
    ready: Arc<AtomicBool>,
}

type ZenohPublishers = HashMap<DataId, DirectOutput>;

/// Declare a direct-zenoh data publisher for every output that may ever take
/// the direct path, plus the per-output ack state the startup handshake needs.
///
/// Outputs the daemon pinned `daemon_only` — a consumer only inter-daemon
/// forwarding can reach (a dynamic node on another daemon, or a remote static
/// one with no dialable endpoint for this node), and forwarding is fed solely
/// by daemon-path sends (#2738) — get no publisher and no markers: they stay on
/// the daemon path for the node's lifetime. Every other output gets a publisher declared eagerly
/// at init (rather than on first send) for two reasons: zenoh starts wiring
/// routes immediately, and [`StartupHandshake`] needs the publishers to probe
/// those routes before the node's first real send. An output with no required
/// ackers (no consumers, or only dynamic local ones) is `ready` immediately.
///
/// QoS is set at declare time so it applies to every put: `express(true)` bypasses
/// zenoh's adaptive batch timer (the single biggest small-message latency win —
/// without it, per-put delivery on the bare local config collapses to a few
/// msg/s), `Priority::RealTime` keeps data-plane messages off the bulk-data
/// queues, and `CongestionControl::Drop` prevents a stalled subscriber from
/// back-pressuring the publishing node.
///
/// An output whose publisher fails to declare is simply absent from the map; its
/// sends then fall back to the reliable daemon path.
fn declare_output_publishers(
    session: &zenoh::Session,
    dataflow_id: DataflowId,
    node_id: &NodeId,
    outputs: &BTreeSet<DataId>,
    routing: &BTreeMap<DataId, OutputRouting>,
) -> (ZenohPublishers, Vec<Arc<AckState>>) {
    use zenoh::Wait;
    use zenoh::qos::{CongestionControl, Priority};

    let mut publishers = HashMap::new();
    let mut ack_states = Vec::new();
    for output_id in outputs {
        let Some(output_routing) = routing.get(output_id) else {
            // Defensive: the daemon computes an entry for every declared
            // output. An output it doesn't know stays on the daemon path.
            warn!(output = %output_id, "no routing entry for output; staying on the daemon path");
            continue;
        };
        if output_routing.daemon_only {
            debug!(
                output = %output_id,
                "output pinned to the daemon path (a consumer is reachable only by \
                 inter-daemon forwarding)"
            );
            continue;
        }
        let topic = dora_core::topics::zenoh_output_publish_topic(dataflow_id, node_id, output_id);
        let key_expr = match zenoh::key_expr::KeyExpr::new(topic) {
            Ok(key) => key.into_owned(),
            Err(e) => {
                warn!(output = %output_id, "invalid zenoh key ({e}); falling back to daemon path");
                continue;
            }
        };
        match session
            .declare_publisher(key_expr)
            .congestion_control(CongestionControl::Drop)
            .express(true)
            .priority(Priority::RealTime)
            .wait()
        {
            Ok(publisher) => {
                let ready = Arc::new(AtomicBool::new(output_routing.required_ackers.is_empty()));
                if !output_routing.required_ackers.is_empty() {
                    ack_states.push(Arc::new(AckState::new(
                        output_id.clone(),
                        &output_routing.required_ackers,
                        ready.clone(),
                    )));
                }
                publishers.insert(output_id.clone(), DirectOutput { publisher, ready });
            }
            Err(e) => {
                warn!(output = %output_id, "failed to declare zenoh publisher ({e}); falling back to daemon path");
            }
        }
    }
    (publishers, ack_states)
}

/// Ack bookkeeping for one output whose startup handshake is in flight.
///
/// Shared between the output's ack-subscriber callback (which records incoming
/// acks) and the [`StartupHandshake`] thread (which publishes markers until
/// completion or the freeze).
struct AckState {
    output_id: DataId,
    /// The (consumer node, input) identities that must ack before the output
    /// may switch to the direct zenoh path — the daemon's required-acker set,
    /// derived from actual placement (local static consumers only).
    required: BTreeSet<(String, String)>,
    /// Identities that have acked so far.
    received: Mutex<BTreeSet<(String, String)>>,
    /// The same flag as the output's [`DirectOutput::ready`]; flipped exactly
    /// once, when `received` covers `required` before the freeze.
    ready: Arc<AtomicBool>,
    /// Set once the grace boundary passed with this output still un-acked: the
    /// output is pinned to the daemon path and can never upgrade (see
    /// [`Self::freeze`]).
    frozen: AtomicBool,
}

impl AckState {
    fn new(
        output_id: DataId,
        required: &BTreeSet<dora_message::daemon_to_node::RequiredAcker>,
        ready: Arc<AtomicBool>,
    ) -> Self {
        Self {
            output_id,
            required: required
                .iter()
                .map(|acker| (acker.node_id.to_string(), acker.input_id.to_string()))
                .collect(),
            received: Mutex::new(BTreeSet::new()),
            ready,
            frozen: AtomicBool::new(false),
        }
    }

    /// The ack lock, recovered from poisoning: a panicking callback leaves the
    /// set intact and losing acks would silently cost the fast path.
    ///
    /// Holding this guard is what makes [`Self::record`] and [`Self::freeze`]
    /// atomic against each other, so every method that touches `received` or
    /// `frozen` goes through here.
    fn received(&self) -> std::sync::MutexGuard<'_, BTreeSet<(String, String)>> {
        self.received
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
    }

    /// Records one ack. Identities outside the required set — a dynamic or
    /// debug consumer may ack too — are ignored; they must never count toward
    /// completion. Flips `ready` once the required set is covered, unless the
    /// output was already frozen on the daemon path.
    fn record(&self, consumer_node: &str, input_id: &str) {
        let identity = (consumer_node.to_owned(), input_id.to_owned());
        if !self.required.contains(&identity) {
            return;
        }
        let mut received = self.received();
        // Read under the same lock `freeze` takes, so the two can't interleave
        // into a `ready` flip that outlives the freeze.
        if self.frozen.load(Ordering::Relaxed) {
            return;
        }
        received.insert(identity);
        if received.len() == self.required.len() {
            self.ready.store(true, Ordering::Relaxed);
        }
    }

    /// Pins this output to the daemon path for the rest of the run: no later
    /// ack may flip `ready`. Called once, at the grace boundary — see
    /// [`wait_for_grace`].
    ///
    /// Returns whether the output was actually frozen — `false` means it had
    /// already completed its handshake and keeps the direct-zenoh path. The
    /// `received` lock makes that decision atomic against a concurrent
    /// [`Self::record`]: either the ack completed the set before the freeze, or
    /// it is ignored.
    fn freeze(&self) -> bool {
        let _guard = self.received();
        if self.ready.load(Ordering::Relaxed) {
            return false;
        }
        self.frozen.store(true, Ordering::Relaxed);
        true
    }

    /// Whether this output was frozen on the daemon path (marker-thread view;
    /// no lock needed, a stale `false` just costs one more marker).
    fn is_frozen(&self) -> bool {
        self.frozen.load(Ordering::Relaxed)
    }

    /// The required identities that have not acked (for the freeze warning).
    fn missing(&self) -> Vec<String> {
        let received = self.received();
        self.required
            .difference(&received)
            .map(|(node, input)| format!("{node}/{input}"))
            .collect()
    }
}

/// Declare one exact-key ack subscriber per awaited output.
///
/// The callback records acks into the output's [`AckState`]. An output whose
/// ack subscriber fails to declare can never complete its handshake, so it is
/// removed from the awaited set right away (its `ready` flag stays `false`
/// and it keeps riding the daemon path) instead of publishing markers no ack
/// could ever answer.
fn declare_ack_subscribers(
    session: &zenoh::Session,
    dataflow_id: DataflowId,
    node_id: &NodeId,
    ack_states: &mut Vec<Arc<AckState>>,
) -> Vec<zenoh::pubsub::Subscriber<()>> {
    use zenoh::Wait;

    let mut subscribers = Vec::new();
    let mut awaited = Vec::new();
    for state in ack_states.drain(..) {
        let topic =
            dora_core::topics::zenoh_output_ack_topic(dataflow_id, node_id, &state.output_id);
        let state_cb = state.clone();
        let subscriber = session
            .declare_subscriber(topic)
            .callback(move |sample| {
                let Some(attachment) = sample.attachment() else {
                    return;
                };
                let Ok(metadata) = dora_message::decode::<Metadata>(&attachment.to_bytes()) else {
                    // Not a dora ack (foreign publisher on the ack key): ignore.
                    return;
                };
                if metadata.metadata_version() != Metadata::CURRENT_VERSION {
                    // A peer speaking another wire format cannot be attributed
                    // reliably; never count its acks.
                    return;
                }
                if let Some((consumer, input)) = metadata.startup_ack_identity() {
                    state_cb.record(consumer, input);
                }
            })
            .wait();
        match subscriber {
            Ok(subscriber) => {
                subscribers.push(subscriber);
                awaited.push(state);
            }
            Err(e) => {
                warn!(
                    output = %state.output_id,
                    "failed to declare startup-ack subscriber ({e}); output stays on the daemon path"
                );
            }
        }
    }
    *ack_states = awaited;
    subscribers
}

/// The producer half of the startup handshake: publishes route-probe markers
/// per output until that output's required consumers have acked, then lets the
/// send path switch it from the reliable daemon path to direct zenoh.
///
/// The zenoh data plane is direct node-to-node pub/sub: zenoh drops samples for
/// a subscription that hasn't propagated to this publisher yet, so a fast
/// source could otherwise lose its first messages. Rather than infer
/// route-readiness from zenoh declarations, the handshake proves it end to end
/// and in both directions: a marker rides the output's *real* topic, and the
/// consumer's ack rides the output's `@ack` topic back — an arrived ack is
/// evidence that the route pair carries data. Until then every send takes the
/// daemon path, so nothing is ever lost; a route that never proves itself only
/// costs the fast path, never correctness ([`ZENOH_STARTUP_GRACE`]).
///
/// The handshake is over by the time `init` returns: [`Self::settle`] either
/// sees an output acked or freezes it on the daemon path.
///
/// Runs on its own thread because the node blocks inside the daemon's "all
/// nodes ready" barrier while markers must already be flowing: consumers ack
/// from their subscriber callbacks (also while parked in the barrier), which is
/// what makes cycles (`a -> b -> a`, and self-loops) resolve rather than
/// deadlock — no node ever waits on another node's post-barrier progress.
///
/// Markers carry an empty payload (so they never touch shared memory) and are
/// tagged with [`dora_message::metadata::STARTUP_MARKER_PARAM`], which
/// consumers filter out before decoding — they never reach user code. This
/// works for late producers too: a dynamic node or a restarted producer runs
/// the same handshake at join time against consumers that are already running
/// (their ack publishers answer markers for the consumer's whole lifetime).
struct StartupHandshake {
    stop: Arc<AtomicBool>,
    /// The outputs whose handshake is (or was) in flight.
    ack_states: Vec<Arc<AckState>>,
    handle: Option<std::thread::JoinHandle<()>>,
    /// Per-output ack subscribers. Kept for the node's lifetime (idle once the
    /// handshake resolves) and dropped before the session in `DoraNode::drop`.
    ack_subscribers: Vec<zenoh::pubsub::Subscriber<()>>,
}

impl StartupHandshake {
    fn start(
        session: &zenoh::Session,
        dataflow_id: DataflowId,
        node_id: &NodeId,
        publishers: &Arc<ZenohPublishers>,
        mut ack_states: Vec<Arc<AckState>>,
        clock: Arc<uhlc::HLC>,
    ) -> Self {
        use zenoh::Wait;

        let stop = Arc::new(AtomicBool::new(false));
        let ack_subscribers =
            declare_ack_subscribers(session, dataflow_id, node_id, &mut ack_states);
        if ack_states.is_empty() {
            // Nothing awaits acks (no consumers, or every ack subscriber
            // failed): no markers to publish, nothing to stop.
            return Self {
                stop,
                ack_states,
                handle: None,
                ack_subscribers,
            };
        }

        let thread_stop = stop.clone();
        let thread_states = ack_states.clone();
        let thread_publishers = publishers.clone();
        let handle = std::thread::Builder::new()
            .name("dora-startup-handshake".into())
            .spawn(move || {
                loop {
                    if thread_stop.load(Ordering::Relaxed) {
                        return;
                    }
                    let mut awaiting = false;
                    for state in &thread_states {
                        // A frozen output can never upgrade, so its markers can
                        // only cost bandwidth; `wait_for_grace` already logged
                        // why it stays on the daemon path.
                        if state.ready.load(Ordering::Relaxed) || state.is_frozen() {
                            continue;
                        }
                        awaiting = true;
                        let Some(output) = thread_publishers.get(&state.output_id) else {
                            continue;
                        };
                        let metadata = Metadata::startup_marker(clock.new_timestamp());
                        let attachment = match dora_message::encode(&metadata) {
                            Ok(bytes) => bytes,
                            Err(e) => {
                                debug!(output = %state.output_id, "failed to serialize startup marker ({e})");
                                continue;
                            }
                        };
                        if let Err(e) = output
                            .publisher
                            .put(&[][..])
                            .attachment(&attachment[..])
                            .wait()
                        {
                            // Expected while the route is still coming up.
                            tracing::trace!(output = %state.output_id, "startup marker put failed ({e})");
                        }
                    }
                    if !awaiting {
                        // Every output is acked or frozen: the handshake is over.
                        return;
                    }
                    std::thread::sleep(ZENOH_STARTUP_MARKER_INTERVAL);
                }
            });
        match handle {
            Ok(handle) => Self {
                stop,
                ack_states,
                handle: Some(handle),
                ack_subscribers,
            },
            Err(e) => {
                // Without markers no consumer will ack, so the awaited outputs
                // simply stay on the reliable daemon path. Loud because the
                // fast path is silently lost for this node.
                error!(
                    "failed to spawn startup-handshake thread ({e}); outputs stay on the daemon path"
                );
                Self {
                    stop,
                    ack_states,
                    handle: None,
                    ack_subscribers,
                }
            }
        }
    }

    /// Settle every output's transport: wait out `grace`, then freeze whatever
    /// has not proven its routes. See [`wait_for_grace`].
    ///
    /// `DoraNode::init` **must** call this before returning to user code —
    /// that is what makes an output's path constant for the run
    /// (dora-rs/dora#2891). It is a method (rather than only the free function
    /// the tests drive) so the requirement is discoverable from the type.
    fn settle(&self, grace: Duration) {
        wait_for_grace(&self.ack_states, grace);
    }

    /// Signal the handshake thread to stop and wait for it to exit, so its
    /// `Arc` clone of the publishers is released. Idempotent.
    fn shutdown(&mut self) {
        self.stop.store(true, Ordering::Relaxed);
        if let Some(handle) = self.handle.take() {
            let _ = handle.join();
        }
    }
}

impl Drop for StartupHandshake {
    fn drop(&mut self) {
        // Safety net for error paths that return before `DoraNode::drop` (e.g.
        // a failed `EventStream::init`). Without this the handshake thread
        // would keep publishing and keep the publishers (and session) alive.
        self.shutdown();
    }
}

/// Post-barrier grace wait: give the handshake `grace` to complete, then freeze
/// whatever is left. This is the boundary that settles every output's transport
/// before user code runs.
///
/// Returns as soon as every awaited output is ready. Anything still un-acked
/// when `grace` expires is pinned to the daemon path for the rest of the run
/// ([`AckState::freeze`]) rather than left to upgrade later. Upgrading later is
/// the bug in dora-rs/dora#2891: user code sends from the moment this returns,
/// so *any* subsequent upgrade is mid-stream, and the direct-zenoh path has
/// fewer hops than the daemon-relay path — a message sent just after the switch
/// can overtake an earlier one still in flight on the daemon path, which the
/// consumer merges into a single arrival-ordered channel with no cross-path
/// resequencing. Freezing keeps a topic's per-input FIFO order unconditional;
/// the cost is the fast path for routes that fail to establish within `grace`.
///
/// A free function taking the states (rather than a `StartupHandshake` method)
/// so tests can exercise the boundary without zenoh, with a `grace` shorter
/// than [`ZENOH_STARTUP_GRACE`].
fn wait_for_grace(ack_states: &[Arc<AckState>], grace: Duration) {
    let grace_deadline = Instant::now() + grace;
    loop {
        // Vacuously true when nothing awaits acks (no consumers, or every ack
        // subscriber failed to declare): there is nothing to wait for.
        if ack_states
            .iter()
            .all(|state| state.ready.load(Ordering::Relaxed))
        {
            break;
        }
        if Instant::now() >= grace_deadline {
            break;
        }
        std::thread::sleep(ZENOH_STARTUP_GRACE_POLL_INTERVAL);
    }

    for state in ack_states {
        // `freeze` re-checks readiness under the ack lock, so an ack landing
        // right now either completes the handshake or is ignored — never a
        // half-applied upgrade.
        if state.freeze() {
            warn!(
                output = %state.output_id,
                missing = ?state.missing(),
                "startup handshake incomplete after {}ms; output stays on the \
                 reliable daemon path for the rest of the run",
                grace.as_millis()
            );
        } else {
            // The positive half of the same decision, and the only signal that
            // an output is *off* the daemon path — which for a consumer on
            // another machine means its data no longer crosses two daemons.
            // Logged per output, once, at the moment it is settled for the run.
            debug!(
                output = %state.output_id,
                "startup handshake complete; output takes the direct zenoh path"
            );
        }
    }
}

/// The per-output routing the daemon computed for this node, or the safe
/// all-daemon-path fallback when it provided none.
///
/// A missing map means the node was spawned by an older daemon (or an
/// interactive/manual setup) that doesn't know about the startup handshake.
/// Without required-acker sets no route can be proven, so every output stays
/// on the reliable daemon path — correct, just without the zenoh fast path.
fn normalize_output_routing(
    routing: Option<BTreeMap<DataId, OutputRouting>>,
    outputs: &BTreeSet<DataId>,
) -> BTreeMap<DataId, OutputRouting> {
    match routing {
        Some(routing) => routing,
        None => {
            if !outputs.is_empty() {
                warn!(
                    "node config carries no output routing (spawned by an older daemon?); \
                     all outputs stay on the reliable daemon path"
                );
            }
            outputs
                .iter()
                .map(|output_id| {
                    (
                        output_id.clone(),
                        OutputRouting {
                            daemon_only: true,
                            required_ackers: Default::default(),
                        },
                    )
                })
                .collect()
        }
    }
}

/// Per-phase deadline for tearing down zenoh state on node shutdown (subscribers,
/// liveliness tokens, and the session/publishers — see [`DoraNode::drop`] and
/// `EventStream::drop`). Each `undeclare`/close blocks indefinitely when zenoh's net
/// runtime is wedged (e.g. retrying an unreachable scouted peer on a headless CI
/// runner), so it is bounded here and abandoned on timeout.
///
/// This MUST stay comfortably below the daemon's force-kill grace
/// (`DEFAULT_STOP_GRACE (10s) + DEFAULT_STOP_GRACE/2 = 15s`, see
/// `binaries/daemon/src/running_dataflow.rs`), including the worst case where all
/// three phases wedge sequentially (`3 *` this value). Otherwise a node with a wedged
/// net runtime is still tearing down when the daemon `TerminateProcess`es it, which on
/// Windows surfaces as `ExitCode(1)` and reddens the nightly (dora-rs/dora#2742). The
/// `zenoh_teardown_fits_within_daemon_force_kill_grace` test guards the invariant.
///
/// This deliberately no longer exceeds zenoh's internal 10s session-close timeout: a
/// semi-wedged close that would settle at ~10s is abandoned instead, and peers fall
/// back to liveliness expiry. That is the right trade for a *shutting-down* node —
/// waiting out the 10s only to be force-killed anyway yields a worse (unclean) exit.
pub(crate) const ZENOH_TEARDOWN_TIMEOUT: Duration = Duration::from_secs(3);

/// Capacity of the testing-mode daemon request channel. Kept comfortably above
/// the number of concurrent requesters (event stream + close channel + control
/// channel) so Drop's CloseOutputs send does not block on a full queue while
/// the daemon thread is busy inside `next_event` (dora-rs/dora#2855).
const TESTING_DAEMON_CHANNEL_CAPACITY: usize = 256;

/// Allows sending outputs and retrieving node information.
///
/// The main purpose of this struct is to send outputs via Dora. There are also functions available
/// for retrieving the node configuration.
pub struct DoraNode {
    id: NodeId,
    dataflow_id: DataflowId,
    node_config: NodeRunConfig,
    control_channel: ControlChannel,
    clock: Arc<uhlc::HLC>,

    /// Zenoh session for direct node-to-node pub/sub (data plane).
    /// `None` in interactive/testing mode.
    zenoh_session: Option<zenoh::Session>,
    /// Zenoh shared memory provider for zero-copy publishing.
    /// Owns the SHM provider and the zero-copy threshold. Cloned out by
    /// [`DoraNode::sample_allocator`] so an operator thread can build output
    /// samples without reaching into the node (dora-rs/dora#2742).
    sample_allocator: SampleAllocator,
    /// Per-output zenoh publishers with their handshake state, declared eagerly
    /// at init (see [`declare_output_publishers`]) so zenoh wires routes
    /// immediately and [`StartupHandshake`] can probe them before the first
    /// real send. An output missing here (declaration failed, or pinned to the
    /// daemon path by the daemon's routing) falls back to the daemon path; one
    /// whose `ready` flag is still `false` (handshake in flight or frozen)
    /// does too. Shared with the handshake thread via `Arc`; the thread is
    /// joined in `drop` before the map is torn down.
    /// `'static` is sound because zenoh `Publisher` internally holds `Arc<Session>`,
    /// so it doesn't borrow from the session field on this struct.
    /// Publishers must be dropped BEFORE the session (enforced in Drop impl).
    zenoh_publishers: Arc<ZenohPublishers>,
    /// The producer half of the startup handshake (marker thread + ack
    /// subscribers). `None` without a zenoh session (interactive/testing).
    startup_handshake: Option<StartupHandshake>,
    /// Per-output schema publishers on the `@schema` subtopic (lazily created on
    /// the first small message). Their cache retains the last schema so a
    /// late-joining subscriber fetches it via a history query, letting the data
    /// topic carry only schema-less batches. Dropped before the session, like
    /// [`zenoh_publishers`](Self::zenoh_publishers).
    zenoh_schema_publishers: HashMap<DataId, zenoh_ext::AdvancedPublisher<'static>>,
    /// Per-output schema-once state: the confirmed-published schema hash (so
    /// the schema is only re-published when it changes or a publish failed) and
    /// the time of the last full-stream send (for the periodic in-band refresh).
    zenoh_schema_state: HashMap<DataId, SchemaOnceState>,
    /// Diagnostic (dora-rs/dora#2742): how many large sends have already been
    /// traced hop-by-hop. The Windows nightly wedges the *runtime's* main loop
    /// inside `send_output` on the very first large output, so tracing only the
    /// first few large sends pins the stuck hop without spamming a healthy run.
    /// Remove together with the runtime's stall watchdog once #2742 is closed.
    large_send_diag_count: u32,

    dataflow_descriptor: serde_yaml::Result<Descriptor>,
    warned_unknown_output: BTreeSet<DataId>,
    interactive: bool,
    restart_count: u32,

    /// Runtime type checking state. `None` when off (zero overhead).
    /// When `Some`, holds the mode (Warn/Error) and a map of output DataId -> expected Arrow DataType.
    runtime_type_checks: Option<(RuntimeTypeCheck, HashMap<DataId, arrow_schema::DataType>)>,

    /// Tokio runtime owned by the node. Populated only when no ambient
    /// runtime was available at init. Must drop after the zenoh session
    /// (which is drained explicitly at the top of [`Drop`]) so that any
    /// async cleanup triggered by session shutdown can still run.
    _owned_runtime: Option<tokio::runtime::Runtime>,

    /// Join handle for the in-process testing daemon thread spawned by
    /// [`Self::init_testing`] / the testing branch of `init_with_options`.
    /// Joined in [`Drop`] after closing the request channel (dora-rs/dora#2855).
    testing_daemon: Option<std::thread::JoinHandle<()>>,
    /// Signals the testing daemon to abort a scheduled `next_event` sleep so
    /// Drop's CloseOutputs handshake can complete.
    testing_shutdown: Option<Arc<AtomicBool>>,
}

impl DoraNode {
    /// Initiate a node from environment variables set by the Dora daemon or fall back to
    /// interactive mode.
    ///
    /// This is the recommended initialization function for Dora nodes, which are spawned by
    /// Dora daemon instances. The daemon will set a `DORA_NODE_CONFIG` environment variable to
    /// configure the node.
    ///
    /// When the node is started manually without the `DORA_NODE_CONFIG` environment variable set,
    /// the initialization will fall back to [`init_interactive`](Self::init_interactive) if `stdin`
    /// is a terminal (detected through
    /// [`isatty`](https://www.man7.org/linux/man-pages/man3/isatty.3.html)).
    ///
    /// If the `DORA_NODE_CONFIG` environment variable is not set and `DORA_TEST_WITH_INPUTS` is
    /// set, the node will be initialized in integration test mode. See the
    /// [integration testing](crate::integration_testing) module for details.
    ///
    /// This function will also initialize the node in integration test mode when the
    /// [`setup_integration_testing`](crate::integration_testing::setup_integration_testing)
    /// function was called before. This takes precedence over all environment variables.
    ///
    /// ```no_run
    /// use dora_node_api::DoraNode;
    ///
    /// let (mut node, mut events) = DoraNode::init_from_env().expect("Could not init node.");
    /// ```
    pub fn init_from_env() -> NodeResult<(Self, EventStream)> {
        Self::init_from_env_inner(true)
    }

    /// Initialize the node from environment variables set by the Dora daemon; error if not set.
    ///
    /// This function behaves the same as [`init_from_env`](Self::init_from_env), but it does _not_
    /// fall back to [`init_interactive`](Self::init_interactive). Instead, an error is returned
    /// when the `DORA_NODE_CONFIG` environment variable is missing.
    pub fn init_from_env_force() -> NodeResult<(Self, EventStream)> {
        Self::init_from_env_inner(false)
    }

    fn init_from_env_inner(fallback_to_interactive: bool) -> NodeResult<(Self, EventStream)> {
        if let Some(testing_comm) = take_testing_communication() {
            let TestingCommunication {
                input,
                output,
                options,
            } = *testing_comm;
            return Self::init_testing(input, output, options);
        }

        // normal execution (started by dora daemon)
        match std::env::var("DORA_NODE_CONFIG") {
            Ok(raw) => {
                let node_config: NodeConfig =
                    serde_yaml::from_str(&raw).context("failed to deserialize node config")?;
                return Self::init(node_config);
            }
            Err(std::env::VarError::NotUnicode(_)) => {
                return Err(NodeError::Init(
                    "DORA_NODE_CONFIG env variable is not valid unicode".into(),
                ));
            }
            Err(std::env::VarError::NotPresent) => {} // continue trying other init methods
        };

        // node integration test mode
        match std::env::var("DORA_TEST_WITH_INPUTS") {
            Ok(raw) => {
                let input_file = PathBuf::from(raw);
                let output_file = match std::env::var("DORA_TEST_WRITE_OUTPUTS_TO") {
                    Ok(raw) => PathBuf::from(raw),
                    Err(std::env::VarError::NotUnicode(_)) => {
                        return Err(NodeError::Init(
                            "DORA_TEST_WRITE_OUTPUTS_TO env variable is not valid unicode".into(),
                        ));
                    }
                    Err(std::env::VarError::NotPresent) => {
                        input_file.with_file_name("outputs.jsonl")
                    }
                };
                let skip_output_time_offsets =
                    std::env::var_os("DORA_TEST_NO_OUTPUT_TIME_OFFSET").is_some();

                let input = TestingInput::FromJsonFile(input_file);
                let output = TestingOutput::ToFile(output_file);
                let options = TestingOptions {
                    skip_output_time_offsets,
                };

                return Self::init_testing(input, output, options);
            }
            Err(std::env::VarError::NotUnicode(_)) => {
                return Err(NodeError::Init(
                    "DORA_TEST_WITH_INPUTS env variable is not valid unicode".into(),
                ));
            }
            Err(std::env::VarError::NotPresent) => {} // continue trying other init methods
        }

        // interactive mode
        if fallback_to_interactive && std::io::stdin().is_terminal() {
            println!(
                "{}",
                "Starting node in interactive mode as DORA_NODE_CONFIG env variable is not set"
                    .green()
            );
            return Self::init_interactive();
        }

        // no run mode applicable
        Err(NodeError::Init(
            "DORA_NODE_CONFIG env variable is not set".into(),
        ))
    }

    /// Create a builder for configuring a node connection.
    ///
    /// Setting a `node_id` selects the dynamic-node path; without one, `build()`
    /// falls back to [`init_from_env`](Self::init_from_env). Use this builder
    /// when you need a custom daemon port — the other init functions cover the
    /// common cases. Source-compatible with upstream dora 0.5.x: `.dynamic()`
    /// is accepted (no-op) so code written against upstream still compiles.
    ///
    /// ```no_run
    /// use dora_node_api::DoraNode;
    /// use dora_node_api::dora_core::config::NodeId;
    ///
    /// let (mut node, mut events) = DoraNode::builder()
    ///     .node_id(NodeId::from("plot".to_string()))
    ///     .daemon_port(6789)
    ///     .build()
    ///     .expect("Could not init node");
    /// ```
    pub fn builder() -> DoraNodeBuilder {
        DoraNodeBuilder::default()
    }

    /// Initiate a node from a dataflow id and a node id.
    ///
    /// This initialization function should be used for [_dynamic nodes_](index.html#dynamic-nodes).
    ///
    /// ```no_run
    /// use dora_node_api::DoraNode;
    /// use dora_node_api::dora_core::config::NodeId;
    ///
    /// let (mut node, mut events) = DoraNode::init_from_node_id(NodeId::from("plot".to_string())).expect("Could not init node plot");
    /// ```
    ///
    pub fn init_from_node_id(node_id: NodeId) -> NodeResult<(Self, EventStream)> {
        Self::builder().node_id(node_id).build()
    }

    /// Dynamic initialization function for nodes that are sometimes used as dynamic nodes.
    ///
    /// This function first tries initializing the traditional way through
    /// [`init_from_env`][Self::init_from_env]. If this fails, it falls back to
    /// [`init_from_node_id`][Self::init_from_node_id].
    pub fn init_flexible(node_id: NodeId) -> NodeResult<(Self, EventStream)> {
        if std::env::var("DORA_NODE_CONFIG").is_ok() {
            info!(
                "Skipping {node_id} specified within the node initialization in favor of `DORA_NODE_CONFIG` specified by `dora start`"
            );
            Self::init_from_env()
        } else {
            Self::init_from_node_id(node_id)
        }
    }

    /// Initialize the node in a standalone mode that prompts for inputs on the terminal.
    ///
    /// Instead of connecting to a `dora daemon`, this interactive mode prompts for node inputs
    /// on the terminal. In this mode, the node is completely isolated from the dora daemon and
    /// other nodes, so it cannot be part of a dataflow.
    ///
    /// Note that this function will hang indefinitely if no input is supplied to the interactive
    /// prompt. So it should be only used through a terminal.
    ///
    /// Because of the above limitations, it is not recommended to use this function directly.
    /// Use [**`init_from_env`**](Self::init_from_env) instead, which supports both normal daemon
    /// connections and manual interactive runs.
    ///
    /// ## Example
    ///
    /// Run any node that uses `init_interactive` or [`init_from_env`](Self::init_from_env) directly
    /// from a terminal. The node will then start in "interactive mode" and prompt you for the next
    /// input:
    ///
    /// ```bash
    /// > cargo build -p rust-dataflow-example-node
    /// > target/debug/rust-dataflow-example-node
    /// hello
    /// Starting node in interactive mode as DORA_NODE_CONFIG env variable is not set
    /// Node asks for next input
    /// ? Input ID
    /// [empty input ID to stop]
    /// ```
    ///
    /// The `rust-dataflow-example-node` expects a `tick` input, so let's set the input ID to
    /// `tick`. Tick messages don't have any data, so we leave the "Data" empty when prompted:
    ///
    /// ```bash
    /// Node asks for next input
    /// > Input ID tick
    /// > Data
    /// tick 0, sending 0x943ed1be20c711a4
    /// node sends output random with data: PrimitiveArray<UInt64>
    /// [
    ///   10682205980693303716,
    /// ]
    /// Node asks for next input
    /// ? Input ID
    /// [empty input ID to stop]
    /// ```
    ///
    /// We see that both the `stdout` output of the node and also the output messages that it sends
    /// are printed to the terminal. Then we get another prompt for the next input.
    ///
    /// If you want to send an input with data, you can either send it as text (for string data)
    /// or as a JSON object (for struct, string, or array data). Other data types are not supported
    /// currently.
    ///
    /// Empty input IDs are interpreted as stop instructions:
    ///
    /// ```bash
    /// > Input ID
    /// given input ID is empty -> stopping
    /// Received stop
    /// Node asks for next input
    /// event channel was stopped -> returning empty event list
    /// node reports EventStreamDropped
    /// node reports closed outputs []
    /// node reports OutputsDone
    /// ```
    ///
    /// In addition to the node output, we see log messages for the different events that the node
    /// reports. After `OutputsDone`, the node should exit.
    ///
    /// ### JSON data
    ///
    /// In addition to text input, the `Data` prompt also supports JSON objects, which will be
    /// converted to Apache Arrow struct arrays:
    ///
    /// ```bash
    /// Node asks for next input
    /// > Input ID some_input
    /// > Data { "field_1": 42, "field_2": { "inner": "foo" } }
    /// ```
    ///
    /// This JSON data is converted to the following Arrow array:
    ///
    /// ```text
    /// StructArray
    /// -- validity: [valid, ]
    /// [
    ///   -- child 0: "field_1" (Int64)
    ///      PrimitiveArray<Int64>
    ///      [42,]
    ///   -- child 1: "field_2" (Struct([Field { name: "inner", data_type: Utf8, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }]))
    ///      StructArray
    ///      -- validity: [valid,]
    ///      [
    ///        -- child 0: "inner" (Utf8)
    ///        StringArray
    ///        ["foo",]
    ///      ]
    /// ]
    /// ```
    pub fn init_interactive() -> NodeResult<(Self, EventStream)> {
        #[cfg(feature = "tracing")]
        {
            TracingBuilder::new("node")
                .with_stdout("debug", false)
                .build()
                .wrap_err("failed to set up tracing subscriber")?;
        }

        let node_config = NodeConfig {
            dataflow_id: DataflowId::new_v4(),
            node_id: "test-node"
                .parse()
                .map_err(|e| NodeError::Init(format!("{e}")))?,
            run_config: NodeRunConfig::default(),
            daemon_communication: Some(DaemonCommunication::Interactive),
            dataflow_descriptor: serde_yaml::Value::Null,
            dynamic: false,
            write_events_to: None,
            restart_count: 0,
            output_routing: None,
        };
        let (mut node, events) = Self::init(node_config)?;
        node.interactive = true;
        Ok((node, events))
    }

    /// Initializes a node in integration test mode.
    ///
    /// No connection to a dora daemon is made in this mode. Instead, inputs are read from the
    /// specified `TestingInput`, and outputs are written to the specified `TestingOutput`.
    /// Additional options for the testing mode can be specified through `TestingOptions`.
    ///
    /// It is recommended to use this function only within test functions.
    pub fn init_testing(
        input: TestingInput,
        output: TestingOutput,
        options: TestingOptions,
    ) -> NodeResult<(Self, EventStream)> {
        let node_config = NodeConfig {
            dataflow_id: DataflowId::new_v4(),
            node_id: "test-node"
                .parse()
                .map_err(|e| NodeError::Init(format!("{e}")))?,
            run_config: NodeRunConfig::default(),
            daemon_communication: None,
            dataflow_descriptor: serde_yaml::Value::Null,
            dynamic: false,
            write_events_to: None,
            restart_count: 0,
            output_routing: None,
        };
        let testing_comm = TestingCommunication {
            input,
            output,
            options,
        };
        let (mut node, events) = Self::init_with_options(node_config, Some(testing_comm))?;
        node.interactive = true;
        Ok((node, events))
    }

    /// Internal initialization routine that should not be used outside of Dora.
    #[doc(hidden)]
    #[tracing::instrument]
    pub fn init(node_config: NodeConfig) -> NodeResult<(Self, EventStream)> {
        Self::init_with_options(node_config, None)
    }

    #[tracing::instrument(skip(testing_communication))]
    fn init_with_options(
        node_config: NodeConfig,
        testing_communication: Option<TestingCommunication>,
    ) -> NodeResult<(Self, EventStream)> {
        // Before anything that can fail or block: a node spawned by `dora run`
        // must not outlive the CLI even if the rest of this initialization
        // stalls (dora-rs/dora#2856). A no-op on every other spawn path.
        crate::orphan_guard::arm_if_run_child();

        let NodeConfig {
            dataflow_id,
            node_id,
            run_config,
            daemon_communication,
            dataflow_descriptor,
            dynamic,
            write_events_to,
            restart_count,
            output_routing,
        } = node_config;
        let clock = Arc::new(uhlc::HLC::default());
        let input_config = run_config.inputs.clone();

        let (daemon_communication, testing_daemon, testing_shutdown) = match daemon_communication {
            Some(comm) => (comm.into(), None, None),
            None => match testing_communication {
                Some(comm) => {
                    let TestingCommunication {
                        input,
                        output,
                        options,
                    } = comm;
                    let (sender, mut receiver) =
                        tokio::sync::mpsc::channel(TESTING_DAEMON_CHANNEL_CAPACITY);
                    let shutdown = Arc::new(AtomicBool::new(false));
                    let new_communication = DaemonCommunicationWrapper::Testing {
                        channel: sender,
                        shutdown: shutdown.clone(),
                    };
                    let mut events =
                        IntegrationTestingEvents::new(input, output, options, shutdown.clone())?;
                    let shutdown_for_loop = shutdown.clone();
                    let handle = std::thread::Builder::new()
                        .name("dora-testing-daemon".into())
                        .spawn(move || {
                            while let Some((request, reply_sender)) = receiver.blocking_recv() {
                                let outputs_done =
                                    matches!(request.inner, DaemonRequest::OutputsDone);
                                let reply = events.request(&request);
                                if reply_sender
                                    .send(reply.unwrap_or_else(|err| {
                                        DaemonReply::Result(Err(format!("{err:?}")))
                                    }))
                                    .is_err()
                                {
                                    eprintln!("failed to send reply");
                                }
                                // Exit after OutputsDone under shutdown even if
                                // EventStream still holds a sender clone — otherwise
                                // node-first Drop waits forever on blocking_recv
                                // (dora-rs/dora#2855).
                                if outputs_done && shutdown_for_loop.load(Ordering::Relaxed) {
                                    break;
                                }
                            }
                        })
                        .map_err(|e| {
                            NodeError::Init(format!("failed to spawn testing daemon thread: {e}"))
                        })?;
                    (new_communication, Some(handle), Some(shutdown))
                }
                None => {
                    return Err(NodeError::Init(
                        "no daemon communication method specified".into(),
                    ));
                }
            },
        };

        // Initialize zenoh session for direct node-to-node data plane.
        // Skip in interactive/testing mode (no daemon, no dataflow topology).
        let is_standard_mode = matches!(
            daemon_communication,
            DaemonCommunicationWrapper::Standard(_)
        );
        // Pool size priority: per-node YAML config > env var > built-in default.
        let shm_pool_size = run_config
            .shared_memory_pool_size
            .map(|bs| bs.as_bytes())
            .or_else(|| {
                std::env::var("DORA_NODE_SHM_POOL_SIZE")
                    .ok()
                    .and_then(|s| s.parse::<usize>().ok())
            })
            // 8 MB default — kept deliberately small so the pool fits a
            // constrained `/dev/shm` (Docker/Kubernetes commonly cap it at
            // 64 MB). A pool that doesn't fit backing memory was observed to
            // break large-output delivery entirely (the segment can't be backed
            // as it fills), so bumping this default is NOT a safe way to widen
            // the large-message pipeline. Throughput under large-message bursts
            // is handled instead by the non-blocking `GarbageCollect` alloc
            // policy (see `allocate_data_sample` / `zenoh_publish`): the producer
            // never stalls on a momentarily-full pool, it just copies via the
            // heap path. Raise this only alongside a matching `/dev/shm` (via
            // `shared_memory_pool_size` / `DORA_NODE_SHM_POOL_SIZE`) to keep more
            // large outputs zero-copy.
            .unwrap_or(8 * 1024 * 1024);
        let zenoh_zero_copy_threshold = std::env::var("DORA_ZERO_COPY_THRESHOLD")
            .ok()
            .and_then(|s| s.parse::<usize>().ok())
            .unwrap_or(ZERO_COPY_THRESHOLD);
        let (zenoh_session, zenoh_shm_provider, owned_runtime) = if !is_standard_mode {
            (None, None, None)
        } else {
            let (handle, owned_runtime) = match tokio::runtime::Handle::try_current() {
                Ok(handle) => (handle, None),
                Err(_) => {
                    let rt = tokio::runtime::Builder::new_multi_thread()
                        .enable_all()
                        .thread_name("dora-node-runtime")
                        .build()
                        .map_err(|e| {
                            NodeError::Init(format!("failed to create owned tokio runtime: {e}"))
                        })?;
                    let handle = rt.handle().clone();
                    (handle, Some(rt))
                }
            };
            // Use scope + spawn to avoid panicking when called from a tokio
            // worker thread (block_on panics in that context on
            // current-thread runtimes).
            let session = std::thread::scope(|s| {
                match s
                    .spawn(|| handle.block_on(dora_core::topics::open_zenoh_session(None)))
                    .join()
                {
                    Ok(Ok(session)) => Ok(session),
                    Ok(Err(e)) => Err(NodeError::Init(format!(
                        "failed to open zenoh session: {e:?}"
                    ))),
                    Err(_panic) => Err(NodeError::Init("zenoh session init panicked".into())),
                }
            })?;
            // SHM provider is best-effort: if the OS rejects the segment
            // allocation (e.g. `/dev/shm` exhausted in CI), fall back to
            // `None`. `send_output_sample` already publishes via heap
            // buffers when the provider is missing.
            let provider = {
                use zenoh::Wait;
                use zenoh::shm::{AllocAlignment, MemoryLayout, ShmProviderBuilder};

                let alignment =
                    AllocAlignment::new(crate::arrow_utils::ARROW_BUFFER_ALIGNMENT_EXPONENT)
                        .expect("ARROW_BUFFER_ALIGNMENT is a valid power-of-two alignment");
                let layout = shm_pool_size
                    .checked_next_multiple_of(crate::arrow_utils::ARROW_BUFFER_ALIGNMENT)
                    .and_then(|aligned| MemoryLayout::new(aligned, alignment).ok());

                match layout {
                    Some(layout) => match ShmProviderBuilder::default_backend(layout).wait() {
                        Ok(provider) => Some(Arc::new(provider)),
                        Err(e) => {
                            warn!(
                                "failed to create zenoh SHM provider ({e}); \
                                 falling back to heap-buffered publishes"
                            );
                            None
                        }
                    },
                    None => {
                        warn!(
                            "invalid zenoh SHM pool size ({shm_pool_size}); \
                             falling back to heap-buffered publishes"
                        );
                        None
                    }
                }
            };
            (Some(session), provider, owned_runtime)
        };

        // Declare output publishers and start the startup handshake *before*
        // `EventStream::init`, which blocks in the daemon's "all nodes ready"
        // barrier: markers must be in flight while we are parked there so that
        // consumers (whose subscriber callbacks ack them, also while parked)
        // can prove the routes. This ordering is what keeps cycles
        // (`a -> b -> a`) and self-loops from deadlocking — every node emits
        // markers before it waits on anyone, and no one blocks on acks.
        let (zenoh_publishers, startup_handshake) = match zenoh_session.as_ref() {
            Some(session) => {
                let routing = normalize_output_routing(output_routing, &run_config.outputs);
                let (publishers, ack_states) = declare_output_publishers(
                    session,
                    dataflow_id,
                    &node_id,
                    &run_config.outputs,
                    &routing,
                );
                let publishers = Arc::new(publishers);
                let handshake = StartupHandshake::start(
                    session,
                    dataflow_id,
                    &node_id,
                    &publishers,
                    ack_states,
                    clock.clone(),
                );
                (publishers, Some(handshake))
            }
            None => (Arc::new(HashMap::new()), None),
        };

        let event_stream = EventStream::init(
            dataflow_id,
            &node_id,
            &daemon_communication,
            input_config,
            &run_config.input_types,
            clock.clone(),
            write_events_to,
            zenoh_session.as_ref(),
        )
        .wrap_err("failed to init event stream")?;

        // The barrier has released: every static consumer is subscribed and
        // acking, so the handshake now gets its bounded window. This settles
        // every output's transport — acked onto direct zenoh, or frozen on the
        // daemon path — before user code sends its first message.
        if let Some(handshake) = &startup_handshake {
            handshake.settle(ZENOH_STARTUP_GRACE);
        }
        let control_channel =
            ControlChannel::init(dataflow_id, &node_id, &daemon_communication, clock.clone())
                .wrap_err("failed to init control channel")?;
        let runtime_type_checks = match RuntimeTypeCheck::from_env() {
            RuntimeTypeCheck::Off => None,
            mode => {
                let registry = TypeRegistry::new();
                let mut checks = HashMap::new();
                for (id, urn) in &run_config.output_types {
                    match registry.resolve_arrow_type(urn) {
                        Some(dt) => {
                            checks.insert(id.clone(), dt);
                        }
                        None => {
                            if registry.resolve(urn).is_some() {
                                info!(
                                    "runtime type check: skipping complex type \"{urn}\" on output \"{id}\""
                                );
                            } else {
                                warn!(
                                    "runtime type check: unknown type URN \"{urn}\" on output \"{id}\""
                                );
                            }
                        }
                    }
                }
                Some((mode, checks))
            }
        };

        let node = Self {
            id: node_id,
            dataflow_id,
            node_config: run_config.clone(),
            control_channel,
            clock,
            zenoh_session,
            zenoh_publishers,
            startup_handshake,
            zenoh_schema_publishers: HashMap::new(),
            zenoh_schema_state: HashMap::new(),
            sample_allocator: SampleAllocator {
                shm_provider: zenoh_shm_provider,
                zero_copy_threshold: zenoh_zero_copy_threshold,
            },
            large_send_diag_count: 0,
            dataflow_descriptor: serde_yaml::from_value(dataflow_descriptor),
            warned_unknown_output: BTreeSet::new(),
            interactive: false,
            restart_count,
            runtime_type_checks,
            _owned_runtime: owned_runtime,
            testing_daemon,
            testing_shutdown,
        };

        if dynamic {
            // Env vars from the dataflow descriptor are already injected by the
            // daemon at spawn time via `Command::env()`.  Setting them here with
            // `std::env::set_var` would be undefined behavior because the tokio
            // multi-threaded runtime is already running and other threads may
            // call `std::env::var` concurrently.
            //
            // If the node was started outside the daemon (manual dynamic node),
            // the user must set the required env vars before launching the
            // process.
            if let Ok(descriptor) = &node.dataflow_descriptor
                && let Some(env_vars) = descriptor
                    .nodes
                    .iter()
                    .find(|n| n.id == node.id)
                    .and_then(|n| n.env.as_ref())
            {
                for key in env_vars.keys() {
                    if std::env::var(key).is_err() {
                        warn!(
                            "env var `{key}` declared in dataflow descriptor is not set; \
                                 it should have been injected by the daemon at spawn time"
                        );
                    }
                }
            }
        }

        Ok((node, event_stream))
    }

    /// Check whether `output_id` is declared as an output of this node.
    ///
    /// Returns `true` if the output is declared (or this node is `interactive`,
    /// which has no static output declaration); `false` and emits a one-time
    /// warning if the output is unknown. Public so callers building higher-level
    /// send helpers (e.g. the Python `send_output_raw` zero-copy path) can
    /// validate before allocating a buffer.
    pub fn validate_output(&mut self, output_id: &DataId) -> bool {
        if !self.node_config.outputs.contains(output_id) && !self.interactive {
            if !self.warned_unknown_output.contains(output_id) {
                warn!("Ignoring output `{output_id}` not in node's output list.");
                self.warned_unknown_output.insert(output_id.clone());
            }
            false
        } else {
            true
        }
    }

    /// Send raw data from the node to the other nodes.
    ///
    /// We take a closure as an input to enable zero copy on send.
    ///
    /// ```no_run
    /// use dora_node_api::{DoraNode, MetadataParameters};
    /// use dora_core::config::DataId;
    ///
    /// let (mut node, mut events) = DoraNode::init_from_env().expect("Could not init node.");
    ///
    /// let output = DataId::from("output_id".to_owned());
    ///
    /// let data: &[u8] = &[0, 1, 2, 3];
    /// let parameters = MetadataParameters::default();
    ///
    /// node.send_output_raw(
    ///    output,
    ///    parameters,
    ///    data.len(),
    ///    |out| {
    ///         out.copy_from_slice(data);
    ///     }).expect("Could not send output");
    /// ```
    ///
    /// Ignores the output if the given `output_id` is not specified as node output in the dataflow
    /// configuration file.
    pub fn send_output_raw<F>(
        &mut self,
        output_id: DataId,
        parameters: MetadataParameters,
        data_len: usize,
        data: F,
    ) -> NodeResult<()>
    where
        F: FnOnce(&mut [u8]),
    {
        if !self.validate_output(&output_id) {
            return Ok(());
        };
        // The receiver expects a self-describing Arrow IPC stream. Build it in
        // place: pre-write the UInt8 IPC header into the (shared-memory) sample,
        // then let the caller write their bytes straight into the data region —
        // zero payload copies (and the SHM sample is moved into zenoh's `put`).
        // Prepare the UInt8 IPC header once, then size and fill the sample from
        // it — avoids rebuilding the layout + IPC headers for the length query.
        let prepared = ipc_encode::PreparedUint8Ipc::new(data_len)
            .map_err(|e| NodeError::Output(format!("Arrow IPC encode: {e}")))?;
        let mut sample = self.allocate_data_sample(prepared.byte_len())?;
        let offset = prepared
            .encode_header_into(&mut sample)
            .map_err(|e| NodeError::Output(format!("Arrow IPC encode: {e}")))?;
        data(&mut sample[offset..offset + data_len]);

        let mut parameters = parameters;
        parameters.insert(
            FRAMING.to_string(),
            Parameter::String(FRAMING_ARROW_IPC.to_string()),
        );
        self.send_output_sample(output_id, parameters, Some(sample))
    }

    /// Sends the given Arrow array as an output message.
    ///
    /// This is the recommended way to emit data from a node: pass any value that
    /// implements [`IntoArrow`] (primitives, `Vec<T>`, `&str`, an Arrow array,
    /// …) and dora moves it into shared memory for an efficient, near-zero-copy
    /// transfer to downstream nodes.
    ///
    /// Uses shared memory for efficient data transfer if suitable. This method
    /// might copy the message once to move it to shared memory.
    ///
    /// Ignores the output if the given `output_id` is not specified as node output in the dataflow
    /// configuration file.
    ///
    /// # Errors
    ///
    /// Returns [`NodeError::Output`] if the payload cannot be Arrow-IPC encoded, or if runtime type
    /// checking is enabled in error mode (`DORA_RUNTIME_TYPE_CHECK=error`) and the array's Arrow
    /// type does not match the output's declared type. An `output_id` that is not declared as an
    /// output is *not* an error — the call is ignored and returns `Ok`.
    ///
    /// ```no_run
    /// use dora_node_api::{DoraNode, MetadataParameters};
    /// use dora_core::config::DataId;
    ///
    /// let (mut node, _events) = DoraNode::init_from_env()?;
    ///
    /// let output = DataId::from("output_id".to_owned());
    /// let parameters = MetadataParameters::default();
    ///
    /// node.send_output(output, parameters, vec![1.0f32, 2.0, 3.0])?;
    /// # Ok::<(), eyre::Report>(())
    /// ```
    pub fn send_output(
        &mut self,
        output_id: DataId,
        parameters: MetadataParameters,
        data: impl IntoArrow,
    ) -> NodeResult<()> {
        if !self.validate_output(&output_id) {
            return Ok(());
        };

        let data = data.into_arrow();
        let arrow_array = dora_arrow_convert::internal::array_ref(&data).to_data();
        self.check_output_type(&output_id, arrow_array.data_type(), &parameters)?;

        let encoded = self.sample_allocator.encode_arrow_data(&arrow_array)?;
        self.send_encoded_unchecked(output_id, parameters, encoded.sample)
    }

    /// Send a payload that has already been IPC-encoded into a dora-owned
    /// [`EncodedSample`] by [`SampleAllocator::encode_arrow`].
    ///
    /// This is the entry point the runtime uses for operator outputs: the
    /// operator thread does the encoding so that no memory owned by the
    /// operator's language runtime is ever released on the node's thread — see
    /// [`SampleAllocator`] (dora-rs/dora#2742).
    ///
    /// Like [`send_output`](Self::send_output), an `output_id` that is not a
    /// declared output is ignored (returns `Ok`).
    ///
    /// # Errors
    ///
    /// Returns [`NodeError::Output`] if runtime type checking is enabled in error mode
    /// (`DORA_RUNTIME_TYPE_CHECK=error`) and the sample's Arrow type does not match the output's
    /// declared type.
    pub fn send_output_encoded(
        &mut self,
        output_id: DataId,
        parameters: MetadataParameters,
        encoded: EncodedSample,
    ) -> NodeResult<()> {
        if !self.validate_output(&output_id) {
            return Ok(());
        };
        self.check_output_type(&output_id, &encoded.data_type, &parameters)?;
        self.send_encoded_unchecked(output_id, parameters, encoded.sample)
    }

    /// Tag an already-encoded sample as an Arrow IPC stream and send it. The
    /// caller has already run `validate_output` and `check_output_type`.
    fn send_encoded_unchecked(
        &mut self,
        output_id: DataId,
        mut parameters: MetadataParameters,
        sample: DataSample,
    ) -> NodeResult<()> {
        parameters.insert(
            FRAMING.to_string(),
            Parameter::String(FRAMING_ARROW_IPC.to_string()),
        );

        self.send_output_sample(output_id, parameters, Some(sample))
            .wrap_err("failed to send output")?;

        Ok(())
    }

    /// Runtime type check (only when `DORA_RUNTIME_TYPE_CHECK` is set).
    ///
    /// Skips the check when this message carries pattern metadata
    /// (`request_id`, `goal_id`, or `goal_status`). Service, action, and
    /// streaming patterns legitimately multiplex multiple Arrow schemas through
    /// a single output — a service server may reply with different response
    /// shapes for different request types — so a single declared Arrow type
    /// cannot cover all variants. Non-pattern messages still get full
    /// validation (dora-rs/adora#150).
    fn check_output_type(
        &self,
        output_id: &DataId,
        actual: &arrow_schema::DataType,
        parameters: &MetadataParameters,
    ) -> NodeResult<()> {
        if let Some((mode, checks)) = &self.runtime_type_checks
            && let Some(expected) = checks.get(output_id)
            && !carries_pattern_correlation(parameters)
            && actual != expected
        {
            let msg =
                format!("output \"{output_id}\": expected Arrow type {expected:?}, got {actual:?}");
            match mode {
                RuntimeTypeCheck::Error => {
                    return Err(NodeError::Output(msg));
                }
                RuntimeTypeCheck::Warn => {
                    warn!("type mismatch: {msg}");
                }
                RuntimeTypeCheck::Off => unreachable!(),
            }
        }
        Ok(())
    }

    /// Send the given raw byte data as output.
    ///
    /// Might copy the data once to move it into shared memory. `data_len` must equal `data.len()`;
    /// the allocated sample is sized from `data_len` and the payload is copied from `data`.
    ///
    /// Ignores the output if the given `output_id` is not specified as node output in the dataflow
    /// configuration file.
    ///
    /// # Errors
    ///
    /// Returns [`NodeError::Output`] if `data_len` does not equal `data.len()` (which would
    /// otherwise panic in the internal `copy_from_slice`).
    pub fn send_output_bytes(
        &mut self,
        output_id: DataId,
        parameters: MetadataParameters,
        data_len: usize,
        data: &[u8],
    ) -> NodeResult<()> {
        if !self.validate_output(&output_id) {
            return Ok(());
        };
        // `send_output_raw` allocates a `data_len`-byte sample and the closure
        // below copies `data` into it. A mismatch would otherwise panic deep
        // inside `copy_from_slice` ("source slice length .. does not match
        // destination slice length .."); return a clear error instead.
        if data.len() != data_len {
            return Err(NodeError::Output(format!(
                "send_output_bytes: data_len ({data_len}) does not match data.len() ({})",
                data.len()
            )));
        }
        self.send_output_raw(output_id, parameters, data_len, |sample| {
            sample.copy_from_slice(data)
        })
    }

    /// Sends the given [`DataSample`] as output.
    ///
    /// The sample must already be a self-describing Arrow IPC stream (the
    /// `FRAMING_ARROW_IPC` parameter should be set). It is recommended to use a
    /// function like [`send_output`][Self::send_output] instead, which handles
    /// the encoding.
    ///
    /// Ignores the output if the given `output_id` is not specified as node output in the dataflow
    /// configuration file.
    pub fn send_output_sample(
        &mut self,
        output_id: DataId,
        mut parameters: MetadataParameters,
        sample: Option<DataSample>,
    ) -> NodeResult<()> {
        // `SCHEMA_HASH` is an internal wire-protocol key that only
        // `publish_schema_once` may set, and only for the schema-less batch it
        // belongs to. A stale value forwarded from an input's metadata (the
        // receive path strips it, but a recorded/hand-built parameter map can
        // still carry one) would make receivers route this output's full
        // self-describing stream to the schema-once decoder, hash-mismatch, and
        // silently drop it (dora-rs/dora#2366 review).
        parameters.remove(SCHEMA_HASH);
        // Auto-inject OpenTelemetry trace context when telemetry is enabled.
        // Uses the ambient OTel context, which is populated when the tracing
        // subscriber has an OpenTelemetry layer (e.g., via with_otlp_tracing).
        // Only trace/span IDs are propagated (via W3C TraceContext propagator).
        // OTel Baggage is NOT propagated to avoid leaking sensitive data across
        // node boundaries. If a user explicitly provides this key, it wins.
        #[cfg(feature = "tracing")]
        if !parameters.contains_key(crate::OPEN_TELEMETRY_CONTEXT) {
            let cx = opentelemetry::Context::current();
            let serialized = dora_tracing::telemetry::serialize_context(&cx);
            if !serialized.is_empty() {
                parameters.insert(
                    crate::OPEN_TELEMETRY_CONTEXT.to_string(),
                    crate::Parameter::String(serialized),
                );
            }
        }

        let metadata = Metadata::from_parameters(self.clock.new_timestamp(), parameters);

        let finalized = sample.map(|sample| sample.finalize());

        // Diagnostic (dora-rs/dora#2742): the Windows nightly wedges a runtime's
        // main loop inside this function on the first large output and is
        // force-killed at the daemon's grace period, so nothing that only logs
        // *after* a hop returns can ever show where it parked. Log on entry to
        // each hop instead: the last line printed names the blocked call. Capped
        // at the first few large sends (the wedge is on the first), so a healthy
        // run pays one comparison per send and prints a handful of lines.
        // `warn!` so it survives the default stdout filter and reaches CI logs.
        let diag_bytes = finalized.as_ref().map_or(0, |f| f.byte_len());
        let diag = diag_bytes >= self.sample_allocator.zero_copy_threshold
            && self.large_send_diag_count < LARGE_SEND_DIAG_LIMIT;
        if diag {
            self.large_send_diag_count += 1;
        }

        // How a data-plane message should be delivered.
        enum Delivery {
            /// zenoh delivered the payload (or it was consumed by a failed SHM
            /// put); only the daemon's control-plane state needs syncing.
            Zenoh,
            /// Deliver via the daemon control channel. `None` is a metadata-only
            /// message with no payload.
            Daemon(Option<DataMessage>),
        }

        // Publish via direct zenoh only when the output may take the direct
        // path: its publisher exists and the startup handshake has proven its
        // routes — every required consumer acked a marker (see
        // `StartupHandshake`). Everything else takes the reliable daemon path:
        // no zenoh session (interactive/testing mode), an output the daemon
        // pinned there (a consumer on another daemon needs inter-daemon
        // forwarding, which only daemon-path sends feed — #2738), or an output
        // whose handshake did not complete before `init` returned and is
        // therefore frozen there for the run. An SHM-backed sample is moved
        // straight into zenoh's `put` (no extra copy); only the daemon path
        // copies it out into a `DataMessage::Vec`.
        let delivery = match finalized {
            Some(finalized) if self.output_direct_ready(&output_id) => {
                tracing::trace!(
                    output = %output_id,
                    size = finalized.byte_len(),
                    "publishing via zenoh"
                );
                if diag {
                    warn!(
                        "output `{output_id}`: {diag_bytes} B -> zenoh direct path \
                         (dora-rs/dora#2742 diagnostic)"
                    );
                }
                match self.zenoh_publish(&output_id, &metadata, finalized, diag) {
                    Ok(PublishOutcome::Published) => Delivery::Zenoh,
                    Ok(PublishOutcome::NotPublished(sample)) => {
                        Delivery::Daemon(Some(sample.into_data_message()))
                    }
                    Err(e) => {
                        tracing::warn!(
                            "zenoh publish failed ({e}); message dropped \
                             (SHM payload consumed, no daemon fallback)"
                        );
                        Delivery::Zenoh
                    }
                }
            }
            Some(finalized) => {
                if diag {
                    warn!(
                        "output `{output_id}`: {diag_bytes} B -> daemon path \
                         (dora-rs/dora#2742 diagnostic)"
                    );
                }
                Delivery::Daemon(Some(finalized.into_data_message()))
            }
            None => Delivery::Daemon(None),
        };

        match delivery {
            Delivery::Zenoh => {
                // Keep the daemon's control-plane state in sync (input
                // deadlines, circuit-breaker recovery) without duplicating the
                // data payload that zenoh already delivered.
                if diag {
                    warn!(
                        "output `{output_id}`: entering report_output_sent \
                         (dora-rs/dora#2742 diagnostic)"
                    );
                }
                self.control_channel
                    .report_output_sent(output_id.clone(), metadata)
                    .wrap_err_with(|| format!("failed to report output {output_id}"))?;
            }
            Delivery::Daemon(data) => {
                // The daemon/TCP path serializes the whole message; an oversized
                // IPC payload would otherwise fail deep in the transport with a
                // generic error. Reject it here with a clear, output-specific
                // message. Large payloads are expected to reach a zenoh
                // subscriber instead, which has no such limit.
                if let Some(DataMessage::Vec(v)) = &data
                    && v.len() > dora_message::MAX_MESSAGE_BYTES
                {
                    return Err(NodeError::Output(format!(
                        "output \"{output_id}\": IPC-encoded message is {} bytes, exceeding \
                         the {}-byte daemon transport limit (the output is on the daemon \
                         path: pinned for a consumer only forwarding can reach, its \
                         startup handshake did not complete, or no zenoh route is \
                         available)",
                        v.len(),
                        dora_message::MAX_MESSAGE_BYTES,
                    )));
                }
                if diag {
                    warn!(
                        "output `{output_id}`: entering control_channel.send_message \
                         (dora-rs/dora#2742 diagnostic)"
                    );
                }
                self.control_channel
                    .send_message(output_id.clone(), metadata, data)
                    .wrap_err_with(|| format!("failed to send output {output_id}"))?;
            }
        }

        Ok(())
    }

    /// Report the given outputs IDs as closed.
    ///
    /// The node is not allowed to send more outputs with the closed IDs.
    ///
    /// Closing outputs early can be helpful to receivers.
    ///
    /// # Errors
    ///
    /// Returns [`NodeError::Output`] if any id is not a declared output of this node. Unlike
    /// [`send_output`](Self::send_output), which silently ignores unknown outputs, this validates
    /// the whole batch *before* closing any output, so on error none of them are closed.
    pub fn close_outputs(&mut self, outputs_ids: Vec<DataId>) -> NodeResult<()> {
        // Validate the whole batch before mutating any local state. Removing
        // outputs eagerly would leave the node's local output set out of sync
        // with the daemon if a later id is unknown: the early ones would be
        // gone locally, yet `report_closed_outputs` is skipped on error so the
        // daemon never learns about them.
        for output_id in &outputs_ids {
            if !self.node_config.outputs.contains(output_id) {
                return Err(NodeError::Output(format!("unknown output {output_id}")));
            }
        }
        for output_id in &outputs_ids {
            self.node_config.outputs.remove(output_id);
        }

        self.control_channel
            .report_closed_outputs(outputs_ids)
            .wrap_err("failed to report closed outputs to daemon")?;

        Ok(())
    }

    /// Whether `output_id` may take the direct zenoh path: its publisher exists
    /// (declared at init, not pinned to the daemon path) and the startup
    /// handshake proved its routes — see [`StartupHandshake`].
    ///
    /// Constant for the node's lifetime — see [`wait_for_grace`].
    fn output_direct_ready(&self, output_id: &DataId) -> bool {
        self.zenoh_publishers
            .get(output_id)
            .is_some_and(|output| output.ready.load(Ordering::Relaxed))
    }

    /// Publish data directly via zenoh (node-to-node, bypassing daemon for data).
    /// Uses SHM for zero-copy when possible, falls back to heap buffer.
    ///
    /// The publisher was declared at init (see [`declare_output_publishers`]) with
    /// `express(true)` to bypass zenoh's adaptive batch timer and with
    /// `Priority::RealTime` so data-plane messages don't share queues with bulk
    /// traffic. Its routes were already proven by the startup handshake
    /// ([`StartupHandshake`]) before [`Self::output_direct_ready`] let the send
    /// take this path, so the first send here cannot be dropped for a
    /// not-yet-established subscription.
    ///
    /// `diag` enables the per-hop entry logging described in
    /// [`Self::send_output_sample`] (dora-rs/dora#2742); it is only ever set for
    /// the first few large sends of a node's lifetime.
    fn zenoh_publish(
        &mut self,
        output_id: &DataId,
        metadata: &Metadata,
        finalized: FinalizedSample,
        diag: bool,
    ) -> eyre::Result<PublishOutcome> {
        use zenoh::Wait;

        // Every failure *before* the payload is moved into `put` returns the
        // sample as `NotPublished` so the caller can still deliver it via the
        // daemon. Only a failed `put` of an SHM buffer (which consumes it)
        // returns `Err` — that is the single non-recoverable case.
        //
        // No publisher means there is no zenoh session, its declaration failed
        // at init, or the output is pinned to the daemon path: fall back to the
        // reliable daemon path (defense in depth — `output_direct_ready`
        // already gates the caller).
        let Some(DirectOutput { publisher, .. }) = self.zenoh_publishers.get(output_id) else {
            return Ok(PublishOutcome::NotPublished(finalized));
        };
        let session = self
            .zenoh_session
            .as_ref()
            .expect("a declared publisher implies a zenoh session");

        // Serialize metadata as zenoh attachment.
        let metadata_bytes = match dora_message::encode(metadata) {
            Ok(bytes) => bytes,
            Err(e) => {
                tracing::warn!(output = %output_id, "failed to serialize metadata ({e}); falling back to daemon path");
                return Ok(PublishOutcome::NotPublished(finalized));
            }
        };

        match finalized {
            // The producer already wrote into shared memory. Move the SHM
            // buffer straight into `put` — no realloc, no copy. This is the
            // path that eliminates the former heap-to-SHM second copy.
            //
            // On a put error the buffer has been consumed and cannot be
            // recovered for the daemon fallback, so this returns an error
            // (the caller logs and drops the message). This is a deliberate
            // trade-off for zero-copy on the common matched-subscriber path.
            // Producer-constructed SHM sample: `put` *moves* (consumes) the SHM
            // buffer, so — unlike the borrowed-heap `Vec` arm below — there is no
            // intact payload left to retry on error. A put failure is therefore
            // best-effort: the message is dropped (the caller logs it). This is
            // the deliberate, accepted trade-off for the zero-copy large-output
            // path, not an oversight.
            FinalizedSample::Shm(sbuf) => {
                if diag {
                    tracing::warn!(
                        "output `{output_id}`: entering zenoh put of an SHM buffer \
                         (dora-rs/dora#2742 diagnostic)"
                    );
                }
                publisher
                    .put(sbuf)
                    .attachment(&metadata_bytes[..])
                    .wait()
                    .map_err(|e| eyre::eyre!("zenoh SHM publish failed: {e}"))?;
                Ok(PublishOutcome::Published)
            }
            // Heap payload. At or above the threshold, copy once into a fresh
            // SHM buffer so local subscribers still get zero-copy delivery;
            // below it, a heap-buffered put is cheaper than a full SHM page.
            // The heap buffer is only borrowed, so any put error can fall back
            // to the daemon path with the payload intact.
            FinalizedSample::Vec(avec) => {
                if avec.len() >= self.sample_allocator.zero_copy_threshold
                    && let Some(provider) = &self.sample_allocator.shm_provider
                {
                    use zenoh::shm::GarbageCollect;
                    // Non-blocking: garbage-collect freed chunks and allocate, but
                    // do NOT block waiting for the pool to drain. Under a burst of
                    // large messages the zero-copy receiver pins each segment for
                    // the whole receive pipeline, so the pool can be momentarily
                    // exhausted; `BlockOn` would then sleep 1 ms per retry (zenoh
                    // 1.8 has no alloc signalling yet), throttling throughput to
                    // ~1k msg/s. Falling back to a heap-buffered put instead keeps
                    // the producer moving (PR #2366).
                    if diag {
                        tracing::warn!(
                            "output `{output_id}`: entering SHM alloc of {} B \
                             (dora-rs/dora#2742 diagnostic)",
                            avec.len()
                        );
                    }
                    match provider
                        .alloc(avec.len())
                        .with_policy::<GarbageCollect>()
                        .wait()
                    {
                        Ok(mut sbuf) => {
                            // Mirror the guard in `allocate_data_sample`: only
                            // copy into the SHM buffer when it is exactly the
                            // requested size. zenoh 1.8 guarantees the logical
                            // length matches the request, but `copy_from_slice`
                            // requires equal lengths and would panic on the
                            // node's send thread if a future provider ever
                            // over-allocated. Fall through to the reliable
                            // daemon path instead of risking that panic.
                            if sbuf.as_mut().len() == avec.len() {
                                sbuf.as_mut().copy_from_slice(&avec);
                                if diag {
                                    tracing::warn!(
                                        "output `{output_id}`: entering zenoh put of a \
                                         copied SHM buffer (dora-rs/dora#2742 diagnostic)"
                                    );
                                }
                                return match publisher
                                    .put(sbuf)
                                    .attachment(&metadata_bytes[..])
                                    .wait()
                                {
                                    Ok(()) => Ok(PublishOutcome::Published),
                                    Err(e) => {
                                        tracing::warn!(
                                            "zenoh SHM publish failed ({e}); \
                                             falling back to daemon path"
                                        );
                                        Ok(PublishOutcome::NotPublished(FinalizedSample::Vec(avec)))
                                    }
                                };
                            }
                            tracing::debug!(
                                "zenoh SHM alloc returned {} bytes for a {}-byte \
                                 request; using daemon path",
                                sbuf.as_ref().len(),
                                avec.len()
                            );
                        }
                        Err(e) => {
                            tracing::debug!("SHM alloc failed ({e}), using heap buffer");
                        }
                    }
                }

                // A large payload that did not make it into SHM (no provider, or
                // the pool was momentarily full) must NOT be published over the
                // zenoh data plane: a payload larger than the transport batch
                // size is fragmented, and the express/`Drop` data publisher
                // silently drops fragmented messages — `put` reports success but
                // the subscriber never receives them (PR #2366). Route it via the
                // reliable daemon path instead (TCP, up to `MAX_MESSAGE_BYTES`).
                // Only sub-threshold payloads, which fit a single batch and never
                // fragment, take the zenoh heap put below.
                if avec.len() >= self.sample_allocator.zero_copy_threshold {
                    return Ok(PublishOutcome::NotPublished(FinalizedSample::Vec(avec)));
                }

                // Only sub-threshold (single-batch, never-fragmented) payloads
                // reach this point — large payloads were routed to the daemon
                // path above. Apply the schema-once optimization to small
                // messages with a stable Arrow schema: publish the schema on the
                // `@schema` subtopic (only on change) and send just the
                // schema-less batch on the data topic, tagged with the schema
                // hash so the receiver matches it to the decoder primed from the
                // subtopic.
                //
                // The message that (re)publishes the schema — the output's first,
                // every schema change, any message after a failed `@schema` put,
                // and a periodic refresh — is itself sent as a full
                // self-describing stream (`publish_schema_once` returns `None`
                // for it). It decodes standalone and primes receivers in-band,
                // in data-plane order, so the express batch can never outrun its
                // own schema (the `@schema` plane's non-express `Block` publish
                // otherwise loses that race) and a one-shot output cannot lose
                // its only message.
                //
                // Service/action request-reply messages (carrying
                // `request_id`/`goal_id`/`goal_status`) are excluded: a server
                // legitimately multiplexes multiple response schemas through one
                // output, interleaved per request, and each per-message schema
                // change would force a full stream + `@schema` publish anyway.
                // Sending them as full self-describing streams (the pre-PR
                // behavior) makes each message decode standalone regardless of
                // schema order, at the cost of ~400 B of framing per message —
                // acceptable for these request/reply-rate patterns.
                //
                // Streaming (`session_id`/`segment_id`) is deliberately NOT
                // excluded: every chunk of a stream shares one schema, so
                // schema-once primes once and each chunk reuses it — streaming is
                // the high-rate small-message case schema-once exists for. A
                // schema change at a segment boundary is just the one-time
                // re-prime window any schema-once output has, not the per-message
                // alternation that makes service/action lossy.
                //
                // `schema_once` is bound here, not inside the match, so its
                // attachment bytes outlive the `put` below.
                let schema_once = if schema_once_eligible(
                    avec.len(),
                    self.sample_allocator.zero_copy_threshold,
                    &metadata.parameters,
                ) {
                    publish_schema_once(
                        &mut self.zenoh_schema_publishers,
                        &mut self.zenoh_schema_state,
                        session,
                        self.dataflow_id,
                        &self.id,
                        output_id,
                        &avec,
                        metadata,
                    )
                } else {
                    None
                };
                // Fall back to a full standalone stream if the batch slice can't
                // be taken (a real IPC stream always can — defensive).
                let (payload, attachment): (&[u8], &[u8]) = match schema_once.as_ref() {
                    Some(att) => match arrow_utils::ipc_encode::batch_slice(&avec) {
                        Some(slice) => (slice, att.as_slice()),
                        None => (&avec[..], &metadata_bytes[..]),
                    },
                    None => (&avec[..], &metadata_bytes[..]),
                };
                match publisher.put(payload).attachment(attachment).wait() {
                    Ok(()) => Ok(PublishOutcome::Published),
                    Err(e) => {
                        tracing::warn!("zenoh publish failed ({e}); falling back to daemon path");
                        // The zenoh data plane did not deliver this message. If
                        // it was the one meant to prime receivers in-band (the
                        // first message of a schema, or a periodic refresh),
                        // `publish_schema_once` already recorded its state and
                        // the following messages would go out schema-less with
                        // no delivered priming stream. Forget the output's
                        // schema-once state so the next message sends a full
                        // stream and re-publishes the schema. (A congestion
                        // drop reports `Ok` and stays undetectable — inherent
                        // to `CongestionControl::Drop`; the periodic refresh
                        // bounds that residual window.)
                        self.zenoh_schema_state.remove(output_id);
                        Ok(PublishOutcome::NotPublished(FinalizedSample::Vec(avec)))
                    }
                }
            }
        }
    }

    /// Returns the ID of the node as specified in the dataflow configuration file.
    pub fn id(&self) -> &NodeId {
        &self.id
    }

    /// Returns the unique identifier for the running dataflow instance.
    ///
    /// Dora assigns each dataflow instance a random identifier when started.
    pub fn dataflow_id(&self) -> &DataflowId {
        &self.dataflow_id
    }

    /// Returns the input and output configuration of this node.
    pub fn node_config(&self) -> &NodeRunConfig {
        &self.node_config
    }

    /// Returns the zero-copy SHM threshold in bytes.
    ///
    /// Outputs whose raw payload is at least this many bytes are published via
    /// zenoh shared memory (zero-copy for local subscribers); smaller outputs
    /// are published via zenoh with a heap-buffered put. Configured via the
    /// `DORA_ZERO_COPY_THRESHOLD` env var, defaulting to
    /// [`ZERO_COPY_THRESHOLD`].
    pub fn zero_copy_threshold(&self) -> usize {
        self.sample_allocator.zero_copy_threshold
    }

    /// Returns true if this node was restarted after a previous exit or failure.
    ///
    /// Nodes can use this to decide whether to restore saved state or start fresh.
    pub fn is_restart(&self) -> bool {
        self.restart_count > 0
    }

    /// Returns how many times this node has been restarted.
    ///
    /// Returns 0 on the first run, 1 after the first restart, etc.
    pub fn restart_count(&self) -> u32 {
        self.restart_count
    }

    /// Returns the current timestamp from the node's Hybrid Logical Clock.
    ///
    /// This generates a new HLC timestamp, which combines the physical
    /// wall-clock time with a logical counter to ensure uniqueness and
    /// monotonicity even across nodes. The HLC is the same clock dora
    /// stamps every outgoing message with, so this is the right value
    /// to subtract from an input event's `metadata.timestamp` when
    /// measuring per-event processing latency — using
    /// `std::time::SystemTime::now()` instead would mix two unrelated
    /// clocks and give meaningless results across daemons.
    pub fn timestamp(&self) -> uhlc::Timestamp {
        self.clock.new_timestamp()
    }

    /// Send a structured log message.
    ///
    /// Outputs a JSONL line to stdout that the daemon parses automatically.
    /// Works with `min_log_level` filtering and `send_logs_as` routing.
    ///
    /// `level` should be one of: `"error"`, `"warn"`, `"info"`, `"debug"`, `"trace"`.
    /// Unknown levels default to `"info"`.
    pub fn log(&self, level: &str, message: &str, target: Option<&str>) {
        self.log_with_fields(level, message, target, None);
    }

    /// Maximum serialized size of the log `fields` object before it is
    /// dropped (60 KB). Matches the downstream 64 KB parse limit with headroom
    /// for the message envelope. Measured on the serialized JSON (see
    /// [`log_fields_within_budget`]), not the raw key/value byte sum.
    const MAX_LOG_FIELDS_BYTES: usize = 60 * 1024;

    /// Send a structured log message with optional key-value fields.
    ///
    /// Like [`log`](Self::log), but accepts additional structured fields that
    /// are included in the JSON payload and preserved through `send_logs_as`.
    pub fn log_with_fields(
        &self,
        level: &str,
        message: &str,
        target: Option<&str>,
        fields: Option<&std::collections::BTreeMap<String, String>>,
    ) {
        let level_str = match level.to_lowercase().as_str() {
            "error" => "error",
            "warn" | "warning" => "warn",
            "info" => "info",
            "debug" => "debug",
            "trace" => "trace",
            _ => "info",
        };
        let timestamp = chrono::Utc::now().to_rfc3339();
        let mut entry = serde_json::json!({
            "timestamp": timestamp,
            "level": level_str,
            "node_id": self.id.to_string(),
            "message": message,
        });
        if let Some(target) = target {
            entry["target"] = serde_json::Value::String(target.to_string());
        }
        if let Some(fields) = fields {
            match log_fields_within_budget(fields, Self::MAX_LOG_FIELDS_BYTES) {
                Some(value) => entry["fields"] = value,
                None => {
                    eprintln!("dora log: fields too large, dropping fields");
                    entry["fields_dropped"] = serde_json::Value::Bool(true);
                }
            }
        }
        match serde_json::to_string(&entry) {
            Ok(json) => println!("{json}"),
            Err(e) => eprintln!("dora log serialization error: {e}"),
        }
    }

    /// Log an error message.
    pub fn log_error(&self, message: &str) {
        self.log("error", message, None);
    }

    /// Log a warning message.
    pub fn log_warn(&self, message: &str) {
        self.log("warn", message, None);
    }

    /// Log an info message.
    pub fn log_info(&self, message: &str) {
        self.log("info", message, None);
    }

    /// Log a debug message.
    pub fn log_debug(&self, message: &str) {
        self.log("debug", message, None);
    }

    /// Log a trace message.
    pub fn log_trace(&self, message: &str) {
        self.log("trace", message, None);
    }

    // -----------------------------------------------------------------
    // Service / Action helpers
    // -----------------------------------------------------------------

    /// Generate a new unique request/goal ID (UUID v7, time-ordered).
    ///
    /// Uses a per-thread monotonic counter context to guarantee uniqueness
    /// even when multiple IDs are generated within the same clock tick.
    pub fn new_request_id() -> String {
        thread_local! {
            static CTX: uuid::ContextV7 = const { uuid::ContextV7::new() };
        }
        CTX.with(|ctx| uuid::Uuid::new_v7(uuid::Timestamp::now(ctx)).to_string())
    }

    /// Generate a new unique goal ID (UUID v7, time-ordered).
    ///
    /// This is an alias for [`new_request_id`](Self::new_request_id) that
    /// reads more naturally in action (goal/feedback/result) contexts.
    pub fn new_goal_id() -> String {
        Self::new_request_id()
    }

    /// Send a service request, automatically injecting a `request_id` into the
    /// metadata parameters. Returns the generated request ID.
    ///
    /// Any existing `request_id` key in `parameters` is replaced.
    ///
    /// # Errors
    ///
    /// Propagates any error from [`send_output`](Self::send_output).
    pub fn send_service_request(
        &mut self,
        output_id: DataId,
        mut parameters: MetadataParameters,
        data: impl IntoArrow,
    ) -> NodeResult<String> {
        if parameters.contains_key(dora_message::metadata::REQUEST_ID) {
            tracing::warn!("send_service_request: caller-provided request_id will be overwritten");
        }
        let request_id = Self::new_request_id();
        parameters.insert(
            dora_message::metadata::REQUEST_ID.to_string(),
            dora_message::metadata::Parameter::String(request_id.clone()),
        );
        self.send_output(output_id, parameters, data)?;
        Ok(request_id)
    }

    /// Send a service response. This is a semantic alias for [`send_output`](Self::send_output).
    ///
    /// The caller is expected to pass through the `request_id` parameter from
    /// the incoming request's metadata.
    pub fn send_service_response(
        &mut self,
        output_id: DataId,
        parameters: MetadataParameters,
        data: impl IntoArrow,
    ) -> NodeResult<()> {
        self.send_output(output_id, parameters, data)
    }

    // -----------------------------------------------------------------
    // Streaming helpers
    // -----------------------------------------------------------------

    /// Send a streaming segment chunk. Convenience wrapper around
    /// [`send_output`](Self::send_output) that builds metadata from the
    /// [`StreamSegment`] builder.
    ///
    /// # Errors
    ///
    /// Propagates any error from [`send_output`](Self::send_output).
    pub fn send_stream_chunk(
        &mut self,
        output_id: DataId,
        segment: &mut StreamSegment,
        fin: bool,
        data: impl IntoArrow,
    ) -> NodeResult<()> {
        self.send_output(output_id, segment.chunk(fin), data)
    }

    /// Allocates a [`DataSample`] of the specified size.
    ///
    /// See [`SampleAllocator::allocate`] for the allocation strategy.
    pub fn allocate_data_sample(&mut self, data_len: usize) -> NodeResult<DataSample> {
        self.sample_allocator.allocate(data_len)
    }

    /// A handle for building output samples off this node's thread.
    ///
    /// The runtime hands one to each operator thread so the operator can encode
    /// its payload into a dora-owned [`DataSample`] itself — see
    /// [`SampleAllocator`] for why that matters (dora-rs/dora#2742).
    pub fn sample_allocator(&self) -> SampleAllocator {
        self.sample_allocator.clone()
    }

    /// Returns the full dataflow descriptor that this node is part of.
    ///
    /// This method returns the parsed dataflow YAML file.
    pub fn dataflow_descriptor(&self) -> NodeResult<&Descriptor> {
        match &self.dataflow_descriptor {
            Ok(d) => Ok(d),
            Err(err) => Err(NodeError::Data(format!(
                "failed to parse dataflow descriptor: {err}\n\n\
                    This might be caused by mismatched version numbers of dora \
                    daemon and the dora node API"
            ))),
        }
    }

    /// Store an opaque value in the daemon's dataflow-scoped extension table.
    ///
    /// This is the seam for transports that live outside the dora tree: dora
    /// brokers the value's lifetime and nothing else — it never interprets
    /// `namespace`, `key` or `value`. See `docs/extensions.md`.
    ///
    /// The daemon remembers which nodes touched a key so that dropping it
    /// notifies them, and reclaims the entry when the dataflow ends or the
    /// storing node exits. Drain the notifications with
    /// [`event_stream::extensions::drain_dropped_keys`](crate::event_stream::extensions::drain_dropped_keys).
    pub fn extension_store(
        &mut self,
        namespace: impl Into<String>,
        key: impl Into<String>,
        value: Vec<u8>,
    ) -> Result<(), eyre::Error> {
        self.control_channel
            .extension_store(namespace.into(), key.into(), value)
    }

    /// Read an opaque value back, optionally removing it in the same round trip.
    ///
    /// Returns `None` if the key is not in the table — never stored, or
    /// already dropped.
    pub fn extension_load(
        &mut self,
        namespace: impl Into<String>,
        key: impl Into<String>,
        remove: bool,
    ) -> Result<Option<Vec<u8>>, eyre::Error> {
        self.control_channel
            .extension_load(namespace.into(), key.into(), remove)
    }

    /// Drop an opaque value, notifying every node that stored or loaded it.
    pub fn extension_drop(
        &mut self,
        namespace: impl Into<String>,
        key: impl Into<String>,
    ) -> Result<(), eyre::Error> {
        self.control_channel
            .extension_drop(namespace.into(), key.into())
    }

    /// Send an opaque request to the extension registered under
    /// `namespace` on this node's daemon, and return its opaque reply.
    ///
    /// Companion to [`DoraNode::extension_store`] / [`DoraNode::extension_load`]:
    /// those broker a descriptor's lifetime, this one carries a call the
    /// extension's daemon half must service. dora interprets neither the
    /// namespace nor the bytes — see `docs/extensions.md`.
    pub fn extension_request(
        &mut self,
        namespace: impl Into<String>,
        payload: Vec<u8>,
    ) -> Result<Vec<u8>, eyre::Error> {
        self.control_channel
            .extension_request(namespace.into(), payload)
    }
}

/// Return the serialized log `fields` object when it fits `limit`, else `None`.
///
/// The budget guards a downstream JSON-line parse limit, so it must measure
/// the *serialized* size: `"fields":{...}` adds structural bytes (quotes,
/// colons, commas) and JSON escaping — a value full of `"`/`\` doubles and
/// control characters expand ~6x via `\uXXXX`. Summing raw key/value byte
/// lengths can pass a map whose serialized form is well over the limit, which
/// the downstream parser then drops or truncates whole.
fn log_fields_within_budget(
    fields: &std::collections::BTreeMap<String, String>,
    limit: usize,
) -> Option<serde_json::Value> {
    // Count the serialized bytes without allocating a throwaway string, then
    // build the JSON value only when it fits.
    struct ByteCounter(usize);
    impl std::io::Write for ByteCounter {
        fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
            self.0 += buf.len();
            Ok(buf.len())
        }
        fn flush(&mut self) -> std::io::Result<()> {
            Ok(())
        }
    }
    let mut counter = ByteCounter(0);
    serde_json::to_writer(&mut counter, fields).ok()?;
    (counter.0 <= limit).then(|| serde_json::json!(fields))
}

/// Builder for initializing a node with custom connection parameters.
///
/// Created via [`DoraNode::builder()`]. Callers who don't need a custom daemon
/// port should prefer [`DoraNode::init_from_env`] or
/// [`DoraNode::init_from_node_id`]. Setting [`node_id`](Self::node_id) selects
/// the dynamic-node path; otherwise [`build`](Self::build) falls back to
/// [`DoraNode::init_from_env`].
#[derive(Default)]
pub struct DoraNodeBuilder {
    node_id: Option<NodeId>,
    daemon_port: Option<u16>,
}

impl DoraNodeBuilder {
    /// Set the node ID. Presence of a node ID selects the dynamic-node path.
    pub fn node_id(mut self, node_id: NodeId) -> Self {
        self.node_id = Some(node_id);
        self
    }

    /// No-op kept for source compatibility with upstream dora 0.5.x
    /// [`#1591`](https://github.com/dora-rs/dora/pull/1591). Upstream gates the
    /// dynamic-node path on an explicit `.dynamic()` call; here, dynamic mode
    /// is selected by the presence of `node_id`, making the flag redundant.
    /// Kept so that `.node_id(id).dynamic().build()` written against upstream
    /// still compiles.
    #[inline]
    pub fn dynamic(self) -> Self {
        self
    }

    /// Override the daemon port. When unset, the builder honours the
    /// `DORA_DAEMON_LOCAL_LISTEN_PORT` env var and falls back to
    /// `DORA_DAEMON_LOCAL_LISTEN_PORT_DEFAULT`.
    pub fn daemon_port(mut self, port: u16) -> Self {
        self.daemon_port = Some(port);
        self
    }

    /// Build and connect the node.
    pub fn build(self) -> NodeResult<(DoraNode, EventStream)> {
        let Some(node_id) = self.node_id else {
            return DoraNode::init_from_env();
        };

        let port = self.daemon_port.unwrap_or_else(|| {
            match std::env::var(DORA_DAEMON_LOCAL_LISTEN_PORT_ENV) {
                Ok(p) => p.parse().unwrap_or_else(|e| {
                    tracing::warn!(
                        "invalid {DORA_DAEMON_LOCAL_LISTEN_PORT_ENV}={p:?}: {e}, using default port"
                    );
                    DORA_DAEMON_LOCAL_LISTEN_PORT_DEFAULT
                }),
                Err(_) => DORA_DAEMON_LOCAL_LISTEN_PORT_DEFAULT,
            }
        });
        let daemon_address = (LOCALHOST, port).into();

        let mut channel =
            DaemonChannel::new_tcp(daemon_address).context("Could not connect to the daemon")?;
        let clock = Arc::new(uhlc::HLC::default());

        let reply = channel
            .request(&Timestamped {
                inner: DaemonRequest::NodeConfig { node_id },
                timestamp: clock.new_timestamp(),
            })
            .wrap_err("failed to request node config from daemon")?;

        match reply {
            DaemonReply::NodeConfig {
                result: Ok(node_config),
            } => DoraNode::init(node_config),
            DaemonReply::NodeConfig { result: Err(error) } => {
                let capped: String = error.chars().take(512).collect();
                Err(NodeError::Init(format!(
                    "failed to get node config from daemon: {capped}"
                )))
            }
            _ => Err(NodeError::Init("unexpected reply from daemon".into())),
        }
    }
}

/// Runs `teardown` on a dedicated thread, waiting at most `timeout` for it to
/// complete. Returns `true` if the teardown finished in time. Panics in
/// `teardown` are contained and count as completion. If spawning the thread
/// fails, the teardown runs inline without a deadline.
///
/// On timeout the thread keeps running detached: everything moved into the
/// closure (zenoh sockets, SHM segments, the owned tokio runtime) is leaked
/// until process exit. That is acceptable for nodes dropped right before
/// exit; long-lived hosts (e.g. a Python interpreter dropping a node during
/// GC) inherit only the bounded delay instead of a permanent hang.
pub(crate) fn teardown_with_timeout(
    label: &str,
    timeout: Duration,
    teardown: impl FnOnce() + Send + 'static,
) -> bool {
    // The closure is handed over via a channel (instead of being captured by
    // the spawned closure) so that it stays available for the inline
    // fallback when spawning fails.
    let (work_tx, work_rx) = std::sync::mpsc::channel();
    let (done_tx, done_rx) = std::sync::mpsc::channel();
    let thread = std::thread::Builder::new()
        .name(format!("dora-teardown-{label}"))
        .spawn(move || {
            if let Ok(work) = work_rx.recv() {
                let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(work));
            }
            let _ = done_tx.send(());
        });
    match thread {
        Ok(_) => {
            let _ = work_tx.send(teardown);
            done_rx.recv_timeout(timeout).is_ok()
        }
        Err(err) => {
            warn!("failed to spawn {label} teardown thread ({err}); running it inline");
            let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(teardown));
            true
        }
    }
}

impl Drop for DoraNode {
    fn drop(&mut self) {
        // The startup handshake's marker thread holds an `Arc` clone of the
        // publishers, so it must be stopped and joined before the publishers
        // are dropped for the undeclare below to see the last reference. Do
        // that join *inside* the bounded teardown: `shutdown()` sets the stop
        // flag but cannot interrupt an in-progress `publisher.put().wait()`, so
        // on a wedged zenoh net runtime the join could otherwise hang node
        // shutdown (and with it the daemon, which waits for `InputClosed`) —
        // the exact hang `teardown_with_timeout` exists to bound (#2425). The
        // consumer side already joins its acker thread under the same deadline.
        let startup_handshake = self.startup_handshake.take();
        // Tear down zenoh before notifying the daemon below, so that
        // daemon-signaled `InputClosed` cannot overtake in-flight zenoh data.
        let publishers = std::mem::take(&mut self.zenoh_publishers);
        let schema_publishers = std::mem::take(&mut self.zenoh_schema_publishers);
        let shm_provider = self.sample_allocator.shm_provider.take();
        let session = self.zenoh_session.take();
        let runtime = self._owned_runtime.take();
        if session.is_none() && shm_provider.is_none() && publishers.is_empty() {
            // no zenoh state (interactive/testing mode): drop inline. A node
            // without a zenoh session never has a handshake, but drop it here
            // too so this branch stays self-contained.
            drop(startup_handshake);
            drop(runtime);
        } else {
            // A wedged zenoh net runtime stalls `Session` close beyond its
            // 10s timeout and `Publisher` undeclare indefinitely, which would
            // hang node shutdown (and with it the daemon, which waits for
            // `InputClosed`). Bound the teardown with a deadline instead.
            let completed = teardown_with_timeout("zenoh", ZENOH_TEARDOWN_TIMEOUT, move || {
                // Stop + join the marker thread first (bounded by this
                // deadline), which releases its publishers `Arc` clone so the
                // `drop(publishers)` below holds the last reference and
                // undeclares them. Then the documented drop order: subscribers
                // (dropped with the handshake) and publishers (data + schema)
                // before the session, owned runtime last so async cleanup can
                // still run.
                if let Some(mut handshake) = startup_handshake {
                    handshake.shutdown();
                    // Undeclare the ack subscribers before the session below.
                    drop(std::mem::take(&mut handshake.ack_subscribers));
                }
                drop(publishers);
                drop(schema_publishers);
                drop(shm_provider);
                drop(session);
                drop(runtime);
            });
            if !completed {
                warn!(
                    "zenoh teardown timed out after {}s; continuing node shutdown",
                    ZENOH_TEARDOWN_TIMEOUT.as_secs()
                );
            }
        }

        // close all outputs first to notify subscribers as early as possible
        //
        // Testing mode (dora-rs/dora#2855): signal shutdown *before* the
        // CloseOutputs handshake so a daemon thread sleeping inside
        // `next_event` wakes up, replies, and can then process Drop's requests.
        if let Some(shutdown) = &self.testing_shutdown {
            shutdown.store(true, Ordering::Relaxed);
        }
        if let Err(err) = self
            .control_channel
            .report_closed_outputs(
                std::mem::take(&mut self.node_config.outputs)
                    .into_iter()
                    .collect(),
            )
            .context("failed to close outputs on drop")
        {
            tracing::warn!("{err:?}")
        }

        if let Err(err) = self.control_channel.report_outputs_done() {
            tracing::warn!("{err:?}")
        }

        // Drop our channel sender and join the testing daemon. The daemon loop
        // exits after OutputsDone under shutdown even when EventStream still
        // holds a sender clone (dora-rs/dora#2855).
        if let Some(handle) = self.testing_daemon.take() {
            self.control_channel.close_channel();
            if handle.join().is_err() {
                tracing::warn!("testing daemon thread panicked");
            }
        }
        self.testing_shutdown = None;
    }
}

/// A payload already encoded as an Arrow IPC stream in a dora-owned sample,
/// together with the Arrow type it was encoded from.
///
/// The two travel together so a consumer can still type-check the output after
/// the source array is gone — see [`DoraNode::send_output_encoded`].
#[derive(Debug)]
pub struct EncodedSample {
    sample: DataSample,
    data_type: arrow_schema::DataType,
}

impl EncodedSample {
    /// A human-readable name for the Arrow type the payload was encoded from,
    /// e.g. `"UInt8"`.
    ///
    /// Returned as a `String` rather than an `arrow_schema::DataType` because
    /// `arrow-schema` is the one non-umbrella Arrow crate dora's public API
    /// used to name, and naming it would pin 1.x to a single Arrow major.
    /// For the real type, enable `arrow-v59` and use
    /// [`data_type`](Self::data_type).
    pub fn type_name(&self) -> String {
        format!("{:?}", self.data_type)
    }

    /// The Arrow type the payload was encoded from.
    ///
    /// Gated on `arrow-v59` — dora's current internal Arrow major — because
    /// `arrow_schema::DataType` is an Arrow type. It is not returned as a
    /// dora-owned type-URN (`dora_core::types::TypeRegistry`) because the URN
    /// catalog only covers the standard scalar/struct types: nested lists,
    /// dictionaries, timestamps-with-timezone and unions have no URN, so a
    /// URN-returning accessor would be lossy for exactly the outputs whose
    /// type a caller most needs to inspect.
    #[cfg(feature = "arrow-v59")]
    pub fn data_type(&self) -> &arrow_schema::DataType {
        &self.data_type
    }

    /// The encoded Arrow IPC stream.
    pub fn as_bytes(&self) -> &[u8] {
        &self.sample
    }
}

/// Builds dora-owned output samples without borrowing the node.
///
/// ## Why this exists (dora-rs/dora#2742)
///
/// An operator runs on its own thread and hands its outputs to the runtime's
/// event loop. If what crosses that boundary is an Arrow array whose buffers
/// belong to the operator's language runtime — a `pyarrow` array wrapping a
/// numpy buffer, say — then the *runtime* ends up freeing them. Releasing a
/// numpy-backed buffer acquires the Python GIL (pyarrow's `NumPyBuffer`
/// destructor does `PyAcquireGIL`), so the runtime's event loop blocks for as
/// long as the operator holds the GIL. That made a node unable to observe
/// `Stop`, and the daemon force-killed it at the grace period.
///
/// Handing the operator thread an allocator instead lets it encode into memory
/// dora owns and release its own payload while it still holds the GIL. It is
/// not an extra copy: the IPC encode is the same single copy the node would
/// otherwise have made, just performed on the other side of the channel.
#[derive(Clone)]
pub struct SampleAllocator {
    shm_provider: Option<Arc<zenoh::shm::ShmProvider<zenoh::shm::PosixShmProviderBackend>>>,
    zero_copy_threshold: usize,
}

impl SampleAllocator {
    /// Allocates a [`DataSample`] of the specified size.
    ///
    /// For payloads at or above the zero-copy threshold the buffer is allocated
    /// directly from the zenoh SHM provider (when available), so the producer
    /// writes straight into shared memory and publishing moves the buffer into
    /// zenoh's `put` without a further copy. Smaller payloads — or the case
    /// where no SHM provider exists (interactive/testing mode) — use a
    /// heap-allocated, 128-byte-aligned buffer; the SHM provider is
    /// page-aligned, so dedicating a full page to a small message is pure waste.
    pub fn allocate(&self, data_len: usize) -> NodeResult<DataSample> {
        if data_len >= self.zero_copy_threshold
            && let Some(provider) = &self.shm_provider
        {
            use zenoh::Wait;
            use zenoh::shm::GarbageCollect;
            // Non-blocking (see `zenoh_publish`): GC and allocate, but fall back
            // to a heap buffer rather than `BlockOn`-sleeping 1 ms when the pool
            // is momentarily full under a large-message burst. The heap buffer
            // costs one extra copy on publish but keeps the producer from
            // stalling, which is what regressed sustained throughput (PR #2366).
            match provider
                .alloc(data_len)
                .with_policy::<GarbageCollect>()
                .wait()
            {
                Ok(sbuf) => {
                    // Use the SHM buffer only when it is exactly the requested
                    // size — zenoh 1.8 guarantees this (the logical length
                    // matches the request even when the backing chunk is
                    // larger). If a future provider ever over-allocates, fall
                    // back to heap rather than expose or publish an oversized
                    // slice (`DataSample` has no length cap of its own).
                    if sbuf.as_ref().len() == data_len {
                        return Ok(DataSample {
                            storage: SampleStorage::Shm(sbuf),
                        });
                    }
                    tracing::debug!(
                        "zenoh SHM alloc returned {} bytes for a {data_len}-byte \
                         request; using heap",
                        sbuf.as_ref().len()
                    );
                }
                Err(e) => {
                    tracing::debug!("SHM alloc failed ({e}), using heap buffer");
                }
            }
        }

        let avec: AVec<u8, ConstAlign<128>> = AVec::__from_elem(128, 0, data_len);
        Ok(avec.into())
    }

    /// Encodes `array` as a complete Arrow IPC stream into a freshly allocated
    /// sample. Uses the hand-rolled 1-copy fast path when the array type is
    /// eligible, falling back to the official writer (one extra copy) otherwise.
    ///
    /// The returned sample shares no memory with `array`, so the caller may —
    /// and, when the payload is owned by a foreign runtime, **must** — drop
    /// `array` on its own thread rather than let it travel to the node.
    pub fn encode_arrow(&self, array: &DoraArray) -> NodeResult<EncodedSample> {
        self.encode_arrow_data(&dora_arrow_convert::internal::array_ref(array).to_data())
    }

    /// Same, for dora-internal callers that already hold an [`ArrayData`].
    pub(crate) fn encode_arrow_data(&self, array: &ArrayData) -> NodeResult<EncodedSample> {
        let sample = match ipc_encode::PreparedIpc::from_data(array) {
            Some(prepared) => {
                // Prepare once: size the sample from the prepared layout, then
                // encode into it — avoids rebuilding the layout + IPC headers.
                let mut sample = self.allocate(prepared.byte_len())?;
                prepared
                    .encode_into(&mut sample)
                    .map_err(|e| NodeError::Output(format!("Arrow IPC encode: {e}")))?;
                sample
            }
            None => {
                let bytes = ipc_encode::encode_ipc_to_vec_data(array)
                    .map_err(|e| NodeError::Output(format!("Arrow IPC encode: {e}")))?;
                let mut sample = self.allocate(bytes.len())?;
                sample.copy_from_slice(&bytes);
                sample
            }
        };
        Ok(EncodedSample {
            sample,
            data_type: array.data_type().clone(),
        })
    }

    /// An allocator with no shared memory, so every sample is heap-backed.
    ///
    /// This is what a node without a zenoh session (interactive/testing mode)
    /// uses; it also lets callers that only need the encoding — tests, most
    /// obviously — build one without a live node.
    pub fn heap() -> Self {
        Self {
            shm_provider: None,
            zero_copy_threshold: ZERO_COPY_THRESHOLD,
        }
    }
}

impl std::fmt::Debug for SampleAllocator {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("SampleAllocator")
            .field("shm", &self.shm_provider.is_some())
            .field("zero_copy_threshold", &self.zero_copy_threshold)
            .finish()
    }
}

/// A data region suitable for sending as an output message.
///
/// `DataSample` implements the [`Deref`](std::ops::Deref) and
/// [`DerefMut`](std::ops::DerefMut) traits to read and write the mapped data.
///
/// The backing storage is either a heap buffer or — for payloads at or above
/// the zero-copy threshold when a zenoh SHM provider is available — a
/// zenoh-shared-memory buffer. Writing into an SHM-backed sample lets the
/// producer construct the message straight in shared memory, so publishing it
/// needs no further copy (the SHM buffer is moved directly into zenoh's `put`).
pub struct DataSample {
    storage: SampleStorage,
}

/// Backing storage for a [`DataSample`]. Kept private so the public API never
/// exposes a zenoh SHM type; callers only ever see the `[u8]` view via
/// `Deref`/`DerefMut`.
enum SampleStorage {
    /// Heap-allocated, 128-byte-aligned buffer (used below the zero-copy
    /// threshold or when no SHM provider is available).
    Heap(AVec<u8, ConstAlign<128>>),
    /// Zenoh shared-memory buffer. The producer writes the payload directly
    /// into it and the buffer is later moved into the zenoh `put` without
    /// copying.
    Shm(zenoh::shm::ZShmMut),
}

impl DataSample {
    /// Consume the sample into a [`FinalizedSample`] ready for transport.
    fn finalize(self) -> FinalizedSample {
        match self.storage {
            SampleStorage::Heap(buffer) => FinalizedSample::Vec(buffer),
            SampleStorage::Shm(sbuf) => FinalizedSample::Shm(sbuf),
        }
    }
}

impl std::ops::Deref for DataSample {
    type Target = [u8];

    fn deref(&self) -> &Self::Target {
        match &self.storage {
            SampleStorage::Heap(buffer) => buffer,
            SampleStorage::Shm(sbuf) => sbuf.as_ref(),
        }
    }
}

impl std::ops::DerefMut for DataSample {
    fn deref_mut(&mut self) -> &mut Self::Target {
        match &mut self.storage {
            SampleStorage::Heap(buffer) => buffer,
            SampleStorage::Shm(sbuf) => sbuf.as_mut(),
        }
    }
}

impl From<AVec<u8, ConstAlign<128>>> for DataSample {
    fn from(value: AVec<u8, ConstAlign<128>>) -> Self {
        Self {
            storage: SampleStorage::Heap(value),
        }
    }
}

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

/// A finalized output payload ready for transport.
///
/// Kept separate from [`DataMessage`] so SHM buffers stay out of the
/// `Serialize`/`Deserialize` TCP path: the zenoh data plane moves an `Shm`
/// buffer straight into `put` (zero extra copy), while the daemon fallback
/// converts to [`DataMessage::Vec`], copying out of shared memory only when the
/// zenoh path could not deliver.
enum FinalizedSample {
    Vec(AVec<u8, ConstAlign<128>>),
    Shm(zenoh::shm::ZShmMut),
}

impl FinalizedSample {
    fn byte_len(&self) -> usize {
        match self {
            FinalizedSample::Vec(v) => v.len(),
            FinalizedSample::Shm(sbuf) => sbuf.as_ref().len(),
        }
    }

    /// Convert into a TCP-transportable [`DataMessage`]. For the `Shm` arm this
    /// copies the payload out of shared memory into a heap buffer; it runs only
    /// on the daemon fallback (no matching zenoh subscriber / no session).
    fn into_data_message(self) -> DataMessage {
        match self {
            FinalizedSample::Vec(v) => DataMessage::Vec(v),
            FinalizedSample::Shm(sbuf) => {
                let bytes = sbuf.as_ref();
                let mut avec: AVec<u8, ConstAlign<128>> = AVec::__from_elem(128, 0, bytes.len());
                avec.copy_from_slice(bytes);
                DataMessage::Vec(avec)
            }
        }
    }
}

/// Outcome of a zenoh publish attempt.
enum PublishOutcome {
    /// The payload was delivered to zenoh (or, on a rare SHM put error,
    /// consumed and lost — see [`DoraNode::zenoh_publish`]).
    Published,
    /// No matching subscriber, or a transport error before the payload was
    /// consumed. The sample is returned so the caller can fall back to the
    /// daemon path.
    NotPublished(FinalizedSample),
}

/// FNV-1a hash of `bytes` with a fixed seed (cross-process deterministic).
/// Delegates to [`dora_message::metadata::fnv1a`] — the single source of truth
/// shared with the daemon's `dora topic` debug path, so schema hashes match.
pub(crate) fn fnv1a(bytes: &[u8]) -> u64 {
    dora_message::metadata::fnv1a(bytes)
}

/// How often a schema-once output re-sends a full self-describing stream on the
/// data topic. Full streams prime receivers in-band, so this bounds how long a
/// consumer that missed the single `@schema` emission (e.g. a failed zenoh-ext
/// history query) drops schema-less batches: it re-primes at the next refresh
/// instead of losing the input permanently. ~400 B of extra framing per output
/// per interval — negligible.
pub(crate) const SCHEMA_ONCE_REFRESH_INTERVAL: Duration = Duration::from_secs(5);

/// Producer-side schema-once state for one output.
struct SchemaOnceState {
    /// Hash of the schema confirmed published on the `@schema` subtopic.
    published_hash: u64,
    /// When the last full self-describing stream was sent on the data topic.
    last_full_stream: Instant,
}

/// What `publish_schema_once` should do for the current message.
#[derive(Debug)]
enum SchemaOnceDecision {
    /// The schema for this hash is not confirmed published (first message,
    /// schema change, or an earlier `@schema` publish failed): publish it and
    /// send this message as a full self-describing stream. The full stream
    /// decodes standalone and primes receivers in-band — in data-plane order —
    /// so the first message of an output cannot be lost to the express batch
    /// racing ahead of the schema on the separate `@schema` plane, and a failed
    /// schema publish degrades to "full stream every message" (decodable)
    /// instead of "hash-tagged but undecodable" (dora-rs/dora#2366 review).
    PublishSchemaAndSendFullStream,
    /// The periodic full-stream refresh is due (see
    /// [`SCHEMA_ONCE_REFRESH_INTERVAL`]).
    SendFullStreamRefresh,
    /// Schema confirmed published and fresh: send only the schema-less batch,
    /// tagged with the schema hash.
    SendSchemaLessBatch,
}

fn schema_once_decision(
    state: Option<&SchemaOnceState>,
    hash: u64,
    now: Instant,
) -> SchemaOnceDecision {
    match state {
        Some(state) if state.published_hash == hash => {
            if now.duration_since(state.last_full_stream) >= SCHEMA_ONCE_REFRESH_INTERVAL {
                SchemaOnceDecision::SendFullStreamRefresh
            } else {
                SchemaOnceDecision::SendSchemaLessBatch
            }
        }
        _ => SchemaOnceDecision::PublishSchemaAndSendFullStream,
    }
}

/// Publish the Arrow IPC schema for `output_id` on its `@schema` subtopic when
/// it changes, and return the attachment metadata (carrying the schema hash)
/// for the schema-less batch the caller sends on the data topic. Returns `None`
/// when the caller must send the full self-describing stream instead: on the
/// message that (re)publishes the schema, when the `@schema` publish failed,
/// for the periodic full-stream refresh, or if `full_stream` is not a parseable
/// IPC stream (see [`SchemaOnceDecision`]).
///
/// Takes the maps by `&mut` (not `&mut self`) so it can run while an immutable
/// borrow of `self.zenoh_publishers` (the data publisher) is live.
#[allow(clippy::too_many_arguments)]
fn publish_schema_once(
    schema_publishers: &mut HashMap<DataId, zenoh_ext::AdvancedPublisher<'static>>,
    schema_state: &mut HashMap<DataId, SchemaOnceState>,
    session: &zenoh::Session,
    dataflow_id: DataflowId,
    node_id: &NodeId,
    output_id: &DataId,
    full_stream: &[u8],
    base_metadata: &Metadata,
) -> Option<Vec<u8>> {
    let (hash, schema_bytes) = arrow_utils::ipc_encode::schema_block_and_hash(full_stream)?;

    let now = Instant::now();
    let decision = schema_once_decision(schema_state.get(output_id), hash, now);
    tracing::debug!(output = %output_id, decision = ?decision, "schema-once decision");

    match decision {
        SchemaOnceDecision::PublishSchemaAndSendFullStream => {
            if let Some(publisher) =
                schema_publisher(schema_publishers, session, dataflow_id, node_id, output_id)
            {
                use zenoh::Wait;
                match publisher.put(schema_bytes).wait() {
                    // Record the hash only on a successful publish, so a failed
                    // emission is retried on the next message rather than
                    // silently skipped.
                    Ok(()) => {
                        tracing::debug!(output = %output_id, hash, "schema published on @schema subtopic");
                        schema_state.insert(
                            output_id.clone(),
                            SchemaOnceState {
                                published_hash: hash,
                                last_full_stream: now,
                            },
                        );
                    }
                    Err(e) => {
                        tracing::warn!(output = %output_id, "failed to publish schema on @schema subtopic ({e})");
                    }
                }
            }
            None
        }
        SchemaOnceDecision::SendFullStreamRefresh => {
            tracing::debug!(output = %output_id, hash, "sending full-stream refresh");
            if let Some(state) = schema_state.get_mut(output_id) {
                state.last_full_stream = now;
            }
            None
        }
        SchemaOnceDecision::SendSchemaLessBatch => {
            tracing::debug!(output = %output_id, hash, "sending schema-less batch with SCHEMA_HASH");
            // Every batch carries the schema hash so the receiver can match it
            // to the primed decoder (and detect a schema change).
            let mut metadata = base_metadata.clone();
            metadata
                .parameters
                .insert(SCHEMA_HASH.to_string(), Parameter::Integer(hash as i64));
            dora_message::encode(&metadata).ok()
        }
    }
}

/// Get or lazily declare the schema `AdvancedPublisher` for `output_id` on its
/// `@schema` subtopic. The cache (depth 1) retains the last schema so a
/// late-joining subscriber's history query can fetch it; `publisher_detection`
/// lets a subscriber that started first discover this publisher and query its
/// cache. `CongestionControl::Block` keeps the single live schema emission from
/// being dropped under congestion.
fn schema_publisher<'a>(
    schema_publishers: &'a mut HashMap<DataId, zenoh_ext::AdvancedPublisher<'static>>,
    session: &zenoh::Session,
    dataflow_id: DataflowId,
    node_id: &NodeId,
    output_id: &DataId,
) -> Option<&'a zenoh_ext::AdvancedPublisher<'static>> {
    if !schema_publishers.contains_key(output_id) {
        use zenoh::Wait;
        use zenoh::qos::CongestionControl;
        use zenoh_ext::{AdvancedPublisherBuilderExt, CacheConfig, MissDetectionConfig};

        let topic = dora_core::topics::zenoh_output_schema_topic(dataflow_id, node_id, output_id);
        let key = zenoh::key_expr::KeyExpr::new(topic).ok()?.into_owned();
        let publisher = match session
            .declare_publisher(key)
            .congestion_control(CongestionControl::Block)
            // `sample_miss_detection` selects SequenceNumber sequencing instead of
            // the cache's default Timestamp sequencing, so the schema publisher
            // doesn't require session-wide timestamping (which would otherwise add
            // an HLC timestamp to every data-plane message too). Its default
            // config adds no heartbeat, so there's no extra periodic traffic.
            .sample_miss_detection(MissDetectionConfig::default())
            .cache(CacheConfig::default())
            .publisher_detection()
            .wait()
        {
            Ok(p) => p,
            Err(e) => {
                tracing::warn!(output = %output_id, "failed to declare schema publisher ({e})");
                return None;
            }
        };
        schema_publishers.insert(output_id.clone(), publisher);
    }
    schema_publishers.get(output_id)
}

pub(crate) use dora_message::metadata::carries_pattern_correlation;

/// Whether the schema-once optimization may be applied to a data-plane message.
///
/// Eligible only when the payload is below the zero-copy threshold *and* the
/// output does not interleave multiple Arrow schemas. Service/action
/// request-reply messages (`request_id`/`goal_id`/`goal_status`) multiplex
/// response schemas per request and must travel as full self-describing streams
/// so each decodes standalone; streaming chunks share one schema and stay
/// eligible. See the rationale at the call site in `zenoh_publish`.
fn schema_once_eligible(
    payload_len: usize,
    zero_copy_threshold: usize,
    params: &MetadataParameters,
) -> bool {
    payload_len < zero_copy_threshold && !carries_pattern_correlation(params)
}

/// Init Opentelemetry Tracing
///
/// This requires a tokio runtime spawning this function to be functional
#[cfg(feature = "tracing")]
pub fn init_tracing(
    node_id: &NodeId,
    dataflow_id: &DataflowId,
) -> NodeResult<Arc<Mutex<Option<OtelGuard>>>> {
    let node_id_str = node_id.to_string();
    let guard: Arc<Mutex<Option<OtelGuard>>> = Arc::new(Mutex::new(None));
    let clone = guard.clone();
    let tracing_monitor = async move {
        let mut builder = TracingBuilder::new(node_id_str.clone());
        // Only enable OTLP if environment variable is set
        if std::env::var("DORA_OTLP_ENDPOINT").is_ok()
            || std::env::var("DORA_JAEGER_TRACING").is_ok()
        {
            match builder.with_otlp_tracing() {
                Ok(b) => {
                    builder = b.with_stdout("info", true);
                    if let Ok(mut guard) = clone.lock() {
                        *guard = builder.guard.take();
                    }
                }
                Err(e) => {
                    eprintln!("warning: failed to set up OTLP tracing: {e:?}");
                    // Rebuild without OTLP — with_otlp_tracing consumed builder
                    builder = TracingBuilder::new(node_id_str).with_stdout("info", true);
                }
            }
        } else {
            builder = builder.with_stdout("info", true);
        }

        if let Err(e) = builder.build() {
            eprintln!("warning: failed to set up tracing subscriber: {e:?}");
        }
    };

    let rt = Handle::try_current().context("failed to get tokio runtime handle")?;
    rt.spawn(tracing_monitor);

    // dataflow_id is only used when metrics feature is enabled
    let _ = &dataflow_id;

    // Only start the OTLP metrics exporter when an endpoint is configured.
    // The exporter schedules via `tokio::time::interval` and would otherwise
    // panic on callers whose runtime lacks the time driver, and would also
    // attempt to connect to `localhost:4317` on every node startup. Mirrors
    // the gating applied to tracing above.
    #[cfg(feature = "metrics")]
    if let Ok(endpoint) = std::env::var("DORA_OTLP_ENDPOINT") {
        let id = format!("{dataflow_id}/{node_id}");
        let monitor_task = async move {
            use dora_metrics::run_metrics_monitor;

            if let Err(e) = run_metrics_monitor(id.clone(), &endpoint)
                .await
                .wrap_err("metrics monitor exited unexpectedly")
            {
                warn!("metrics monitor failed: {:#?}", e);
            }
        };
        let rt = Handle::try_current().context("failed to get tokio runtime handle")?;
        rt.spawn(monitor_task);
    }
    Ok(guard)
}

/// Builder for streaming segment metadata.
///
/// Manages session/segment IDs and auto-incrementing sequence numbers
/// for real-time streaming patterns (voice, video, sensor streams).
///
/// The state transitions are easy to get subtly wrong, so they are worth
/// spelling out: [`chunk`](Self::chunk) stamps the current `(segment_id, seq)`
/// and then auto-increments `seq`; [`next_segment`](Self::next_segment) bumps
/// `segment_id` and resets `seq` to 0; and [`flush`](Self::flush) advances to a
/// new segment and emits a chunk marked `flush = true`, `fin = false` (the
/// prior segment is discarded, not completed, so it intentionally never gets a
/// `fin = true`).
///
/// # Example
///
/// ```
/// use dora_node_api::{
///     StreamSegment,
///     metadata::{FIN, FLUSH, SEGMENT_ID, SEQ, get_bool_param, get_integer_param},
/// };
///
/// let mut seg = StreamSegment::with_session_id("session-1".to_string());
///
/// // `chunk` stamps the current (segment, seq), then advances seq.
/// let first = seg.chunk(false);
/// assert_eq!(get_integer_param(&first, SEGMENT_ID), Some(0));
/// assert_eq!(get_integer_param(&first, SEQ), Some(0));
/// assert_eq!(get_bool_param(&first, FIN), Some(false));
///
/// let second = seg.chunk(true); // mark this chunk as the end of the segment
/// assert_eq!(get_integer_param(&second, SEQ), Some(1)); // seq auto-incremented
/// assert_eq!(get_bool_param(&second, FIN), Some(true));
///
/// // `flush` starts a new segment (seq reset to 0) and marks flush=true,
/// // fin=false: the old queued data is discarded, not completed.
/// let flushed = seg.flush();
/// assert_eq!(get_integer_param(&flushed, SEGMENT_ID), Some(1));
/// assert_eq!(get_integer_param(&flushed, SEQ), Some(0));
/// assert_eq!(get_bool_param(&flushed, FLUSH), Some(true));
/// assert_eq!(get_bool_param(&flushed, FIN), Some(false));
/// ```
pub struct StreamSegment {
    session_id: String,
    segment_id: i64,
    seq: i64,
}

impl StreamSegment {
    /// Start a new session with a generated session ID and segment 0.
    pub fn new() -> Self {
        Self {
            session_id: DoraNode::new_request_id(),
            segment_id: 0,
            seq: 0,
        }
    }

    /// Start a new session with an explicit session ID.
    pub fn with_session_id(session_id: String) -> Self {
        Self {
            session_id,
            segment_id: 0,
            seq: 0,
        }
    }

    /// Advance to a new segment (resets seq to 0). Returns the new segment_id.
    pub fn next_segment(&mut self) -> i64 {
        self.segment_id += 1;
        self.seq = 0;
        self.segment_id
    }

    /// Build metadata parameters for a chunk. Auto-increments seq.
    pub fn chunk(&mut self, fin: bool) -> MetadataParameters {
        let mut params = MetadataParameters::new();
        params.insert(
            SESSION_ID.into(),
            Parameter::String(self.session_id.clone()),
        );
        params.insert(SEGMENT_ID.into(), Parameter::Integer(self.segment_id));
        params.insert(SEQ.into(), Parameter::Integer(self.seq));
        params.insert(FIN.into(), Parameter::Bool(fin));
        self.seq += 1;
        params
    }

    /// Build metadata for a flush message (new segment, discards older queued data).
    ///
    /// Advances to a new segment, then emits a chunk with `flush=true` and
    /// `fin=false`. The prior segment ends without a `fin=true` signal -- this
    /// is intentional for interruption semantics (the old data is being
    /// discarded, not completed).
    ///
    /// **Note**: flush discards *all* queued messages on the receiver's input
    /// regardless of `session_id`. Do not multiplex independent sessions on a
    /// single `DataId` when using flush.
    pub fn flush(&mut self) -> MetadataParameters {
        self.next_segment();
        let mut params = self.chunk(false);
        params.insert(FLUSH.into(), Parameter::Bool(true));
        params
    }

    /// Returns the session ID.
    pub fn session_id(&self) -> &str {
        &self.session_id
    }

    /// Returns the current segment ID.
    pub fn segment_id(&self) -> i64 {
        self.segment_id
    }

    /// Returns the sequence number that will be used by the next `chunk()` call.
    pub fn seq(&self) -> i64 {
        self.seq
    }
}

impl Default for StreamSegment {
    fn default() -> Self {
        Self::new()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::integration_testing::{
        IntegrationTestInput, TestingInput, TestingOptions, TestingOutput,
        integration_testing_format::{IncomingEvent, TimedIncomingEvent},
    };

    fn required_acker(node: &str, input: &str) -> dora_message::daemon_to_node::RequiredAcker {
        dora_message::daemon_to_node::RequiredAcker {
            node_id: NodeId::from(node.to_string()),
            input_id: DataId::from(input.to_string()),
        }
    }

    /// A not-yet-ready `AckState` for `output` awaiting a single `acker`.
    fn test_ack_state(output: &str, acker: (&str, &str)) -> Arc<AckState> {
        Arc::new(AckState::new(
            DataId::from(output.to_string()),
            &BTreeSet::from([required_acker(acker.0, acker.1)]),
            Arc::new(AtomicBool::new(false)),
        ))
    }

    #[test]
    fn log_fields_budget_measures_serialized_json_not_raw_bytes() {
        use std::collections::BTreeMap;
        let limit = DoraNode::MAX_LOG_FIELDS_BYTES;

        // A small map fits.
        let mut small = BTreeMap::new();
        small.insert("k".to_string(), "v".to_string());
        assert!(log_fields_within_budget(&small, limit).is_some());

        // A value that is 20 KB of raw bytes — comfortably under the 60 KB
        // budget by the old raw-sum measure — but made entirely of control
        // characters, each of which JSON-escapes to `
` (6 bytes). Its
        // serialized form is ~120 KB, over the budget, so it must be dropped.
        // The pre-fix raw-byte check would have let it through and blown the
        // downstream parse limit.
        let mut big = BTreeMap::new();
        big.insert("k".to_string(), "\u{1}".repeat(20 * 1024));
        assert!(big.values().map(String::len).sum::<usize>() < limit);
        assert!(log_fields_within_budget(&big, limit).is_none());
    }

    #[test]
    fn ack_state_completes_only_when_required_set_is_covered() {
        let ready = Arc::new(AtomicBool::new(false));
        let required = BTreeSet::from([
            required_acker("sink-a", "camera"),
            required_acker("sink-b", "cam"),
        ]);
        let state = AckState::new(DataId::from("image".to_string()), &required, ready.clone());

        // An acker outside the required set (dynamic or debug consumer) must
        // never count toward completion.
        state.record("stranger", "camera");
        assert!(!ready.load(Ordering::Relaxed));

        // Duplicate acks of one required identity don't complete the set.
        state.record("sink-a", "camera");
        state.record("sink-a", "camera");
        assert!(!ready.load(Ordering::Relaxed));
        assert_eq!(state.missing(), vec!["sink-b/cam".to_string()]);

        // The last required identity completes it.
        state.record("sink-b", "cam");
        assert!(ready.load(Ordering::Relaxed));
        assert!(state.missing().is_empty());
    }

    #[test]
    fn ack_state_requires_exact_identity_match() {
        // (node, input) is one identity: the right node acking the wrong input
        // (or vice versa) must not count.
        let ready = Arc::new(AtomicBool::new(false));
        let required = BTreeSet::from([required_acker("sink", "camera")]);
        let state = AckState::new(DataId::from("image".to_string()), &required, ready.clone());

        state.record("sink", "other-input");
        state.record("other-node", "camera");
        assert!(!ready.load(Ordering::Relaxed));

        state.record("sink", "camera");
        assert!(ready.load(Ordering::Relaxed));
    }

    // Regression guard for dora-rs/dora#2891: a frozen output must never
    // upgrade. Before the fix an ack arriving after the grace (but before the
    // old 10s deadline) flipped `ready` while user code was already sending,
    // so a message on the shorter direct-zenoh path could overtake an earlier
    // one still relaying through the daemon.
    #[test]
    fn ack_state_freeze_blocks_a_late_upgrade() {
        let state = test_ack_state("image", ("sink", "camera"));

        assert!(state.freeze(), "an un-acked output is frozen");
        assert!(state.is_frozen());

        // The consumer's ack arrives late — it must not move the output onto
        // the direct path mid-stream.
        state.record("sink", "camera");
        assert!(
            !state.ready.load(Ordering::Relaxed),
            "a frozen output must stay on the daemon path for the rest of the run"
        );
        assert_eq!(state.missing(), vec!["sink/camera".to_string()]);
    }

    #[test]
    fn ack_state_freeze_spares_an_output_that_acked_in_time() {
        let state = test_ack_state("image", ("sink", "camera"));

        state.record("sink", "camera");
        assert!(!state.freeze(), "a ready output is not frozen");
        assert!(!state.is_frozen());
        assert!(
            state.ready.load(Ordering::Relaxed),
            "an output that proved its routes keeps the direct zenoh path"
        );
    }

    // dora-rs/dora#2891: whatever has not proven its routes when the grace
    // expires is pinned to the daemon path, so every output's transport is
    // decided before user code sends its first message.
    #[test]
    fn grace_boundary_freezes_unacked_outputs_only() {
        let acked = test_ack_state("image", ("sink", "camera"));
        let unacked = test_ack_state("status", ("sink", "state"));
        acked.record("sink", "camera");

        wait_for_grace(&[acked.clone(), unacked.clone()], Duration::from_millis(20));

        assert!(acked.ready.load(Ordering::Relaxed));
        assert!(!acked.is_frozen());
        assert!(unacked.is_frozen());

        // The straggler's ack lands after the boundary and is ignored.
        unacked.record("sink", "state");
        assert!(!unacked.ready.load(Ordering::Relaxed));
    }

    #[test]
    fn grace_returns_early_once_every_output_is_acked() {
        // Also covers the no-awaited-outputs case (no consumers, or every ack
        // subscriber failed to declare): `all` is vacuously true on an empty
        // slice, so there is nothing to wait for and nothing to freeze.
        let state = test_ack_state("image", ("sink", "camera"));
        state.record("sink", "camera");

        let start = Instant::now();
        wait_for_grace(std::slice::from_ref(&state), Duration::from_secs(30));
        wait_for_grace(&[], Duration::from_secs(30));

        assert!(!state.is_frozen());
        assert!(state.ready.load(Ordering::Relaxed));
        assert!(
            start.elapsed() < Duration::from_secs(5),
            "a completed handshake must not wait out the grace, took {:?}",
            start.elapsed()
        );
    }

    #[test]
    fn missing_output_routing_pins_every_output_to_the_daemon_path() {
        // `None` means an older daemon spawned this node: without
        // required-acker sets no route can be proven, so the safe result is
        // daemon-only for every declared output.
        let outputs = BTreeSet::from([
            DataId::from("image".to_string()),
            DataId::from("status".to_string()),
        ]);
        let routing = normalize_output_routing(None, &outputs);
        assert_eq!(routing.len(), 2);
        for output_id in &outputs {
            let entry = routing.get(output_id).expect("entry per output");
            assert!(entry.daemon_only);
            assert!(entry.required_ackers.is_empty());
        }

        // No outputs → nothing to pin (interactive mode).
        assert!(normalize_output_routing(None, &BTreeSet::new()).is_empty());
    }

    #[test]
    fn provided_output_routing_is_passed_through() {
        let outputs = BTreeSet::from([DataId::from("image".to_string())]);
        let provided = BTreeMap::from([(
            DataId::from("image".to_string()),
            OutputRouting {
                daemon_only: false,
                required_ackers: BTreeSet::from([required_acker("sink", "camera")]),
            },
        )]);
        let routing = normalize_output_routing(Some(provided.clone()), &outputs);
        assert_eq!(routing, provided);
    }

    #[test]
    fn new_request_id_returns_valid_uuid() {
        let id = DoraNode::new_request_id();
        uuid::Uuid::parse_str(&id).expect("should be valid UUID");
    }

    #[test]
    fn new_request_id_is_unique() {
        let ids: Vec<String> = (0..100).map(|_| DoraNode::new_request_id()).collect();
        let unique: std::collections::HashSet<_> = ids.iter().collect();
        assert_eq!(ids.len(), unique.len(), "all IDs should be unique");
    }

    #[test]
    fn new_goal_id_returns_valid_uuid() {
        let id = DoraNode::new_goal_id();
        uuid::Uuid::parse_str(&id).expect("should be valid UUID");
    }

    /// `DoraNode::timestamp()` must read from the SAME HLC the node
    /// uses to stamp outgoing messages. If a refactor accidentally
    /// gives `timestamp()` its own clock, the latency-measurement use
    /// case in the docstring silently breaks (subtracting against an
    /// `event.metadata.timestamp` from the data plane would mix two
    /// unrelated HLCs). Guard by asserting two calls share an HLC ID
    /// and that the second reads strictly later than the first.
    ///
    /// The strict `t2 > t1` assertion holds by HLC construction: if
    /// the wall clock advanced between calls, the physical component
    /// strictly increases; if not, the logical counter bumps. The
    /// lexicographic ordering on `uhlc::Timestamp` puts `t2` strictly
    /// after `t1` in either case, so this assertion does not flake on
    /// fast machines whose OS clock rounds both calls to the same tick.
    #[test]
    fn timestamp_uses_node_clock_and_is_monotonic() {
        let (node, events, _rx) = test_node();
        let t1 = node.timestamp();
        let t2 = node.timestamp();
        assert_eq!(
            t1.get_id(),
            t2.get_id(),
            "two timestamp() calls must come from the same HLC instance",
        );
        assert!(
            t2 > t1,
            "HLC timestamps must be strictly monotonic: {t1:?} >= {t2:?}"
        );
        drop(node);
        drop(events);
    }

    use crate::integration_testing::{OutputReceiver, drain_outputs};

    /// Helper: create a minimal test node with a channel output.
    fn test_node() -> (DoraNode, crate::EventStream, OutputReceiver) {
        let events = vec![TimedIncomingEvent {
            time_offset_secs: 0.1,
            event: IncomingEvent::Stop,
        }];
        let inputs = TestingInput::Input(IntegrationTestInput::new(
            "test-node".parse().unwrap(),
            events,
        ));
        let (tx, rx) = crate::integration_testing::output_channel();
        let outputs = TestingOutput::ToChannel(tx);
        let options = TestingOptions {
            skip_output_time_offsets: true,
        };
        let (node, event_stream) = DoraNode::init_testing(inputs, outputs, options).unwrap();
        (node, event_stream, rx)
    }

    /// Comfortably below any multi-second join/sleep budget so node-first Drop
    /// regressions surface without waiting on an internal timeout boundary.
    const INIT_TESTING_DROP_BUDGET: Duration = Duration::from_millis(500);

    fn init_testing_node_mid_scheduled_wait() -> (DoraNode, crate::EventStream) {
        let events = vec![TimedIncomingEvent {
            // Long enough that a hang is obvious; the shutdown flag must
            // interrupt well before this elapses.
            time_offset_secs: 30.0,
            event: IncomingEvent::Stop,
        }];
        let inputs = TestingInput::Input(IntegrationTestInput::new(
            "drop-hang-node".parse().unwrap(),
            events,
        ));
        let (tx, _rx) = crate::integration_testing::output_channel();
        let outputs = TestingOutput::ToChannel(tx);
        let (node, event_stream) =
            DoraNode::init_testing(inputs, outputs, TestingOptions::default()).unwrap();

        // Give the testing daemon a moment to enter next_event's sleep.
        std::thread::sleep(Duration::from_millis(50));
        (node, event_stream)
    }

    /// Regression for dora-rs/dora#2855: events-then-node Drop while the daemon
    /// is inside a scheduled `next_event` wait must not hang.
    #[test]
    fn init_testing_drop_events_then_node_during_scheduled_wait_does_not_hang() {
        let (node, event_stream) = init_testing_node_mid_scheduled_wait();

        let start = Instant::now();
        drop(event_stream);
        drop(node);
        let elapsed = start.elapsed();
        assert!(
            elapsed < INIT_TESTING_DROP_BUDGET,
            "events-then-node Drop hung for {elapsed:?}; expected interruptible testing-daemon shutdown"
        );
    }

    /// Regression for dora-rs/dora#2855: node-then-events Drop must exit the
    /// testing daemon after OutputsDone under shutdown even while EventStream
    /// still holds a channel sender clone.
    #[test]
    fn init_testing_drop_node_then_events_during_scheduled_wait_does_not_hang() {
        let (node, event_stream) = init_testing_node_mid_scheduled_wait();

        let start = Instant::now();
        drop(node);
        drop(event_stream);
        let elapsed = start.elapsed();
        assert!(
            elapsed < INIT_TESTING_DROP_BUDGET,
            "node-then-events Drop hung for {elapsed:?}; expected OutputsDone under shutdown to exit the testing daemon"
        );
    }

    #[test]
    fn send_service_request_returns_valid_id_and_sends_output() {
        let (mut node, events, mut rx) = test_node();

        let request_id = node
            .send_service_request("request".into(), Default::default(), ())
            .unwrap();

        // Returned ID should be a valid UUID
        uuid::Uuid::parse_str(&request_id).expect("returned request_id should be valid UUID");

        // Output should have been sent to the channel
        drop(node);
        drop(events);
        let outputs = drain_outputs(&mut rx);
        assert_eq!(outputs.len(), 1);
        assert_eq!(outputs[0]["id"], "request");
    }

    #[test]
    fn send_service_request_returns_unique_ids() {
        let (mut node, events, _rx) = test_node();

        let id1 = node
            .send_service_request("out".into(), Default::default(), ())
            .unwrap();
        let id2 = node
            .send_service_request("out".into(), Default::default(), ())
            .unwrap();

        assert_ne!(id1, id2, "successive request IDs should differ");

        drop(node);
        drop(events);
    }

    #[test]
    fn send_service_response_sends_output() {
        let (mut node, events, mut rx) = test_node();

        // Simulate passing through a request_id from the incoming request
        let mut params = MetadataParameters::default();
        params.insert(
            dora_message::metadata::REQUEST_ID.to_string(),
            dora_message::metadata::Parameter::String("test-req-id".into()),
        );
        node.send_service_response("response".into(), params, ())
            .unwrap();

        drop(node);
        drop(events);
        let outputs = drain_outputs(&mut rx);
        assert_eq!(outputs.len(), 1);
        assert_eq!(outputs[0]["id"], "response");
    }

    /// `send_output_bytes` must reject a `data_len` that disagrees with
    /// `data.len()` with a clear error instead of panicking inside
    /// `copy_from_slice` deep in `send_output_raw`.
    #[test]
    fn send_output_bytes_rejects_len_mismatch() {
        let (mut node, events, _rx) = test_node();

        let result = node.send_output_bytes("out".into(), Default::default(), 8, &[1, 2, 3, 4]);

        let err = result.expect_err("mismatched data_len must error, not panic");
        assert!(
            err.to_string().contains("does not match"),
            "unexpected error message: {err}"
        );

        drop(node);
        drop(events);
    }

    /// A heap-backed `DataSample` is writable through `DerefMut`, readable
    /// through `Deref`, and `finalize().into_data_message()` preserves the bytes
    /// as the `DataMessage::Vec` daemon-path payload. (The SHM-backed arm needs
    /// a live zenoh provider and is covered by the copy-count harness/smoke.)
    #[test]
    fn data_sample_heap_roundtrip() {
        let avec: AVec<u8, ConstAlign<128>> = AVec::__from_elem(128, 0, 8);
        let mut sample: DataSample = avec.into();
        sample.copy_from_slice(&[1, 2, 3, 4, 5, 6, 7, 8]);

        assert_eq!(&sample[..], &[1, 2, 3, 4, 5, 6, 7, 8]);
        assert_eq!(sample.len(), 8);

        match sample.finalize().into_data_message() {
            DataMessage::Vec(v) => assert_eq!(v.as_slice(), &[1, 2, 3, 4, 5, 6, 7, 8]),
        }
    }

    /// End-to-end wire contract: a few representative arrays IPC-encoded (fast
    /// path) into a sample and decoded back via `decode_arrow_ipc_zero_copy`
    /// must equal the input. This is the send->receive round-trip the data
    /// plane relies on (zenoh can't be smoke-tested here, so this stands in).
    #[test]
    fn send_output_ipc_roundtrip() {
        use crate::arrow_utils::decode_arrow_ipc_zero_copy_raw;
        use crate::arrow_utils::ipc_encode::{encode_ipc_into_data, ipc_fast_path_len_data};
        use arrow::array::{ArrayRef, Float32Array, StringArray, StructArray, UInt64Array};
        use arrow_schema::{DataType, Field};
        use std::ptr::NonNull;

        fn roundtrip(data: ArrayData) {
            let len = ipc_fast_path_len_data(&data).expect("array should be fast-path eligible");
            let mut buf: AVec<u8, ConstAlign<128>> = AVec::__from_elem(128, 0, len);
            encode_ipc_into_data(&data, &mut buf).expect("fast-path IPC encode");

            // Wrap the aligned sample as an Arrow Buffer (no copy), mirroring the
            // receive path, then decode.
            let ptr = NonNull::new(buf.as_ptr() as *mut u8).unwrap();
            let blen = buf.len();
            // SAFETY: ptr/len describe `buf`; the Arc keeps it alive.
            let buffer =
                unsafe { arrow::buffer::Buffer::from_custom_allocation(ptr, blen, Arc::new(buf)) };
            let decoded = decode_arrow_ipc_zero_copy_raw(buffer).expect("zero-copy IPC decode");
            assert_eq!(
                data, decoded,
                "IPC send->receive round-trip must preserve the array"
            );
        }

        roundtrip(Float32Array::from(vec![1.0, 2.5, -3.0, 4.0]).into_data());
        roundtrip(UInt64Array::from(vec![Some(1), None, Some(3)]).into_data());
        roundtrip(StringArray::from(vec![Some("hello"), None, Some("world")]).into_data());
        roundtrip(
            StructArray::from(vec![
                (
                    Arc::new(Field::new("v", DataType::UInt64, true)),
                    Arc::new(UInt64Array::from(vec![Some(1), None, Some(3)])) as ArrayRef,
                ),
                (
                    Arc::new(Field::new("s", DataType::Utf8, true)),
                    Arc::new(StringArray::from(vec![Some("a"), Some("bb"), None])) as ArrayRef,
                ),
            ])
            .into_data(),
        );
    }

    /// `close_outputs` must be atomic: if any id in the batch is unknown, the
    /// call fails *without* removing the valid ids from the local output set.
    /// Otherwise the daemon (never notified, because `report_closed_outputs` is
    /// skipped on error) and the node disagree about which outputs are open, and
    /// the node would silently drop subsequent sends to a still-open output.
    #[test]
    fn close_outputs_is_atomic_on_unknown_id() {
        let (mut node, events, _rx) = test_node();
        let valid: DataId = "valid".into();
        node.node_config.outputs.insert(valid.clone());

        let result = node.close_outputs(vec![valid.clone(), "unknown".into()]);

        assert!(
            result.is_err(),
            "closing a batch containing an unknown output must fail"
        );
        assert!(
            node.node_config.outputs.contains(&valid),
            "a failed close_outputs must not remove the valid output from local state"
        );

        drop(node);
        drop(events);
    }

    // ---- dora-rs/adora#150: pattern polymorphism exemption ----

    #[test]
    fn carries_pattern_correlation_detects_request_id() {
        let mut params = MetadataParameters::default();
        params.insert(
            dora_message::metadata::REQUEST_ID.to_string(),
            dora_message::metadata::Parameter::String("req-1".into()),
        );
        assert!(carries_pattern_correlation(&params));
    }

    #[test]
    fn carries_pattern_correlation_detects_goal_id() {
        let mut params = MetadataParameters::default();
        params.insert(
            dora_message::metadata::GOAL_ID.to_string(),
            dora_message::metadata::Parameter::String("goal-1".into()),
        );
        assert!(carries_pattern_correlation(&params));
    }

    #[test]
    fn carries_pattern_correlation_detects_goal_status() {
        let mut params = MetadataParameters::default();
        params.insert(
            dora_message::metadata::GOAL_STATUS.to_string(),
            dora_message::metadata::Parameter::String("succeeded".into()),
        );
        assert!(carries_pattern_correlation(&params));
    }

    #[test]
    fn carries_pattern_correlation_empty_is_not_a_pattern() {
        let params = MetadataParameters::default();
        assert!(!carries_pattern_correlation(&params));
    }

    #[test]
    fn carries_pattern_correlation_ignores_non_pattern_keys() {
        let mut params = MetadataParameters::default();
        params.insert(
            "custom_key".to_string(),
            dora_message::metadata::Parameter::String("value".into()),
        );
        assert!(!carries_pattern_correlation(&params));
    }

    #[test]
    fn schema_once_excludes_pattern_correlation_outputs() {
        // Regression (dora-rs/dora#2366 review): a small service/action
        // request-reply message — which multiplexes response schemas per request
        // — must NOT use schema-once, or a schema-less batch could reach a
        // consumer primed for a different schema and be silently dropped. It must
        // travel as a full self-describing stream instead.
        const THRESHOLD: usize = 4096;

        let plain = MetadataParameters::default();
        assert!(
            schema_once_eligible(100, THRESHOLD, &plain),
            "small message on a stable-schema output is eligible"
        );
        assert!(
            !schema_once_eligible(THRESHOLD, THRESHOLD, &plain),
            "a message at/above the threshold is not eligible (goes via SHM/full stream)"
        );

        for key in [
            dora_message::metadata::REQUEST_ID,
            dora_message::metadata::GOAL_ID,
            dora_message::metadata::GOAL_STATUS,
        ] {
            let mut params = MetadataParameters::default();
            params.insert(
                key.to_string(),
                dora_message::metadata::Parameter::String("x".into()),
            );
            assert!(
                !schema_once_eligible(100, THRESHOLD, &params),
                "small pattern-correlation message ({key}) must bypass schema-once"
            );
        }

        // Streaming is deliberately NOT excluded: every chunk of a stream shares
        // one schema, so streaming stays the high-rate beneficiary of
        // schema-once. Locking this in guards against a well-meaning "also
        // exclude streaming" change that would defeat the optimization.
        let mut stream = MetadataParameters::default();
        stream.insert(
            dora_message::metadata::SESSION_ID.to_string(),
            dora_message::metadata::Parameter::String("s1".into()),
        );
        stream.insert(
            dora_message::metadata::SEGMENT_ID.to_string(),
            dora_message::metadata::Parameter::Integer(0),
        );
        assert!(
            schema_once_eligible(100, THRESHOLD, &stream),
            "small streaming chunk (stable schema) stays eligible for schema-once"
        );
    }

    #[test]
    fn schema_once_decision_covers_publish_refresh_and_schema_less() {
        let start = Instant::now();
        let later = start + SCHEMA_ONCE_REFRESH_INTERVAL;
        let state = SchemaOnceState {
            published_hash: 7,
            last_full_stream: start,
        };

        // No state yet (first message of this output) → publish the schema and
        // send THIS message as a full stream: it decodes standalone and primes
        // receivers in-band, so the first message can never be lost to the
        // batch racing ahead of the schema on the separate `@schema` plane
        // (dora-rs/dora#2366 review).
        assert!(matches!(
            schema_once_decision(None, 7, start),
            SchemaOnceDecision::PublishSchemaAndSendFullStream
        ));
        // Schema changed (or an earlier `@schema` publish failed, which leaves
        // the recorded hash stale) → same: publish + full stream.
        assert!(matches!(
            schema_once_decision(Some(&state), 8, start),
            SchemaOnceDecision::PublishSchemaAndSendFullStream
        ));
        // Schema confirmed published and refresh not due → schema-less batch.
        assert!(matches!(
            schema_once_decision(Some(&state), 7, start),
            SchemaOnceDecision::SendSchemaLessBatch
        ));
        // Refresh due → send a full stream so any consumer that missed the
        // single `@schema` emission re-primes in-band within the interval
        // instead of losing the input permanently.
        assert!(matches!(
            schema_once_decision(Some(&state), 7, later),
            SchemaOnceDecision::SendFullStreamRefresh
        ));
    }

    #[test]
    fn stream_segment_new_generates_valid_session_id() {
        let seg = StreamSegment::new();
        uuid::Uuid::parse_str(seg.session_id()).expect("session_id should be valid UUID");
        assert_eq!(seg.segment_id(), 0);
    }

    #[test]
    fn stream_segment_with_session_id() {
        let seg = StreamSegment::with_session_id("my-session".into());
        assert_eq!(seg.session_id(), "my-session");
        assert_eq!(seg.segment_id(), 0);
        assert_eq!(seg.seq(), 0);
    }

    #[test]
    fn stream_segment_seq_accessor_tracks_next_seq() {
        let mut seg = StreamSegment::with_session_id("s1".into());
        assert_eq!(seg.seq(), 0);
        seg.chunk(false);
        assert_eq!(seg.seq(), 1);
        seg.chunk(false);
        assert_eq!(seg.seq(), 2);
        seg.next_segment();
        assert_eq!(seg.seq(), 0);
    }

    #[test]
    fn stream_segment_chunk_auto_increments_seq() {
        let mut seg = StreamSegment::with_session_id("s1".into());
        let p0 = seg.chunk(false);
        let p1 = seg.chunk(false);
        let p2 = seg.chunk(true);

        assert_eq!(p0.get(SEQ), Some(&Parameter::Integer(0)));
        assert_eq!(p1.get(SEQ), Some(&Parameter::Integer(1)));
        assert_eq!(p2.get(SEQ), Some(&Parameter::Integer(2)));
        assert_eq!(p0.get(FIN), Some(&Parameter::Bool(false)));
        assert_eq!(p2.get(FIN), Some(&Parameter::Bool(true)));
        assert_eq!(p0.get(SESSION_ID), Some(&Parameter::String("s1".into())));
        assert_eq!(p0.get(SEGMENT_ID), Some(&Parameter::Integer(0)));
    }

    #[test]
    fn stream_segment_next_segment_resets_seq() {
        let mut seg = StreamSegment::with_session_id("s1".into());
        seg.chunk(false); // seq=0
        seg.chunk(false); // seq=1
        let new_id = seg.next_segment();
        assert_eq!(new_id, 1);
        assert_eq!(seg.segment_id(), 1);

        let p = seg.chunk(false);
        assert_eq!(p.get(SEQ), Some(&Parameter::Integer(0)));
        assert_eq!(p.get(SEGMENT_ID), Some(&Parameter::Integer(1)));
    }

    #[test]
    fn stream_segment_flush_advances_segment_and_sets_flush() {
        let mut seg = StreamSegment::with_session_id("s1".into());
        seg.chunk(false);
        let p = seg.flush();
        assert_eq!(seg.segment_id(), 1);
        assert_eq!(p.get(FLUSH), Some(&Parameter::Bool(true)));
        assert_eq!(p.get(SEGMENT_ID), Some(&Parameter::Integer(1)));
        // flush resets seq, then chunk increments it to 1
        assert_eq!(p.get(SEQ), Some(&Parameter::Integer(0)));
    }

    #[test]
    fn send_stream_chunk_sends_output() {
        let (mut node, events, mut rx) = test_node();
        let mut seg = StreamSegment::with_session_id("s1".into());

        node.send_stream_chunk("audio".into(), &mut seg, false, ())
            .unwrap();

        drop(node);
        drop(events);
        let outputs = drain_outputs(&mut rx);
        assert_eq!(outputs.len(), 1);
        assert_eq!(outputs[0]["id"], "audio");
    }

    #[test]
    fn teardown_with_timeout_completes_fast_closure() {
        let start = Instant::now();
        let completed = teardown_with_timeout("fast", Duration::from_secs(5), || {});
        assert!(completed, "fast teardown should report completion");
        assert!(
            start.elapsed() < Duration::from_secs(5),
            "fast teardown should not wait for the full timeout"
        );
    }

    #[test]
    fn teardown_with_timeout_gives_up_on_wedged_closure() {
        let start = Instant::now();
        let completed = teardown_with_timeout("wedged", Duration::from_millis(300), || {
            std::thread::sleep(Duration::from_secs(60))
        });
        let elapsed = start.elapsed();
        assert!(!completed, "wedged teardown should report a timeout");
        assert!(
            elapsed >= Duration::from_millis(300),
            "should wait the full deadline, returned after {elapsed:?}"
        );
        assert!(
            elapsed < Duration::from_secs(5),
            "should give up shortly after the deadline, took {elapsed:?}"
        );
    }

    #[test]
    fn teardown_with_timeout_contains_panics() {
        let completed = teardown_with_timeout("panicking", Duration::from_secs(5), || {
            panic!("teardown panicked")
        });
        assert!(completed, "panicking teardown still counts as completed");
    }

    // Regression guard for dora-rs/dora#2742: the node's worst-case zenoh teardown must
    // stay under the daemon's force-kill grace (full rationale on `ZENOH_TEARDOWN_TIMEOUT`).
    //
    // Teardown runs in sequential bounded phases (two in `EventStream::drop`, one in
    // `DoraNode::drop`), so the worst case is `TEARDOWN_PHASES * ZENOH_TEARDOWN_TIMEOUT` —
    // bump `TEARDOWN_PHASES` if you add another. The grace is `DEFAULT_STOP_GRACE +
    // DEFAULT_STOP_GRACE/2 = 15s` (`binaries/daemon/src/running_dataflow.rs`); it lives in
    // a binary crate that can't be imported here, hence the hard-coded copy, which the
    // daemon const back-references so the two can't silently drift apart.
    #[test]
    fn zenoh_teardown_fits_within_daemon_force_kill_grace() {
        const DAEMON_FORCE_KILL_GRACE: Duration = Duration::from_secs(15);
        const TEARDOWN_PHASES: u32 = 3;
        let worst_case = ZENOH_TEARDOWN_TIMEOUT * TEARDOWN_PHASES;
        assert!(
            worst_case < DAEMON_FORCE_KILL_GRACE,
            "worst-case zenoh teardown ({worst_case:?}) must stay under the daemon \
             force-kill grace ({DAEMON_FORCE_KILL_GRACE:?}); raising ZENOH_TEARDOWN_TIMEOUT \
             reintroduces dora-rs/dora#2742"
        );
    }
}

/// Ownership invariant at the operator -> runtime boundary (dora-rs/dora#2742).
///
/// A [`SampleAllocator`] lets an operator thread encode its payload into a
/// dora-owned [`EncodedSample`] itself, so the runtime never holds a reference
/// to memory whose owner lives in another language runtime. The tests below pin
/// the properties that make the result safe to hand across a thread boundary.
#[cfg(test)]
mod operator_boundary_tests {
    use super::*;
    use arrow::buffer::Buffer;
    use std::ptr::NonNull;
    use std::sync::atomic::{AtomicBool, Ordering};

    /// Stands in for a foreign owner of an Arrow payload buffer — a numpy array
    /// held alive by pyarrow, or a buffer owned by an operator's `.so`. Flips
    /// `released` when the last Arrow reference to the buffer goes away.
    struct ForeignOwner {
        released: Arc<AtomicBool>,
        _backing: Vec<u8>,
    }

    impl Drop for ForeignOwner {
        fn drop(&mut self) {
            self.released.store(true, Ordering::SeqCst);
        }
    }

    /// A `UInt8` array whose payload buffer is owned by `ForeignOwner`.
    fn foreign_owned_array(len: usize) -> (ArrayData, Arc<AtomicBool>) {
        let backing = vec![0xABu8; len];
        // A `Vec`'s heap allocation does not move when the `Vec` itself is
        // moved into `ForeignOwner` below, so this pointer stays valid for as
        // long as the owner is alive — which is exactly what the `Allocation`
        // contract requires.
        let ptr = NonNull::new(backing.as_ptr() as *mut u8).expect("non-null");
        let released = Arc::new(AtomicBool::new(false));
        let owner = Arc::new(ForeignOwner {
            released: released.clone(),
            _backing: backing,
        });
        let buffer = unsafe { Buffer::from_custom_allocation(ptr, len, owner) };
        let array = ArrayData::builder(arrow::datatypes::DataType::UInt8)
            .len(len)
            .add_buffer(buffer)
            .build()
            .expect("valid UInt8 array");
        (array, released)
    }

    /// The heart of the #2742 fix: the sample the operator hands to the runtime
    /// must be an independent, dora-owned copy. If it kept the source buffer
    /// alive, the runtime would be the one releasing foreign memory — and for a
    /// Python operator that release takes the GIL (pyarrow's `NumPyBuffer`
    /// destructor), which stalls the runtime's event loop for as long as the
    /// operator holds it.
    #[test]
    fn encoded_sample_does_not_retain_the_source_payload() {
        let allocator = SampleAllocator::heap();
        let (array, released) = foreign_owned_array(8192);

        let sample = allocator
            .encode_arrow_data(&array)
            .expect("encoding a UInt8 array must succeed");

        assert!(
            !released.load(Ordering::SeqCst),
            "sanity: the source buffer is still alive while the array is"
        );
        drop(array);
        assert!(
            released.load(Ordering::SeqCst),
            "the encoded sample must not keep the operator's payload alive; \
             otherwise the runtime frees foreign memory (dora-rs/dora#2742)"
        );
        drop(sample);
    }

    /// The encode must be lossless: the sample is the same Arrow IPC stream the
    /// node would have produced when it did the encoding itself.
    #[test]
    fn encoded_sample_round_trips_to_the_source_array() {
        let allocator = SampleAllocator::heap();
        let (array, _released) = foreign_owned_array(1024);

        let encoded = allocator.encode_arrow_data(&array).expect("encode");
        assert_eq!(encoded.type_name(), format!("{:?}", array.data_type()));
        let decoded = crate::node::arrow_utils::decode_arrow_ipc_data(encoded.as_bytes())
            .expect("the sample must be a well-formed Arrow IPC stream");

        assert_eq!(decoded, array);
    }

    /// The allocator is handed to operator threads, so it has to cross thread
    /// boundaries — and so does the sample it produces.
    #[test]
    fn allocator_and_sample_cross_thread_boundaries() {
        const fn assert_send<T: Send>() {}
        const fn assert_send_sync_clone<T: Send + Sync + Clone>() {}
        assert_send::<DataSample>();
        assert_send::<EncodedSample>();
        assert_send_sync_clone::<SampleAllocator>();
    }
}