serve/src/lib.rs
// Apache-2.0 licensed. See LICENSE-APACHE.
//! The library behind the `skade-katalog serve` binary: open **one**
//! lock-owning [`RedbCatalog`] and front it over REST and/or gRPC.
//!
//! The serving process owns the redb write-lock and serializes every commit;
//! both fronts (REST via [`skade_katalog_rest`], gRPC via [`skade_katalog_grpc`])
//! run over *clones of the same* `RedbCatalog`, which share the one redb file and
//! its optimistic group-commit path — so a table created over gRPC is
//! immediately loadable over REST, and concurrent commits from either wire are
//! serialized correctly. This is the "ONE live catalog" (PLAN CATALOG DOCTRINE):
//! remote clients are thin and lock-free; co-located callers link skade-katalog
//! and bypass the wire entirely.
//!
//! ## Serving a warehouse the skade data plane wrote (feature `object-store`)
//!
//! `skade::Warehouse::open`/`open_with_store`/`open_with_object_store_config`
//! all build the SAME `RedbCatalog` under the SAME catalog name (`skade`), so a
//! redb file written by the data plane is already readable here — listing works
//! out of the box. What does *not* work by default is a warehouse whose Iceberg
//! blobs live somewhere other than this box's filesystem: the catalog resolves
//! a table to an `s3://…` metadata-location, and a `LocalFsStorageFactory`
//! cannot read it, so `loadTable` fails even though the table is listed.
//!
//! With `--features object-store`, [`ServeConfig::storage`] can carry a skade
//! [`skade::object_store::ObjectStoreConfig`] and [`ServeConfig::warehouse_uri`]
//! the matching `s3://bucket/prefix` root — the same two things the data plane
//! was opened with. See `.nornir/serve-object-store-design.md`.
use std::collections::HashMap;
use std::net::SocketAddr;
use std::path::PathBuf;
use std::sync::Arc;
use anyhow::{Result, bail};
use iceberg::CatalogBuilder;
use iceberg::io::{LocalFsStorageFactory, StorageFactory};
use skade_katalog::{RedbCatalog, RedbCatalogBuilder};
/// Where the served warehouse's Iceberg blobs (metadata JSON, Avro manifests,
/// Parquet) actually live — i.e. which `iceberg::io::StorageFactory` the
/// catalog reads and writes them through.
///
/// [`LocalFs`](Self::LocalFs) is the default and is exactly what
/// `skade-katalog serve` has always used; the object-store variant is additive
/// and only exists under the `object-store` feature.
#[derive(Clone, Debug, Default)]
pub enum WarehouseStorage {
/// The local filesystem (`iceberg::io::LocalFsStorageFactory`). The
/// unchanged default.
#[default]
LocalFs,
/// A skade pluggable object store — S3/MinIO (`RustS3`/`AwsS3`), the
/// embedded local-fs object store, or memory — described by the same
/// serializable config the data plane's
/// `Warehouse::open_with_object_store_config` takes. Requires the
/// `object-store` feature.
#[cfg(feature = "object-store")]
ObjectStore(skade::object_store::ObjectStoreConfig),
}
/// What to serve and where. At least one of `rest` / `grpc` must be set.
#[derive(Clone, Debug)]
pub struct ServeConfig {
/// REST (Iceberg OpenAPI) bind address, if any.
pub rest: Option<SocketAddr>,
/// gRPC (`skade.katalog.v1`) bind address, if any.
pub grpc: Option<SocketAddr>,
/// TLS on the gRPC front (mutual when it names a client CA). `None` = plain HTTP/2.
pub grpc_tls: Option<skade_katalog_grpc::ServerTls>,
/// Warehouse root directory (data files live here); served as `file://…`.
///
/// When [`warehouse_uri`](Self::warehouse_uri) is set this is only the
/// local directory the redb file defaults into — the blobs then live at the
/// URI instead.
pub warehouse: PathBuf,
/// The redb catalog file. Defaults to `<warehouse>/catalog.redb`.
pub db_path: PathBuf,
/// Catalog name (redb key namespace). Defaults to `skade`.
///
/// Must match the name the writer used. Every `skade::Warehouse`
/// constructor uses `skade` — the same default as here — so a data-plane
/// warehouse needs no override.
pub catalog: String,
/// Warehouse root **URI** stamped into new table metadata, overriding the
/// `file://<warehouse>` default. Set this to the `warehouse_uri` the data
/// plane was opened with (e.g. `s3://red/skade-warehouse`) when serving an
/// object-store warehouse. `None` (the default) reproduces the historical
/// `file://<warehouse>` behaviour exactly.
pub warehouse_uri: Option<String>,
/// Which storage the blobs are read/written through. Defaults to
/// [`WarehouseStorage::LocalFs`] — the historical behaviour.
pub storage: WarehouseStorage,
}
impl ServeConfig {
/// Build a config from a warehouse root, defaulting `db_path` and `catalog`.
pub fn new(warehouse: impl Into<PathBuf>) -> Self {
let warehouse = warehouse.into();
let db_path = warehouse.join("catalog.redb");
Self {
rest: None,
grpc: None,
grpc_tls: None,
warehouse,
db_path,
catalog: "skade".to_string(),
warehouse_uri: None,
storage: WarehouseStorage::default(),
}
}
/// The warehouse root location handed to `RedbCatalogBuilder` —
/// [`warehouse_uri`](Self::warehouse_uri) when set, else the historical
/// `file://<warehouse>`.
pub fn warehouse_location(&self) -> String {
self.warehouse_uri
.clone()
.unwrap_or_else(|| format!("file://{}", self.warehouse.display()))
}
/// The `iceberg::io::StorageFactory` implied by [`storage`](Self::storage).
fn storage_factory(&self) -> Arc<dyn StorageFactory> {
match &self.storage {
WarehouseStorage::LocalFs => Arc::new(LocalFsStorageFactory),
#[cfg(feature = "object-store")]
WarehouseStorage::ObjectStore(config) => Arc::new(
skade::object_store::ObjectStoreFactory::from_config(config.clone()),
),
}
}
}
/// Open the single lock-owning `RedbCatalog` for `cfg` (creates the warehouse
/// dir if needed). redb rejects a second process opening the same file, so this
/// handle *is* the write-lock owner for the lifetime of the serve process.
pub async fn open_catalog(cfg: &ServeConfig) -> Result<RedbCatalog> {
std::fs::create_dir_all(&cfg.warehouse)?;
let cat = RedbCatalogBuilder::default()
.db_path(cfg.db_path.clone())
.warehouse_location(cfg.warehouse_location())
.with_storage_factory(cfg.storage_factory())
.load(&cfg.catalog, HashMap::new())
.await?;
Ok(cat)
}
/// How the gRPC front is served, for the start line.
fn grpc_mode(cfg: &ServeConfig) -> &'static str {
match &cfg.grpc_tls {
None => "",
Some(t) if t.client_ca_pem.is_some() => " (mutual TLS)",
Some(_) => " (TLS)",
}
}
/// Open one catalog and run the configured fronts **until the process exits**
/// (blocks). Both fronts share the same catalog (clones of one redb handle).
pub async fn run(cfg: ServeConfig) -> Result<()> {
let cat = open_catalog(&cfg).await?;
match (cfg.rest, cfg.grpc) {
(Some(r), Some(g)) => {
eprintln!(
"skade-katalog: REST on {r}, gRPC on {g}{} (warehouse {})",
grpc_mode(&cfg),
cfg.warehouse_location()
);
tokio::try_join!(
skade_katalog_rest::serve(cat.clone(), r),
skade_katalog_grpc::serve_with(cat, g, cfg.grpc_tls.clone()),
)?;
}
(Some(r), None) => {
eprintln!(
"skade-katalog: REST on {r} (warehouse {})",
cfg.warehouse_location()
);
skade_katalog_rest::serve(cat, r).await?;
}
(None, Some(g)) => {
eprintln!(
"skade-katalog: gRPC on {g}{} (warehouse {})",
grpc_mode(&cfg),
cfg.warehouse_location()
);
skade_katalog_grpc::serve_with(cat, g, cfg.grpc_tls.clone()).await?;
}
(None, None) => bail!("nothing to serve: pass --rest ADDR and/or --grpc ADDR"),
}
Ok(())
}
/// Open one catalog and **spawn** the configured fronts on their (possibly
/// ephemeral) ports in the background, returning the bound `(rest, grpc)`
/// addresses. The testable seam: both fronts front the one catalog.
pub async fn spawn(cfg: &ServeConfig) -> Result<(Option<SocketAddr>, Option<SocketAddr>)> {
let cat = open_catalog(cfg).await?;
let rest = match cfg.rest {
Some(addr) => Some(skade_katalog_rest::spawn(cat.clone(), addr).await?),
None => None,
};
let grpc = match cfg.grpc {
Some(addr) => Some(skade_katalog_grpc::spawn_with(cat, addr, cfg.grpc_tls.clone()).await?),
None => None,
};
Ok((rest, grpc))
}