nornir 0.5.4

//! **Slice 5 — the Flight-backed `BuildInfoSink`** (design §f1/§f2/§f5).
//!
//! This closes the build-info-flight-stream arc. Slices 1–4 shipped the raw
//! Flight seam ([`nornir_flight`], edda), the transport-agnostic
//! [`BuildInfoSink`](nornir_build_thing::build_info::BuildInfoSink) trait + the
//! §f3 [`build_info`](nornir_build_thing::build_info) Arrow schema
//! ([`nornir_build_thing`]), the native warehouse landing + at-rest seal, and the
//! jera pipeline-DAG sub-jobs. This module is the ONE place that deps **both**
//! `nornir-build-thing` (the trait + rows) and `edda/nornir-flight` (the wire) —
//! no cycle — so the SAME build-info stream is:
//!
//! * **in-process** (default) — [`InMemBuildInfoSink`](nornir_build_thing::build_info::InMemBuildInfoSink),
//!   zero transport tax, single box; OR
//! * **Flight** (remote) — [`FlightBuildInfoSink`], `do_put` to a Flight endpoint,
//!
//! behind the one `BuildInfoSink` trait, exactly as the design says. The
//! in-process default is unchanged and pays nothing (this whole module is gated
//! behind the `build-info-flight` feature).
//!
//! # Why a dedicated arrow-56 columnizer
//!
//! `nornir-flight` rides **arrow-56** on purpose (its `arrow-flight` maps to
//! `tonic 0.13`, matching `nornir-service`'s interceptor; `arrow-flight 57+` needs
//! `tonic 0.14`). The build-info **`build-info-arrow`** binding is **arrow-58**
//! (the warehouse/seal universe). Those are semver-incompatible arrow majors, so
//! the Flight wire is encoded here with a small arrow-56 columnizer
//! ([`rows_to_batch`] / [`batch_to_rows`]) over the **same §f3 columns**. A test
//! asserts the column names/order/nullability are identical to build-thing's
//! canonical [`build_info_schema`](nornir_build_thing::build_info::build_info_schema),
//! so the two encodings can never silently drift.
//!
//! # The seam
//!
//! * [`FlightBuildInfoSink::connect`] — connect to a Flight endpoint (bearer
//!   optional) and hand back a `BuildInfoSink`. `emit` buffers rows to a worker
//!   that groups them by route key and `do_put`s a §f3 batch per key
//!   (best-effort — a transport error never fails the build).
//! * [`FlightBuildInfoSink::flush`] / [`flush_blocking`](FlightBuildInfoSink::flush_blocking)
//!   — drain the worker and report [`FlightStats`] (puts ok/err), so a test can
//!   assert delivery (or an auth rejection) RED-when-broken.
//! * [`route_key`] — a build-info stream is keyed by `build_id[/step_id]`, so a
//!   consumer subscribes **per sub-job `step_id`** (§f5) by ticket.
//! * [`subscribe`] — the consumer side: `do_get` a key and decode back to
//!   [`BuildInfoRow`]s byte-identical to what was emitted.

use std::collections::HashMap;
use std::sync::Arc;

use anyhow::{anyhow, Result};
use arrow_array_56::builder::{BinaryBuilder, Int64Builder, StringBuilder};
use arrow_array_56::cast::AsArray;
use arrow_array_56::types::Int64Type;
use arrow_array_56::{Array, ArrayRef, RecordBatch};
use arrow_schema_56::{DataType, Field, Schema};
use nornir_build_thing::build_info::{BuildInfoRow, BuildInfoSink, BUILD_INFO_SCHEMA};
use nornir_flight::FlightConn;
use tokio::sync::{mpsc, oneshot};

/// The route key for a build-info row on the Flight wire: `build_id` for root /
/// manifest rows (empty `step_id`), `build_id/step_id` for a sub-job's feed.
///
/// This is the ticket a consumer subscribes on — one per sub-job `step_id`, so
/// the viz live-tail can pull exactly one step's `log`/`metric`/`step` feed
/// (§f5: "each step's log/metric stream is the f3 feed filtered by `step_id`").
pub fn route_key(build_id: &str, step_id: &str) -> String {
    if step_id.is_empty() {
        build_id.to_string()
    } else {
        format!("{build_id}/{step_id}")
    }
}

