src/warehouse/remote.rs
//! **RemoteWarehouse** — a [`Warehouse`](super::Warehouse) trait impl backed by a
//! running `nornir-server`'s generic warehouse gRPC (`Warehouse.Tables` /
//! `Warehouse.Scan`, package `nornir.v1`), NOT a local Iceberg tree.
//!
//! This is the second concrete backend behind the durable [`Warehouse`] seam
//! (the first is [`iceberg::IcebergWarehouse`](super::iceberg::IcebergWarehouse)).
//! Its whole reason to exist is to **prove the trait is a real seam**: a thin
//! client that owns no redb catalog and no Parquet, yet satisfies the same trait
//! the ~150 sync call sites drive — so a future `nornir` running purely as a
//! thin client can read the warehouse a server owns without any call site
//! knowing it is not local.
//!
//! ## What is wired (Phase 3 skeleton)
//!
//! The server holds the single-writer redb lock, so a thin client is a
//! **reader**. The read surface that the `Warehouse.Tables`/`Warehouse.Scan`
//! RPCs already expose maps 1:1 onto the trait's *browser* methods:
//!
//! | trait method | transport |
//! |-------------------------|---------------------------------------------|
//! | [`table_names`] | `Warehouse.Tables` RPC |
//! | [`scan_preview`] | `Warehouse.Scan` RPC (stringified rows) |
//! | [`scan_limited`] | `Warehouse.Scan` (limit) → all-`Utf8` batch |
//! | [`scan_arrow`] | `Warehouse.Scan` (server default) → batch |
//!
//! [`table_names`]: super::Warehouse::table_names
//! [`scan_preview`]: super::Warehouse::scan_preview
//! [`scan_limited`]: super::Warehouse::scan_limited
//! [`scan_arrow`]: super::Warehouse::scan_arrow
//!
//! The `Warehouse.Scan` RPC returns *stringified* cells (the display preview the
//! egui browser renders), so the reconstructed [`RecordBatch`] is honestly
//! **all-`Utf8`** — types are lost on the wire. A typed remote read (Arrow
//! Flight, or a per-table BenchRuns/… RPC) is the documented follow-up; until it
//! lands, [`scan_filtered`](super::Warehouse::scan_filtered) (needs real
//! predicate pushdown) and [`query_bench_runs`](super::Warehouse::query_bench_runs)
//! (needs the typed row) return a typed [`RemoteWarehouseError::Unsupported`].
//!
//! ## What is NOT wired (typed errors, by design)
//!
//! Every **write** ([`append_arrow`](super::Warehouse::append_arrow),
//! [`ensure_table`](super::Warehouse::ensure_table),
//! [`ensure_columns`](super::Warehouse::ensure_columns),
//! [`append_bench_run`](super::Warehouse::append_bench_run)) returns
//! [`RemoteWarehouseError::Unsupported`]: a thin client never opens the warehouse
//! to write it — it *submits* rows via the `Telemetry.*` RPCs and the server
//! appends them (the contract in `.nornir/data-submit-rpcs.md`). These are typed
//! errors with a clear TODO, never silent no-ops (L2).
use std::sync::Arc;
use anyhow::Result;
use arrow::array::{ArrayRef, RecordBatch, StringArray};
use arrow::datatypes::{DataType, Field, Schema as ArrowSchema};
use uuid::Uuid;
use super::iceberg::TablePreview;
use super::{BenchFilter, ScanFilter, Warehouse};
use crate::bench::BenchRun;
/// The generated `nornir.v1` prost/tonic bindings, sourced from the shared
/// `nornir-wire` edda leaf (`nornir_wire::pb`) instead of a private
/// `include_proto!` copy. The `warehouse_client` half is present whenever the
/// leaf's `client` feature is on (default), the same `server`/`viz`/`mcp` gate
/// that compiles this module.
use nornir_wire::pb;
/// Why a [`RemoteWarehouse`] method is not (yet) serviceable — a **typed** error
/// so callers can match on it instead of string-sniffing an `anyhow` chain.
#[derive(Debug, Clone, thiserror::Error)]
pub enum RemoteWarehouseError {
/// The method is not wired on the remote backend yet. `todo` names the
/// concrete follow-up (a new RPC, Arrow Flight, or the write-submit path).
#[error("RemoteWarehouse::{method} is not wired on the thin client yet — {todo}")]
Unsupported {
/// The trait method name (e.g. `"append_arrow"`).
method: &'static str,
/// The concrete follow-up that would wire it.
todo: &'static str,
},
}
fn unsupported(method: &'static str, todo: &'static str) -> anyhow::Error {
RemoteWarehouseError::Unsupported { method, todo }.into()
}
/// A thin `Warehouse` client that reads a server-owned warehouse over gRPC.
///
/// Holds a dedicated current-thread tokio runtime so the **sync** trait methods
/// can `block_on` the async RPC without a "runtime within a runtime" panic — the
/// exact shape [`IcebergWarehouse`](super::iceberg::IcebergWarehouse) uses for
/// its own runtime, so the ~150 sync call sites stay sync no matter which backend
/// is behind the trait object.
pub struct RemoteWarehouse {
endpoint: String,
token: String,
workspace: String,
rt: tokio::runtime::Runtime,
}
impl RemoteWarehouse {
/// Connect a thin warehouse client to `endpoint` (an `http://host:port`, or a
/// bare `host:port` that is `http://`-prefixed). `token` is the bearer auth
/// (empty ⇒ no real auth); `workspace` selects the served workspace (empty ⇒
/// the server default) via the `nornir-workspace` metadata header. Building
/// the runtime is the only fallible step — the channel is dialed lazily per
/// RPC (matching `viz::remote`), so `connect` never blocks on the network.
pub fn connect(
endpoint: impl Into<String>,
token: impl Into<String>,
workspace: impl Into<String>,
) -> Result<Self> {
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.map_err(|e| anyhow::anyhow!("build tokio runtime for RemoteWarehouse: {e}"))?;
Ok(Self {
endpoint: normalize_endpoint(&endpoint.into()),
token: token.into(),
workspace: workspace.into(),
rt,
})
}
/// The configured endpoint (normalized, `http://`-prefixed).
pub fn endpoint(&self) -> &str {
&self.endpoint
}
// ── the two networked reads: Warehouse.Tables / Warehouse.Scan ───────────
fn tables_rpc(&self) -> Result<pb::WarehouseTables> {
let (endpoint, token, workspace) = (
self.endpoint.clone(),
self.token.clone(),
self.workspace.clone(),
);
self.rt.block_on(async move {
let mut client = connect_client(&endpoint, &token, &workspace).await?;
Ok(client
.tables(pb::Empty {})
.await
.map_err(|s| anyhow::anyhow!("Warehouse.Tables RPC: {s}"))?
.into_inner())
})
}
fn scan_rpc(&self, table: &str, limit: u32) -> Result<pb::WarehouseScan> {
let (endpoint, token, workspace, table) = (
self.endpoint.clone(),
self.token.clone(),
self.workspace.clone(),
table.to_string(),
);
self.rt.block_on(async move {
let mut client = connect_client(&endpoint, &token, &workspace).await?;
Ok(client
.scan(pb::WarehouseScanRequest { table, limit })
.await
.map_err(|s| anyhow::anyhow!("Warehouse.Scan RPC: {s}"))?
.into_inner())
})
}
/// Drain the `Warehouse.ScanVectors` server stream into a typed
/// [`nornir_warehouse_trait::VectorScan`].
///
/// The two states a caller must never conflate are kept apart HERE: an
/// `UNIMPLEMENTED` status becomes the typed [`RemoteWarehouseError::Unsupported`]
/// ("this transport cannot carry vectors" — a pre-`ScanVectors` server), while a
/// stream that completes carrying nothing is `Ok` with zero vectors ("the warehouse
/// holds 0 embeddings").
fn scan_vectors_rpc(
&self,
table: &str,
key_col: &str,
vec_col: &str,
limit: u32,
) -> Result<nornir_warehouse_trait::VectorScan> {
use nornir_warehouse_trait::{VectorRow, VectorScan};
let (endpoint, token, workspace) = (
self.endpoint.clone(),
self.token.clone(),
self.workspace.clone(),
);
let req = pb::WarehouseVectorRequest {
table: table.to_string(),
key_col: key_col.to_string(),
vec_col: vec_col.to_string(),
limit,
};
self.rt.block_on(async move {
let mut client = connect_client(&endpoint, &token, &workspace).await?;
let mut stream = client
.scan_vectors(req)
.await
.map_err(|s| {
if s.code() == tonic::Code::Unimplemented {
unsupported(
"scan_vectors",
"the server does not serve Warehouse.ScanVectors (pre-typed-vector \
build) — vectors CANNOT cross this transport, which is not the same \
as the warehouse holding none",
)
} else {
anyhow::anyhow!("Warehouse.ScanVectors RPC: {s}")
}
})?
.into_inner();
let mut out = VectorScan::default();
while let Some(batch) = stream
.message()
.await
.map_err(|s| anyhow::anyhow!("Warehouse.ScanVectors stream: {s}"))?
{
out.rows_scanned = out.rows_scanned.max(batch.rows_scanned);
out.clipped |= batch.clipped;
for v in batch.vectors {
out.vectors.push(VectorRow {
key: v.key,
vector: le_f32(&v.values, v.dims)?,
model_profile: v.model_profile,
});
}
}
Ok(out)
})
}
}
/// Decode `dims` little-endian IEEE-754 `f32`s out of a wire payload.
///
/// A length that does not divide evenly is an ERROR, never a truncated vector: a short
/// vector still computes a plausible cosine, which is exactly the confidently-wrong
/// answer this whole typed route exists to prevent.
pub fn le_f32(bytes: &[u8], dims: u32) -> Result<Vec<f32>> {
let dims = dims as usize;
if bytes.len() != dims * 4 {
anyhow::bail!(
"ScanVectors: declared dims={dims} needs {} bytes, got {} — refusing to \
truncate (a short vector yields a plausible-looking cosine)",
dims * 4,
bytes.len()
);
}
Ok(bytes
.chunks_exact(4)
.map(|c| f32::from_le_bytes([c[0], c[1], c[2], c[3]]))
.collect())
}
/// Encode `f32`s as the wire's little-endian payload. The ONE encoder — the server
/// writes with it and [`le_f32`] reads it back, so the two halves cannot drift.
pub fn f32_le_bytes(v: &[f32]) -> Vec<u8> {
let mut out = Vec::with_capacity(v.len() * 4);
for f in v {
out.extend_from_slice(&f.to_le_bytes());
}
out
}
// ── pure, network-free mappers (unit-testable without a server) ─────────────
/// Map a `Warehouse.Tables` response to the trait's sorted table-name list.
pub fn tables_to_names(t: pb::WarehouseTables) -> Vec<String> {
let mut names = t.names;
names.sort();
names
}
/// Map a `Warehouse.Scan` response (columns + stringified rows) to a
/// [`TablePreview`] — the exact struct `IcebergWarehouse::scan_preview` returns,
/// so the generic warehouse browser is source-agnostic.
pub fn scan_to_preview(s: pb::WarehouseScan) -> TablePreview {
TablePreview {
columns: s.columns,
rows: s.rows.into_iter().map(|r| r.cells).collect(),
}
}
/// Reconstruct an **all-`Utf8`** [`RecordBatch`] from a stringified preview. Types
/// were lost on the `Warehouse.Scan` wire, so every column is `Utf8` (honest —
/// a typed remote read is the Flight follow-up). Empty columns ⇒ no batch.
pub fn preview_to_string_batch(p: &TablePreview) -> Result<Vec<RecordBatch>> {
if p.columns.is_empty() {
return Ok(Vec::new());
}
let fields: Vec<Field> = p
.columns
.iter()
.map(|c| Field::new(c, DataType::Utf8, true))
.collect();
let schema = Arc::new(ArrowSchema::new(fields));
let ncols = p.columns.len();
let cols: Vec<ArrayRef> = (0..ncols)
.map(|ci| {
let vals: Vec<Option<String>> = p.rows.iter().map(|r| r.get(ci).cloned()).collect();
Arc::new(StringArray::from(vals)) as ArrayRef
})
.collect();
Ok(vec![RecordBatch::try_new(schema, cols)?])
}
fn normalize_endpoint(endpoint: &str) -> String {
if endpoint.starts_with("http") {
endpoint.to_string()
} else {
format!("http://{endpoint}")
}
}
/// Dial a `WarehouseClient` with the shared bearer + `nornir-workspace` header
/// interceptor (the same connect/auth shape as `viz::remote`).
async fn connect_client(
endpoint: &str,
token: &str,
workspace: &str,
) -> Result<
pb::warehouse_client::WarehouseClient<
tonic::service::interceptor::InterceptedService<
tonic::transport::Channel,
impl FnMut(tonic::Request<()>) -> Result<tonic::Request<()>, tonic::Status>,
>,
>,
> {
let channel = tonic::transport::Channel::from_shared(endpoint.to_string())
.map_err(|e| anyhow::anyhow!("invalid server url `{endpoint}`: {e}"))?
.connect()
.await
.map_err(|e| anyhow::anyhow!("connect to nornir-server at {endpoint}: {e}"))?;
// The ONE shared client bearer-builder (Law #1) — no hand-rolled twin. Empty
// token OR an unencodable one ⇒ `None` (open); preserves this site's prior
// error-swallowing shape (`…parse().ok()`), which never failed the connect.
let bearer = nornir_service::client::bearer(token).ok().flatten();
let ws: Option<tonic::metadata::MetadataValue<tonic::metadata::Ascii>> = (!workspace
.is_empty())
.then(|| workspace.parse().ok())
.flatten();
Ok(pb::warehouse_client::WarehouseClient::with_interceptor(
channel,
move |mut req: tonic::Request<()>| {
if let Some(b) = &bearer {
req.metadata_mut().insert("authorization", b.clone());
}
if let Some(w) = &ws {
req.metadata_mut().insert("nornir-workspace", w.clone());
}
Ok(req)
},
))
}
#[async_trait::async_trait]
impl Warehouse for RemoteWarehouse {
// ── generic Arrow core ──────────────────────────────────────────────────
fn append_arrow(&self, _table: &str, _batch: RecordBatch) -> Result<()> {
Err(unsupported(
"append_arrow",
"the server is the single writer; submit rows via the Telemetry.* RPCs \
(see .nornir/data-submit-rpcs.md), not the trait write path",
))
}
fn scan_arrow(&self, table: &str) -> Result<Vec<RecordBatch>> {
// limit 0 ⇒ the server's default page (~200/500 rows). Honest all-Utf8.
let scan = self.scan_rpc(table, 0)?;
preview_to_string_batch(&scan_to_preview(scan))
}
fn scan_filtered(
&self,
_table: &str,
_filter: &ScanFilter,
_columns: &[&str],
) -> Result<Vec<RecordBatch>> {
Err(unsupported(
"scan_filtered",
"the Warehouse.Scan RPC has no predicate pushdown; wire an Arrow Flight \
read (or a filtered-scan RPC) so the server prunes files server-side",
))
}
fn scan_limited(&self, table: &str, max_rows: usize) -> Result<Vec<RecordBatch>> {
let limit = u32::try_from(max_rows).unwrap_or(u32::MAX);
let scan = self.scan_rpc(table, limit)?;
preview_to_string_batch(&scan_to_preview(scan))
}
// ── table / catalog lifecycle ───────────────────────────────────────────
fn ensure_table(
&self,
_table: &str,
_schema: ::iceberg::spec::Schema,
_partition_cols: &[&str],
) -> Result<()> {
Err(unsupported(
"ensure_table",
"table/catalog lifecycle is the server's (single-writer) job; the thin \
client cannot create tables — the server ensures them on first append",
))
}
fn ensure_columns(&self, _table: &str, _canonical: &::iceberg::spec::Schema) -> Result<()> {
Err(unsupported(
"ensure_columns",
"schema evolution is a server-side write; not exposed over the thin gRPC",
))
}
fn table_names(&self) -> Result<Vec<String>> {
Ok(tables_to_names(self.tables_rpc()?))
}
fn scan_preview(&self, table: &str, limit: usize) -> Result<TablePreview> {
let limit = u32::try_from(limit).unwrap_or(u32::MAX);
Ok(scan_to_preview(self.scan_rpc(table, limit)?))
}
/// The thin client CAN carry vectors — that is the point of this RPC. Before it
/// existed the only route was the stringified `Warehouse.Scan`, whose renderer turns
/// a `list<float32>` into a constant Arrow type name, so a thin consumer's semantic
/// tier was blind by construction while a fat one's was not.
fn scan_vectors(
&self,
table: &str,
key_col: &str,
vec_col: &str,
limit: usize,
) -> Result<nornir_warehouse_trait::VectorScan> {
self.scan_vectors_rpc(
table,
key_col,
vec_col,
u32::try_from(limit).unwrap_or(u32::MAX),
)
}
fn describe_columns(&self, _table: &str) -> Result<Vec<super::sql::ColumnInfo>> {
Err(unsupported(
"describe_columns",
"column TYPES are lost on the stringified Warehouse.Scan wire (every \
cell is Utf8); a typed describe needs an Arrow Flight read or a \
dedicated Describe RPC",
))
}
// ── named per-table convenience wrappers ────────────────────────────────
fn append_bench_run(&self, _repo: &str, _run: &BenchRun) -> Result<Uuid> {
Err(unsupported(
"append_bench_run",
"submit bench runs via Telemetry.SubmitBakeoff/SubmitTestResults; the \
server owns the write",
))
}
fn query_bench_runs(&self, _filter: &BenchFilter) -> Result<Vec<BenchRun>> {
Err(unsupported(
"query_bench_runs",
"needs a typed row over the wire (the Warehouse.Scan preview is \
stringified); add a BenchRuns RPC or an Arrow Flight read",
))
}
// ── async surface ────────────────────────────────────────────────────────
// The thin client is a READ-ONLY, server-owned-write backend: every async
// write is Unsupported (submit via the Telemetry.* RPCs), and the async
// reads have no pushdown/typed wire yet (Arrow Flight follow-up), matching
// how the sync `scan_filtered`/`query_bench_runs` are already typed-Unsupported.
async fn append_arrow_async(&self, _table: &str, _batch: RecordBatch) -> Result<()> {
Err(unsupported(
"append_arrow_async",
"the server is the single writer; submit rows via the Telemetry.* RPCs",
))
}
async fn scan_arrow_async(&self, _table: &str) -> Result<Vec<RecordBatch>> {
Err(unsupported(
"scan_arrow_async",
"the async release readers need a typed Arrow read; wire an Arrow \
Flight scan (the sync scan_arrow returns a stringified preview only)",
))
}
async fn append_bench_run_async(&self, _repo: &str, _run: &BenchRun) -> Result<Uuid> {
Err(unsupported(
"append_bench_run_async",
"submit bench runs via Telemetry.SubmitBakeoff/SubmitTestResults; the \
server owns the write",
))
}
async fn query_bench_runs_async(&self, _filter: &BenchFilter) -> Result<Vec<BenchRun>> {
Err(unsupported(
"query_bench_runs_async",
"needs a typed row over the wire; add a BenchRuns RPC or an Arrow Flight read",
))
}
async fn append_git_heat_scan_async(
&self,
_scan: &crate::knowledge::git_heat::GitHeatScan,
) -> Result<()> {
Err(unsupported(
"append_git_heat_scan_async",
"server-owned write; submit knowledge scans server-side",
))
}
fn as_any(&self) -> &dyn std::any::Any {
self
}
}
#[cfg(test)]
mod tests {
use super::*;
fn demo_scan() -> pb::WarehouseScan {
pb::WarehouseScan {
columns: vec!["repo".into(), "n".into()],
rows: vec![
pb::WarehouseRow {
cells: vec!["holger".into(), "3".into()],
},
pb::WarehouseRow {
cells: vec!["znippy".into(), "20".into()],
},
],
}
}
/// RED-when-broken: the `Warehouse.Scan` response maps 1:1 onto a
/// [`TablePreview`] — every column header and every row cell survives. A
/// mapper that dropped a column or a row (or transposed cells) fails here.
#[test]
fn remote_scan_response_maps_to_preview() {
let p = scan_to_preview(demo_scan());
assert_eq!(p.columns, vec!["repo".to_string(), "n".to_string()]);
assert_eq!(p.rows.len(), 2, "both rows survive");
assert_eq!(p.rows[0], vec!["holger".to_string(), "3".to_string()]);
assert_eq!(p.rows[1], vec!["znippy".to_string(), "20".to_string()]);
}
/// The stringified preview reconstructs as an all-`Utf8` [`RecordBatch`] with
/// the right shape (columns → fields, rows → values). RED-when-broken: a
/// mapper that lost the second column or a cell fails the shape asserts.
#[test]
fn remote_preview_reconstructs_utf8_batch() {
let p = scan_to_preview(demo_scan());
let batches = preview_to_string_batch(&p).unwrap();
assert_eq!(batches.len(), 1);
let b = &batches[0];
assert_eq!(b.num_columns(), 2, "both columns present");
assert_eq!(b.num_rows(), 2, "both rows present");
assert_eq!(b.schema().field(0).name(), "repo");
assert_eq!(b.schema().field(1).name(), "n");
assert!(matches!(b.schema().field(0).data_type(), DataType::Utf8));
let repo = b.column(0).as_any().downcast_ref::<StringArray>().unwrap();
assert_eq!(repo.value(0), "holger");
assert_eq!(repo.value(1), "znippy");
let n = b.column(1).as_any().downcast_ref::<StringArray>().unwrap();
assert_eq!(n.value(0), "3");
assert_eq!(n.value(1), "20");
}
/// `Warehouse.Tables` names come back sorted (the trait contract).
#[test]
fn remote_tables_response_sorts_names() {
let t = pb::WarehouseTables {
names: vec!["z_table".into(), "a_table".into(), "m".into()],
};
assert_eq!(
tables_to_names(t),
vec![
"a_table".to_string(),
"m".to_string(),
"z_table".to_string()
]
);
}
/// **The seam proof.** A `RemoteWarehouse` is a real, boxable
/// [`Warehouse`] — it can stand behind the trait object exactly where an
/// `IcebergWarehouse` does. Its writes are typed
/// [`RemoteWarehouseError::Unsupported`] errors (NEVER silent no-ops), which
/// is what a thin reader must be. RED-when-broken: if a write silently
/// returned `Ok(())`, or the error stopped being typed, this fails.
#[test]
fn remote_warehouse_is_a_boxed_warehouse_with_typed_write_errors() {
let wh: Box<dyn Warehouse> =
Box::new(RemoteWarehouse::connect("127.0.0.1:9", "", "").unwrap());
let schema = Arc::new(ArrowSchema::new(vec![Field::new(
"x",
DataType::Utf8,
true,
)]));
let batch = RecordBatch::new_empty(schema);
let err = wh.append_arrow("t", batch).unwrap_err();
let typed = err.downcast_ref::<RemoteWarehouseError>();
assert!(
matches!(
typed,
Some(RemoteWarehouseError::Unsupported {
method: "append_arrow",
..
})
),
"append_arrow is a TYPED Unsupported error, got: {err:#}"
);
// query_bench_runs is likewise a typed error, not a silent empty vec.
let err = wh.query_bench_runs(&BenchFilter::default()).unwrap_err();
assert!(
matches!(
err.downcast_ref::<RemoteWarehouseError>(),
Some(RemoteWarehouseError::Unsupported {
method: "query_bench_runs",
..
})
),
"query_bench_runs is a TYPED Unsupported error, got: {err:#}"
);
// describe_columns (typed schema discovery) is likewise a typed error on
// the thin client — column types are lost on the stringified scan wire.
let err = wh.describe_columns("bench_runs").unwrap_err();
assert!(
matches!(
err.downcast_ref::<RemoteWarehouseError>(),
Some(RemoteWarehouseError::Unsupported {
method: "describe_columns",
..
})
),
"describe_columns is a TYPED Unsupported error, got: {err:#}"
);
}
}