crates/nornir-build-info-flight/src/lib.rs
//! **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::builder::{BinaryBuilder, Int64Builder, StringBuilder};
use arrow_array::cast::AsArray;
use arrow_array::types::Int64Type;
use arrow_array::{Array, ArrayRef, RecordBatch};
use arrow_schema::{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:?}",
);
}
}