/// The §f3 build-info Arrow schema, in **arrow-56** (the Flight wire's arrow).
///
/// Column-identical to build-thing's canonical
/// [`build_info_schema`](nornir_build_thing::build_info::build_info_schema)
/// (arrow-58): the nine §f3 columns in declared order, `blob` the only nullable
/// column, and the `schema` metadata key set to [`BUILD_INFO_SCHEMA`]. A test
/// pins the two shapes equal so they cannot drift across the version boundary.
pub fn build_info_flight_schema() -> Schema {
    let fields = vec![
        Field::new("build_id", DataType::Utf8, false),
        Field::new("step_id", DataType::Utf8, false),
        Field::new("ts_micros", DataType::Int64, false),
        Field::new("kind", DataType::Utf8, false),
        Field::new("level", DataType::Utf8, false),
        Field::new("name", DataType::Utf8, false),
        Field::new("value", DataType::Utf8, false),
        Field::new("content_id", DataType::Utf8, false),
        Field::new("blob", DataType::Binary, true),
    ];
    let mut meta = HashMap::new();
    meta.insert("schema".to_string(), BUILD_INFO_SCHEMA.to_string());
    Schema::new(fields).with_metadata(meta)
}

/// Columnize `rows` into one arrow-56 [`RecordBatch`] matching
/// [`build_info_flight_schema`] — the on-the-wire encoding for `do_put`.
pub fn rows_to_batch(rows: &[BuildInfoRow]) -> Result<RecordBatch> {
    let n = rows.len();
    let mut build_id = StringBuilder::with_capacity(n, n * 40);
    let mut step_id = StringBuilder::with_capacity(n, n * 16);
    let mut ts = Int64Builder::with_capacity(n);
    let mut kind = StringBuilder::with_capacity(n, n * 10);
    let mut level = StringBuilder::with_capacity(n, n * 8);
    let mut name = StringBuilder::with_capacity(n, n * 16);
    let mut value = StringBuilder::with_capacity(n, n * 32);
    let mut content_id = StringBuilder::with_capacity(n, n * 64);
    let mut blob = BinaryBuilder::with_capacity(n, n * 32);

    for r in rows {
        build_id.append_value(&r.build_id);
        step_id.append_value(&r.step_id);
        ts.append_value(r.ts_micros);
        kind.append_value(&r.kind);
        level.append_value(&r.level);
        name.append_value(&r.name);
        value.append_value(&r.value);
        content_id.append_value(&r.content_id);
        match &r.blob {
            Some(b) => blob.append_value(b),
            None => blob.append_null(),
        }
    }

    let columns: Vec<ArrayRef> = vec![
        Arc::new(build_id.finish()),
        Arc::new(step_id.finish()),
        Arc::new(ts.finish()),
        Arc::new(kind.finish()),
        Arc::new(level.finish()),
        Arc::new(name.finish()),
        Arc::new(value.finish()),
        Arc::new(content_id.finish()),
        Arc::new(blob.finish()),
    ];

    RecordBatch::try_new(Arc::new(build_info_flight_schema()), columns)
        .map_err(|e| anyhow!("build_info flight batch: {e}"))
}

