serve/src/main.rs
// Apache-2.0 licensed. See LICENSE-APACHE.
//! `skade-katalog serve [--rest ADDR] [--grpc ADDR] --warehouse ROOT [--db PATH] [--catalog NAME]`
//!
//! Opens one lock-owning `RedbCatalog` and fronts it over REST and/or gRPC.
//! Pure-Rust, zero-shell: args are parsed by hand (no external CLI dep).
//!
//! With `--features object-store` the same binary can front a warehouse whose
//! Iceberg blobs live in S3/MinIO (`--warehouse-uri` + `--s3-*`), which is what
//! `skade::Warehouse::open_with_store` writes. Credentials come from the
//! ENVIRONMENT only — never from a flag, so they never land in `ps` output or a
//! shell history.
use std::net::SocketAddr;
use std::path::PathBuf;
use anyhow::{Context, Result, bail};
use skade_katalog_serve::{ServeConfig, run};
const USAGE: &str = "\
skade-katalog serve — one lock-owning RedbCatalog, fronted over REST and/or gRPC
USAGE:
skade-katalog serve --warehouse ROOT [--rest ADDR] [--grpc ADDR] [--db PATH] [--catalog NAME]
skade-katalog serve --warehouse-uri s3://BUCKET/PREFIX --db PATH --s3-bucket BUCKET \\
--s3-endpoint URL [--rest ADDR] [--grpc ADDR]
OPTIONS:
--warehouse ROOT Warehouse root directory (data files; served as file://ROOT).
Required unless --warehouse-uri is given.
--rest ADDR Serve the Iceberg-REST front on ADDR (e.g. 0.0.0.0:8181).
--grpc ADDR Serve the native gRPC front on ADDR (e.g. 0.0.0.0:8182).
--grpc-tls-cert PEM --grpc-tls-key PEM
Serve the gRPC front over TLS with this certificate + key.
--grpc-client-ca PEM
Mutual TLS: a client must present a certificate signed by
this CA (needs --grpc-tls-cert/--grpc-tls-key).
--db PATH redb catalog file (default: ROOT/catalog.redb).
--catalog NAME Catalog name (default: skade). skade::Warehouse also uses `skade`.
Object-store warehouse (requires a build with --features object-store):
--warehouse-uri U Warehouse root URI stamped into new table metadata, e.g.
s3://red/skade-warehouse. Default: file://ROOT.
--s3-bucket NAME S3/MinIO bucket holding the warehouse. Turns on the S3 store.
--s3-endpoint URL Custom S3 endpoint (MinIO), e.g. https://minio:9000.
Omit for real AWS.
--s3-region NAME Region for SigV4 (default: us-east-1).
--s3-virtual-hosted
Use virtual-hosted addressing. Default is path-style (MinIO).
Credentials are read from the environment, never from a flag:
SKADE_S3_ACCESS_KEY_ID | AWS_ACCESS_KEY_ID | MINIO_ACCESS_KEY
SKADE_S3_SECRET_ACCESS_KEY | AWS_SECRET_ACCESS_KEY | MINIO_SECRET_KEY
-V, --version Print `skade v<ver> (<git-hash>)` and exit.
-h, --help Print this help and exit.
At least one of --rest / --grpc must be given.";
/// The gRPC front's TLS from its three flags: cert + key together, a client
/// CA only on top of them. Files are read here, so a bad path fails at start.
fn grpc_tls(
cert: Option<PathBuf>,
key: Option<PathBuf>,
client_ca: Option<PathBuf>,
) -> Result<Option<skade_katalog_grpc::ServerTls>> {
let read = |what: &str, p: &PathBuf| std::fs::read(p).with_context(|| format!("{what} {}", p.display()));
match (cert, key, client_ca) {
(None, None, None) => Ok(None),
(Some(c), Some(k), ca) => Ok(Some(skade_katalog_grpc::ServerTls {
cert_pem: read("--grpc-tls-cert", &c)?,
key_pem: read("--grpc-tls-key", &k)?,
client_ca_pem: ca.as_ref().map(|p| read("--grpc-client-ca", p)).transpose()?,
})),
(None, None, Some(_)) => bail!("--grpc-client-ca needs --grpc-tls-cert and --grpc-tls-key"),
_ => bail!("--grpc-tls-cert and --grpc-tls-key come together"),
}
}
/// First non-empty environment variable among `names`.
#[cfg_attr(not(feature = "object-store"), allow(dead_code))]
fn env_any(names: &[&str]) -> Option<String> {
names
.iter()
.find_map(|n| std::env::var(n).ok().filter(|v| !v.is_empty()))
}
/// `skade v<version> (<git-hash>)` — the canonical skade product version
/// (`skade_katalog::VERSION`) paired with the commit this binary was built from
/// (`SKADE_GIT_HASH`, stamped by build.rs from `.git/` with no shell; `unknown`
/// off-repo). This is the single version surface for the serve binary.
fn version_line() -> String {
format!(
"skade v{} ({})",
skade_katalog::VERSION,
env!("SKADE_GIT_HASH")
)
}
#[tokio::main]
async fn main() -> Result<()> {
let mut args = std::env::args().skip(1);
// `-V` / `--version` short-circuits before anything else (accepted with or
// without a leading `serve` subcommand): print `skade v<ver> (<hash>)` and exit.
let mut first = args.next();
if first.as_deref() == Some("serve") {
first = args.next();
}
if matches!(first.as_deref(), Some("-V") | Some("--version")) {
println!("{}", version_line());
return Ok(());
}
if matches!(first.as_deref(), Some("-h") | Some("--help") | None) {
println!("{USAGE}");
if first.is_none() {
bail!("no arguments given");
}
return Ok(());
}
let mut rest: Option<SocketAddr> = None;
let mut grpc: Option<SocketAddr> = None;
let mut grpc_tls_cert: Option<PathBuf> = None;
let mut grpc_tls_key: Option<PathBuf> = None;
let mut grpc_client_ca: Option<PathBuf> = None;
let mut warehouse: Option<PathBuf> = None;
let mut db: Option<PathBuf> = None;
let mut catalog: Option<String> = None;
// Additive object-store flags. Parsed unconditionally so a binary built
// WITHOUT the feature gives a precise "rebuild with --features
// object-store" instead of "unknown argument".
let mut warehouse_uri: Option<String> = None;
let mut s3_bucket: Option<String> = None;
let mut s3_endpoint: Option<String> = None;
let mut s3_region: Option<String> = None;
let mut s3_virtual_hosted = false;
// Re-thread the first token back into the iterator by handling it, then the rest.
let mut pending = first;
loop {
let Some(flag) = pending.take().or_else(|| args.next()) else {
break;
};
let val = |args: &mut dyn Iterator<Item = String>, f: &str| -> Result<String> {
args.next().with_context(|| format!("{f} requires a value"))
};
match flag.as_str() {
"--rest" => rest = Some(val(&mut args, "--rest")?.parse().context("--rest ADDR")?),
"--grpc" => grpc = Some(val(&mut args, "--grpc")?.parse().context("--grpc ADDR")?),
"--grpc-tls-cert" => grpc_tls_cert = Some(PathBuf::from(val(&mut args, "--grpc-tls-cert")?)),
"--grpc-tls-key" => grpc_tls_key = Some(PathBuf::from(val(&mut args, "--grpc-tls-key")?)),
"--grpc-client-ca" => grpc_client_ca = Some(PathBuf::from(val(&mut args, "--grpc-client-ca")?)),
"--warehouse" => warehouse = Some(PathBuf::from(val(&mut args, "--warehouse")?)),
"--db" => db = Some(PathBuf::from(val(&mut args, "--db")?)),
"--catalog" => catalog = Some(val(&mut args, "--catalog")?),
"--warehouse-uri" => warehouse_uri = Some(val(&mut args, "--warehouse-uri")?),
"--s3-bucket" => s3_bucket = Some(val(&mut args, "--s3-bucket")?),
"--s3-endpoint" => s3_endpoint = Some(val(&mut args, "--s3-endpoint")?),
"--s3-region" => s3_region = Some(val(&mut args, "--s3-region")?),
"--s3-virtual-hosted" => s3_virtual_hosted = true,
"-h" | "--help" => {
println!("{USAGE}");
return Ok(());
}
"-V" | "--version" => {
println!("{}", version_line());
return Ok(());
}
other => bail!("unknown argument {other:?}\n\n{USAGE}"),
}
}
// `--warehouse` stays required in the historical (local-fs) shape. It is
// only optional in the additive object-store shape, where the blobs live at
// `--warehouse-uri` and the only local artefact is the redb file — which
// must then be named explicitly with `--db`.
let warehouse = match (warehouse, &warehouse_uri, &db) {
(Some(w), _, _) => w,
(None, Some(_), Some(db)) => db
.parent()
.filter(|p| !p.as_os_str().is_empty())
.map(PathBuf::from)
.unwrap_or_else(|| PathBuf::from(".")),
(None, Some(_), None) => {
bail!("--warehouse-uri needs --db PATH (the local redb file)\n\n{USAGE}")
}
(None, None, _) => bail!("--warehouse ROOT is required\n\n{USAGE}"),
};
let mut cfg = ServeConfig::new(warehouse);
cfg.rest = rest;
cfg.grpc = grpc;
cfg.grpc_tls = grpc_tls(grpc_tls_cert, grpc_tls_key, grpc_client_ca)?;
if let Some(db) = db {
cfg.db_path = db;
}
if let Some(catalog) = catalog {
cfg.catalog = catalog;
}
cfg.warehouse_uri = warehouse_uri;
apply_object_store(
&mut cfg,
s3_bucket,
s3_endpoint,
s3_region,
s3_virtual_hosted,
)?;
if cfg.rest.is_none() && cfg.grpc.is_none() {
bail!("at least one of --rest / --grpc must be given\n\n{USAGE}");
}
run(cfg).await
}
/// Whether `uri` names storage this box's filesystem cannot reach — i.e. it
/// carries a scheme and that scheme is not `file`.
///
/// A bare path and a `file://` URI both stay local; `s3://…` (or anything else
/// with a scheme) needs a storage factory wired in, and saying so at parse time
/// is the difference between one clear line and a catalog that lists tables it
/// then cannot load.
fn needs_object_store(uri: &str) -> bool {
uri.split_once("://")
.is_some_and(|(scheme, _)| !scheme.eq_ignore_ascii_case("file"))
}
/// Wire the `--s3-*` flags onto `cfg` as a skade object store. Credentials come
/// from the environment (see `USAGE`), never from a flag.
///
/// Every consistency check runs BEFORE the environment is read, so the
/// misconfiguration is reported even on a box that has no credentials set (and
/// so the checks stay testable without touching the process environment).
#[cfg(feature = "object-store")]
fn apply_object_store(
cfg: &mut ServeConfig,
bucket: Option<String>,
endpoint: Option<String>,
region: Option<String>,
virtual_hosted: bool,
) -> Result<()> {
use skade::object_store::{ObjectStoreConfig, S3Config};
use skade_katalog_serve::WarehouseStorage;
let Some(bucket) = bucket else {
if endpoint.is_some() || region.is_some() || virtual_hosted {
bail!("--s3-endpoint/--s3-region/--s3-virtual-hosted need --s3-bucket NAME");
}
// The other half of the pairing. Without it the catalog would come up
// on LocalFs while stamping `s3://…` locations: every listing works and
// every loadTable fails with "No such file or directory" — exactly the
// half-working catalog this whole change exists to remove.
if let Some(uri) = &cfg.warehouse_uri {
if needs_object_store(uri) {
bail!(
"--warehouse-uri {uri} is not on this filesystem: \
it needs --s3-bucket NAME (and --s3-endpoint for MinIO)"
);
}
}
return Ok(());
};
// Checked before the credential lookup: a missing --warehouse-uri is a
// misconfiguration whether or not the box has keys in its environment.
if cfg.warehouse_uri.is_none() {
bail!("--s3-bucket needs --warehouse-uri s3://BUCKET/PREFIX (the root the writer used)");
}
let access_key_id = env_any(&[
"SKADE_S3_ACCESS_KEY_ID",
"AWS_ACCESS_KEY_ID",
"MINIO_ACCESS_KEY",
])
.context(
"no S3 access key in the environment \
(SKADE_S3_ACCESS_KEY_ID | AWS_ACCESS_KEY_ID | MINIO_ACCESS_KEY)",
)?;
let secret_access_key = env_any(&[
"SKADE_S3_SECRET_ACCESS_KEY",
"AWS_SECRET_ACCESS_KEY",
"MINIO_SECRET_KEY",
])
.context(
"no S3 secret key in the environment \
(SKADE_S3_SECRET_ACCESS_KEY | AWS_SECRET_ACCESS_KEY | MINIO_SECRET_KEY)",
)?;
let s3 = S3Config {
endpoint: endpoint.unwrap_or_default(),
region: region.unwrap_or_else(|| "us-east-1".to_string()),
bucket,
access_key_id,
secret_access_key,
path_style: !virtual_hosted,
tls: None,
};
cfg.storage = WarehouseStorage::ObjectStore(ObjectStoreConfig::RustS3(s3));
Ok(())
}
/// Without the `object-store` feature the `--s3-*` flags are accepted by the
/// parser but cannot be honoured — say so precisely instead of failing later.
#[cfg(not(feature = "object-store"))]
fn apply_object_store(
cfg: &mut ServeConfig,
bucket: Option<String>,
endpoint: Option<String>,
region: Option<String>,
virtual_hosted: bool,
) -> Result<()> {
if bucket.is_some() || endpoint.is_some() || region.is_some() || virtual_hosted {
bail!(
"the --s3-* flags need a build with the `object-store` feature: \
cargo build --release --features object-store"
);
}
// `--warehouse-uri` alone is legitimate here — it may relocate a `file://`
// root — but a remote scheme cannot be honoured by a binary that links no
// storage able to reach it, and a stock build links none.
if let Some(uri) = &cfg.warehouse_uri {
if needs_object_store(uri) {
bail!(
"--warehouse-uri {uri} needs a build with the `object-store` feature: \
cargo build --release --features object-store"
);
}
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
/// A `ServeConfig` in the shape `main` hands to `apply_object_store`.
fn cfg(uri: Option<&str>) -> ServeConfig {
let mut c = ServeConfig::new("/tmp/does-not-need-to-exist");
c.warehouse_uri = uri.map(str::to_string);
c
}
#[test]
fn a_scheme_other_than_file_needs_a_store() {
assert!(needs_object_store("s3://red/skade-warehouse"));
assert!(needs_object_store("gs://bucket/prefix"));
assert!(!needs_object_store("file:///var/lib/skade"));
assert!(!needs_object_store("FILE:///var/lib/skade"));
// A bare path has no scheme and stays local — `//` alone must not read
// as one, or `ServeConfig::new`'s own default would trip this.
assert!(!needs_object_store("/var/lib/skade"));
assert!(!needs_object_store("relative/path"));
}
/// The half-working catalog this change exists to remove: an `s3://` root
/// with no store behind it lists every table and loads none. It must fail
/// at startup, not at the first `loadTable`.
#[test]
fn an_s3_root_without_a_bucket_is_refused() {
let mut c = cfg(Some("s3://red/skade-warehouse"));
let e = apply_object_store(&mut c, None, None, None, false)
.expect_err("s3:// root with no --s3-bucket");
let msg = e.to_string();
assert!(msg.contains("s3://red/skade-warehouse"), "{msg}");
assert!(
msg.contains("--s3-bucket") || msg.contains("object-store"),
"the message must name the missing piece: {msg}"
);
}
/// The historical local-fs invocation must be untouched: no new flag, no
/// new requirement, no new failure.
#[test]
fn the_plain_local_shape_is_unchanged() {
let mut c = cfg(None);
apply_object_store(&mut c, None, None, None, false).expect("plain --warehouse");
assert_eq!(c.warehouse_location(), "file:///tmp/does-not-need-to-exist");
let mut c = cfg(Some("file:///srv/wh"));
apply_object_store(&mut c, None, None, None, false).expect("relocated file:// root");
assert_eq!(c.warehouse_location(), "file:///srv/wh");
}
/// `--s3-*` without `--s3-bucket` is a typo, not a configuration.
/// (Feature-gated: a stock build refuses every `--s3-*` flag earlier, with
/// the "rebuild with the feature" message instead — see below.)
#[cfg(feature = "object-store")]
#[test]
fn the_s3_knobs_need_a_bucket() {
let mut c = cfg(Some("s3://red/wh"));
let e = apply_object_store(&mut c, None, Some("https://minio:9000".into()), None, false)
.expect_err("--s3-endpoint alone");
assert!(e.to_string().contains("--s3-bucket"), "{e}");
}
/// A bucket with no root: the writer's `warehouse_uri` is not guessable
/// from the bucket name (it carries a prefix), so demand it. Checked before
/// the credential lookup, which is why this passes with an empty
/// environment.
#[cfg(feature = "object-store")]
#[test]
fn a_bucket_without_a_root_is_refused() {
let mut c = cfg(None);
let e = apply_object_store(&mut c, Some("red".into()), None, None, false)
.expect_err("--s3-bucket with no --warehouse-uri");
assert!(e.to_string().contains("--warehouse-uri"), "{e}");
}
/// A stock (feature-off) binary must refuse a remote root outright rather
/// than come up on LocalFs pretending to serve it.
#[cfg(not(feature = "object-store"))]
#[test]
fn a_stock_build_refuses_a_remote_root() {
let mut c = cfg(Some("s3://red/wh"));
let e = apply_object_store(&mut c, None, None, None, false).expect_err("no feature");
assert!(e.to_string().contains("object-store"), "{e}");
}
/// …and it names the feature rather than the flag when an `--s3-*` knob is
/// passed to a binary that cannot honour it. "unknown argument" would send
/// the operator hunting for a typo that is not there.
#[cfg(not(feature = "object-store"))]
#[test]
fn a_stock_build_points_at_the_feature_not_a_typo() {
let mut c = cfg(None);
let e = apply_object_store(&mut c, Some("red".into()), None, None, false)
.expect_err("no feature");
let msg = e.to_string();
assert!(msg.contains("object-store"), "{msg}");
assert!(!msg.contains("unknown argument"), "{msg}");
}
}