/// Decode an arrow-56 §f3 [`RecordBatch`] back into [`BuildInfoRow`]s — the
/// consumer side of [`rows_to_batch`]. Round-trips byte-identical (including a
/// NULL `blob` vs. an empty one).
pub fn batch_to_rows(batch: &RecordBatch) -> Result<Vec<BuildInfoRow>> {
    if batch.num_columns() != 9 {
        return Err(anyhow!(
            "build_info batch has {} columns, expected 9",
            batch.num_columns()
        ));
    }
    let build_id = batch.column(0).as_string::<i32>();
    let step_id = batch.column(1).as_string::<i32>();
    let ts = batch.column(2).as_primitive::<Int64Type>();
    let kind = batch.column(3).as_string::<i32>();
    let level = batch.column(4).as_string::<i32>();
    let name = batch.column(5).as_string::<i32>();
    let value = batch.column(6).as_string::<i32>();
    let content_id = batch.column(7).as_string::<i32>();
    let blob = batch.column(8).as_binary::<i32>();

    let mut out = Vec::with_capacity(batch.num_rows());
    for i in 0..batch.num_rows() {
        out.push(BuildInfoRow {
            build_id: build_id.value(i).to_string(),
            step_id: step_id.value(i).to_string(),
            ts_micros: ts.value(i),
            kind: kind.value(i).to_string(),
            level: level.value(i).to_string(),
            name: name.value(i).to_string(),
            value: value.value(i).to_string(),
            content_id: content_id.value(i).to_string(),
            blob: if blob.is_null(i) {
                None
            } else {
                Some(blob.value(i).to_vec())
            },
        });
    }
    Ok(out)
}

/// Delivery outcome reported by [`FlightBuildInfoSink::flush`]: how many keyed
/// `do_put`s succeeded / failed, and the last error seen. A caller (or test)
/// asserts `puts_ok > 0` for a green stream, or `puts_err > 0` for an auth /
/// transport rejection — RED-when-broken either way.
#[derive(Debug, Clone, Default)]
pub struct FlightStats {
    /// Keyed `do_put`s that were acknowledged by the server.
    pub puts_ok: usize,
    /// Keyed `do_put`s that failed (transport / auth / encode).
    pub puts_err: usize,
    /// The most recent error message, if any.
    pub last_error: Option<String>,
}

/// Worker mailbox message.
enum Msg {
    /// A batch of rows to ship (grouped by route key inside the worker).
    Rows(Vec<BuildInfoRow>),
    /// Drain barrier: reply with the cumulative stats once all prior `Rows` are
    /// processed (FIFO ordering guarantees this).
    Flush(oneshot::Sender<FlightStats>),
}

/// Group rows by [`route_key`], preserving first-seen key order and the row
/// order within each key (so the stream a consumer replays is the stream that
/// was emitted).
fn group_rows(rows: &[BuildInfoRow]) -> Vec<(String, Vec<BuildInfoRow>)> {
    let mut order: Vec<String> = Vec::new();
    let mut map: HashMap<String, Vec<BuildInfoRow>> = HashMap::new();
    for r in rows {
        let k = route_key(&r.build_id, &r.step_id);
        if !map.contains_key(&k) {
            order.push(k.clone());
        }
        map.entry(k).or_default().push(r.clone());
    }
    order
        .into_iter()
        .map(|k| {
            let v = map.remove(&k).expect("key just inserted");
            (k, v)
        })
        .collect()
}

/// The worker loop: owns the [`FlightConn`], ships each keyed group over
/// `do_put`, and answers flush barriers. Runs on the sink's owned runtime.
async fn worker(mut conn: FlightConn, mut rx: mpsc::UnboundedReceiver<Msg>) {
    let mut stats = FlightStats::default();
    while let Some(msg) = rx.recv().await {
        match msg {
            Msg::Rows(rows) => {
                for (key, group) in group_rows(&rows) {
                    match rows_to_batch(&group) {
                        Ok(batch) => match conn.put(&key, vec![batch]).await {
                            Ok(()) => {
                                stats.puts_ok += 1;
                                emit_status(
                                    "do_put", true, &key, group.len(),
                                );
                            }
                            Err(e) => {
                                stats.puts_err += 1;
                                stats.last_error = Some(e.to_string());
                                emit_status(
                                    "do_put", false, &key, group.len(),
                                );
                            }
                        },
                        Err(e) => {
                            stats.puts_err += 1;
                            stats.last_error = Some(e.to_string());
                        }
                    }
                }
            }
            Msg::Flush(reply) => {
                let _ = reply.send(stats.clone());
            }
        }
    }
}

/// A **Flight-backed** [`BuildInfoSink`]: serializes build-info rows to the §f3
/// Arrow `RecordBatch` and ships them via `do_put` to an `edda/nornir-flight`
/// endpoint. The remote counterpart of the in-process
/// [`InMemBuildInfoSink`](nornir_build_thing::build_info::InMemBuildInfoSink),
/// behind the one `BuildInfoSink` trait.
///
/// Hosts its tokio runtime on a **dedicated OS thread** (created and dropped
/// there), so [`emit`](BuildInfoSink::emit) (a sync trait method, called from
/// sync build code) never blocks and never needs an ambient runtime — and,
/// crucially, dropping the sink from within an async context never trips tokio's
/// "cannot drop a runtime in an async context" guard (the runtime lives and dies
/// on its own thread). The sink holds only the channel; dropping it closes the
/// channel, the worker loop ends, and the worker thread winds down its runtime.
pub struct FlightBuildInfoSink {
    tx: mpsc::UnboundedSender<Msg>,
}

impl FlightBuildInfoSink {
    /// Connect to a Flight `endpoint` (e.g. `http://127.0.0.1:50051`), optionally
    /// carrying a bearer `token` (to satisfy the shared `nornir-service` auth
    /// interceptor; `None` / empty ⇒ open).
    ///
    /// Safe to call from **either** a sync context (a build's xtask) or inside an
    /// async runtime (a test): the connect + all `do_put`s run on a dedicated
    /// thread's own runtime, so there is never a runtime nested in the caller's.
    pub fn connect(endpoint: impl Into<String>, token: Option<String>) -> Result<Self> {
        let endpoint = endpoint.into();
        let (tx, rx) = mpsc::unbounded_channel();
        // Startup handshake result, delivered over a std channel (blocking recv on
        // the caller — the connect is quick, and this works in any context).
        let (ready_tx, ready_rx) = std::sync::mpsc::channel::<Result<()>>();

        std::thread::Builder::new()
            .name("build-info-flight".into())
            .spawn(move || {
                let rt = match tokio::runtime::Builder::new_current_thread()
                    .enable_all()
                    .build()
                {
                    Ok(rt) => rt,
                    Err(e) => {
                        let _ = ready_tx.send(Err(anyhow!("flight sink runtime: {e}")));
                        return;
                    }
                };
                rt.block_on(async move {
                    let conn = match FlightConn::connect(endpoint).await {
                        Ok(c) => match token {
                            Some(t) if !t.is_empty() => match c.with_bearer(&t) {
                                Ok(c) => c,
                                Err(e) => {
                                    let _ = ready_tx.send(Err(anyhow!("flight bearer: {e}")));
                                    return;
                                }
                            },
                            _ => c,
                        },
                        Err(e) => {
                            let _ = ready_tx.send(Err(anyhow!("flight connect: {e}")));
                            return;
                        }
                    };
                    let _ = ready_tx.send(Ok(()));
                    worker(conn, rx).await;
                });
                // `rt` drops here, on this thread — never in the caller's async ctx.
            })
            .map_err(|e| anyhow!("flight sink thread: {e}"))?;

        ready_rx
            .recv()
            .map_err(|_| anyhow!("flight sink connect thread died"))??;
        Ok(Self { tx })
    }

    /// Drain the worker and report [`FlightStats`]. Awaitable — call from an
    /// async context (`sink.flush().await`). Resolves once every row emitted
    /// before this call has been `do_put` (a FIFO barrier).
    pub async fn flush(&self) -> Result<FlightStats> {
        let reply_rx = self.send_flush()?;
        reply_rx
            .await
            .map_err(|_| anyhow!("flight sink flush dropped"))
    }

    /// Blocking [`flush`](Self::flush) for sync callers (a build xtask). Safe from
    /// any context — it drives the reply future with a plain executor, not a
    /// nested tokio runtime.
    pub fn flush_blocking(&self) -> Result<FlightStats> {
        let reply_rx = self.send_flush()?;
        futures::executor::block_on(async move {
            reply_rx
                .await
                .map_err(|_| anyhow!("flight sink flush dropped"))
        })
    }

    /// Send the flush barrier, returning the reply receiver.
    fn send_flush(&self) -> Result<oneshot::Receiver<FlightStats>> {
        let (reply_tx, reply_rx) = oneshot::channel();
        self.tx
            .send(Msg::Flush(reply_tx))
            .map_err(|_| anyhow!("flight sink worker gone"))?;
        Ok(reply_rx)
    }
}

impl BuildInfoSink for FlightBuildInfoSink {
    /// Best-effort: buffer the rows to the worker for `do_put`. Never fails the
    /// build (the trait contract) — a transport error is surfaced via
    /// [`FlightStats`] on [`flush`](FlightBuildInfoSink::flush), not here.
    fn emit(&self, rows: &[BuildInfoRow]) -> Result<()> {
        let _ = self.tx.send(Msg::Rows(rows.to_vec()));
        Ok(())
    }
}

/// The **consumer** side (§f5): subscribe to a build-info stream by ticket `key`
/// (from [`route_key`]) over `do_get`, decoding the arrow-56 batches back into
/// [`BuildInfoRow`]s byte-identical to what was emitted. `token` bearer optional.
///
/// This is what the viz bench-live / jobview live-tail pane calls per sub-job
/// `step_id` to tail one step's feed.
pub async fn subscribe(
    endpoint: &str,
    token: Option<&str>,
    key: &str,
) -> Result<Vec<BuildInfoRow>> {
    let mut conn = FlightConn::connect(endpoint.to_string())
        .await
        .map_err(|e| anyhow!("flight connect: {e}"))?;
    if let Some(t) = token {
        if !t.is_empty() {
            conn = conn.with_bearer(t).map_err(|e| anyhow!("flight bearer: {e}"))?;
        }
    }
    let batches = conn
        .get(key)
        .await
        .map_err(|e| anyhow!("do_get {key:?}: {e}"))?;
    let mut rows = Vec::new();
    for b in &batches {
        rows.extend(batch_to_rows(b)?);
    }
    emit_status("do_get", true, key, rows.len());
    Ok(rows)
}

/// Emit a functional-status marker into the nornir test-matrix (L4). Behind the
/// `testmatrix` feature; a zero-cost no-op otherwise (release strips it).
#[cfg(feature = "testmatrix")]
#[inline]
pub(crate) fn emit_status(verb: &str, ok: bool, key: &str, rows: usize) {
    nornir_testmatrix::functional_status(
        "nornir-build-info-flight",
        verb,
        ok,
        &format!("key={key:?} rows={rows}"),
    );
}

/// No-op when `testmatrix` is off.
#[cfg(not(feature = "testmatrix"))]
#[inline]
pub(crate) fn emit_status(_verb: &str, _ok: bool, _key: &str, _rows: usize) {}

#[cfg(test)]
mod tests {
    use super::*;
    use std::net::{SocketAddr, TcpListener};
    use std::time::Duration;

    use nornir_build_thing::build_info::{
        build_feed_rows, BuildInfoKind, BuildInfoRow,
    };
    use nornir_flight::{FlightRoutes, FlightServer, MemFlightStore};

    /// An OS-assigned free port (bind, read, drop — the server rebinds it).
    fn free_addr() -> SocketAddr {
        TcpListener::bind("127.0.0.1:0")
            .unwrap()
            .local_addr()
            .unwrap()
    }

    /// Start a Flight server (fresh `MemFlightStore`) with `token` auth (empty ⇒
    /// open), returning its `http://…` endpoint once it accepts connections.
    async fn start_server(token: &str) -> String {
        let addr = free_addr();
        let routes = FlightRoutes::shared(Arc::new(MemFlightStore::new()));
        let tok = Arc::new(token.to_string());
        tokio::spawn(async move {
            let _ = FlightServer::serve_with_auth(routes, addr, tok).await;
        });
        for _ in 0..100 {
            if tokio::net::TcpStream::connect(addr).await.is_ok() {
                break;
            }
            tokio::time::sleep(Duration::from_millis(20)).await;
        }
        format!("http://{addr}")
    }

    /// A small mixed build-info stream: manifest rows (empty `step_id`) plus one
    /// sub-job's feed (`step_id = "cargo-build"`), including a NULL and a present
    /// `blob` so the nullable column is exercised.
    fn sample_stream(build_id: &str) -> Vec<BuildInfoRow> {
        let mut rows = vec![
            BuildInfoRow::new(build_id, 1, BuildInfoKind::Provenance).kv("repo", "holger"),
            BuildInfoRow::new(build_id, 2, BuildInfoKind::Dep).kv("feature", "full"),
            BuildInfoRow::new(build_id, 3, BuildInfoKind::Artifact)
                .kv("target/release/holger", "1024")
                .content_id("deadbeef"),
        ];
        rows.extend(build_feed_rows(build_id, 4, "full", true, 1, 12, 999));
        // A log row carrying an inline blob (the only nullable column).
        rows.push(
            BuildInfoRow::new(build_id, 5, BuildInfoKind::Log)
                .step("cargo-build")
                .level("info")
                .kv("chunk", "line")
                .blob(b"raw log bytes".to_vec()),
        );
        rows
    }

    /// RED-when-broken: the arrow-56 Flight schema is the canonical §f3 shape —
    /// the nine columns in declared order, `blob` the only nullable column, the
    /// `schema` metadata key set. Asserted against the literal §f3 contract so
    /// the guard runs with just `build-info-flight` (no arrow-58 needed); when
    /// `build-info-arrow` is also on, it additionally pins equality with
    /// build-thing's canonical arrow-58 `build_info_schema` so the two encodings
    /// cannot silently drift across the version boundary.
    #[test]
    fn flight_schema_matches_canonical_f3_columns() {
        let flight = build_info_flight_schema();

        let f_names: Vec<&str> = flight.fields().iter().map(|f| f.name().as_str()).collect();
        assert_eq!(
            f_names,
            vec![
                "build_id", "step_id", "ts_micros", "kind", "level", "name", "value",
                "content_id", "blob",
            ],
            "§f3 column names/order must match",
        );

        // blob (col 8) is the ONLY nullable column.
        assert!(flight.field(8).is_nullable());
        assert!(flight.fields().iter().take(8).all(|f| !f.is_nullable()));

        assert_eq!(
            flight.metadata().get("schema").map(String::as_str),
            Some(BUILD_INFO_SCHEMA),
        );

        // When the arrow-58 canonical schema is compiled in, pin the names equal.
        #[cfg(feature = "build-info-arrow")]
        {
            let canonical = nornir_build_thing::build_info::build_info_schema();
            let c_names: Vec<&str> =
                canonical.fields().iter().map(|f| f.name().as_str()).collect();
            assert_eq!(f_names, c_names, "arrow-56 wire vs arrow-58 canonical drift");
        }
    }

    /// RED-when-broken: rows → arrow-56 batch → rows is byte-identical, incl. the
    /// NULL-vs-present `blob` distinction.
    #[test]
    fn batch_round_trips_rows_byte_identical() {
        let rows = sample_stream("b-local");
        let batch = rows_to_batch(&rows).unwrap();
        let back = batch_to_rows(&batch).unwrap();
        assert_eq!(back, rows, "arrow-56 columnize/decode must be lossless");
        // The blob-bearing row survived as Some; a manifest row stayed None.
        assert!(back.iter().any(|r| r.blob.as_deref() == Some(b"raw log bytes")));
        assert!(back.iter().any(|r| r.kind == "provenance" && r.blob.is_none()));
    }

    /// The headline round-trip (design §f1 acceptance, slice-5 flavour): a
    /// `FlightBuildInfoSink` ships a mixed stream over `do_put`; a consumer
    /// subscribes **per route key** via `do_get` and receives the rows
    /// byte-identical to what was emitted. Auth via the nornir-service bearer.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn flight_sink_ships_and_consumer_receives_byte_identical() {
        let token = "a-long-enough-secret-token-1234";
        let ep = start_server(token).await;

        let build_id = "b-remote";
        let rows = sample_stream(build_id);

        let sink = FlightBuildInfoSink::connect(&ep, Some(token.to_string())).unwrap();
        // The sink is a plain `BuildInfoSink` — the emitter deals only in rows.
        let dyn_sink: &dyn BuildInfoSink = &sink;
        dyn_sink.emit(&rows).unwrap();

        let stats = sink.flush().await.unwrap();
        assert!(
            stats.puts_ok >= 2 && stats.puts_err == 0,
            "expected the manifest + feed keys delivered, got {stats:?}",
        );

        // Consumer: subscribe per route key. Manifest rows (empty step_id) ride
        // key = build_id; the feed rides key = build_id/cargo-build.
        let manifest_key = route_key(build_id, "");
        let step_key = route_key(build_id, "cargo-build");

        let want_manifest: Vec<BuildInfoRow> =
            rows.iter().filter(|r| r.step_id.is_empty()).cloned().collect();
        let want_step: Vec<BuildInfoRow> =
            rows.iter().filter(|r| r.step_id == "cargo-build").cloned().collect();

        let got_manifest = subscribe(&ep, Some(token), &manifest_key).await.unwrap();
        let got_step = subscribe(&ep, Some(token), &step_key).await.unwrap();

        // RED-when-broken: received == emitted, per subscribed step ticket.
        assert_eq!(got_manifest, want_manifest, "manifest stream must round-trip");
        assert_eq!(got_step, want_step, "per-step_id feed must round-trip");

        // And the whole stream is recovered by concatenating the tickets.
        let mut all = got_manifest;
        all.extend(got_step);
        assert_eq!(all.len(), rows.len(), "every emitted row is reachable by ticket");
    }

    /// RED-when-broken auth: a sink with **no** bearer against an auth server has
    /// its `do_put`s rejected (stats show errors, nothing lands), and an
    /// unauthenticated consumer is likewise refused. Proves the shared
    /// nornir-service interceptor is actually enforced on the build-info wire.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn auth_is_enforced_on_the_build_info_wire() {
        let token = "a-long-enough-secret-token-1234";
        let ep = start_server(token).await;
        let rows = sample_stream("b-noauth");

        // No token → every keyed do_put is rejected.
        let sink = FlightBuildInfoSink::connect(&ep, None).unwrap();
        sink.emit(&rows).unwrap();
        let stats = sink.flush().await.unwrap();
        assert!(
            stats.puts_ok == 0 && stats.puts_err > 0,
            "unauthenticated do_put MUST be rejected, got {stats:?}",
        );

        // No token → consumer refused too.
        let err = subscribe(&ep, None, &route_key("b-noauth", ""))
            .await
            .expect_err("unauthenticated do_get MUST be rejected");
        let msg = err.to_string().to_lowercase();
        assert!(
            msg.contains("bearer") || msg.contains("unauthenticated") || msg.contains("no stream"),
            "expected an auth rejection, got: {msg}",
        );
    }

    /// The `testmatrix` mode records honest functional rows for the Flight
    /// build-info surfaces (do_put on the sink, do_get on the consumer).
    #[cfg(feature = "testmatrix")]
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn emits_functional_rows_for_flight_build_info() {
        let ep = start_server("").await;
        let rows = sample_stream("b-matrix");
        let sink = FlightBuildInfoSink::connect(&ep, None).unwrap();
        sink.emit(&rows).unwrap();
        let _ = sink.flush().await.unwrap();
        let _ = subscribe(&ep, None, &route_key("b-matrix", "")).await.unwrap();

        let drained = nornir_testmatrix::drain_functional_rows();
        assert!(
            drained.iter().any(|r| r.suite == "nornir-build-info-flight"
                && r.test_name == "do_put"
                && r.status == "pass"),
            "missing green do_put matrix row: {drained:?}",
        );
        assert!(
            drained.iter().any(|r| r.suite == "nornir-build-info-flight"
                && r.test_name == "do_get"
                && r.status == "pass"),
            "missing green do_get matrix row: {drained:?}",
        );
    }
}