nordisk/skade

grpc/tests/grpc_roundtrip.rs

//! In-process gRPC round-trip against the native `skade.katalog.v1.Catalog`
//! service — the RED-when-broken conformance test for skade-katalog-grpc.
//!
//! Spins the tonic service over an embedded `RedbCatalog` on an ephemeral port,
//! connects a generated client, and drives create-namespace → create-table →
//! load-table → commit-table (a property update). Asserts the commit is
//! persisted (a fresh load sees the new property). A second test commits with a
//! stale `assert-table-uuid` requirement and asserts the server returns gRPC
//! **ABORTED** — the optimistic-concurrency guard mapped from
//! `CatalogCommitConflicts`.

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

use iceberg::io::LocalFsStorageFactory;
use iceberg::spec::{NestedField, PrimitiveType, Schema, Type};
use iceberg::{CatalogBuilder, TableRequirement, TableUpdate};
use skade_katalog::{RedbCatalog, RedbCatalogBuilder};
use skade_katalog_grpc::pb;
use tempfile::TempDir;

#[cfg(feature = "testmatrix")]
macro_rules! matrix {
    ($check:expr, $ok:expr, $detail:expr) => {
        nornir_testmatrix::functional_status("skade-katalog-grpc", $check, $ok, $detail);
    };
}
#[cfg(not(feature = "testmatrix"))]
macro_rules! matrix {
    ($check:expr, $ok:expr, $detail:expr) => {{
        let _ = ($check, $ok, $detail);
    }};
}

fn schema() -> Schema {
    Schema::builder()
        .with_schema_id(0)
        .with_fields(vec![
            NestedField::required(1, "id", Type::Primitive(PrimitiveType::Long)).into(),
        ])
        .build()
        .expect("schema")
}

async fn embedded(tmp: &TempDir) -> RedbCatalog {
    let warehouse = tmp.path().join("warehouse");
    std::fs::create_dir_all(&warehouse).unwrap();
    RedbCatalogBuilder::default()
        .db_path(tmp.path().join("catalog.redb"))
        .warehouse_location(format!("file://{}", warehouse.display()))
        .with_storage_factory(Arc::new(LocalFsStorageFactory))
        .load("skade", HashMap::new())
        .await
        .expect("embedded catalog")
}

type Client = pb::catalog_client::CatalogClient<tonic::transport::Channel>;

async fn connect(addr: std::net::SocketAddr) -> Client {
    // Tiny retry: the spawned server may not have accepted yet.
    for _ in 0..50 {
        if let Ok(c) = Client::connect(format!("http://{addr}")).await {
            return c;
        }
        tokio::time::sleep(std::time::Duration::from_millis(10)).await;
    }
    panic!("could not connect to gRPC server at {addr}");
}

#[tokio::test]
async fn grpc_create_load_commit_round_trips() {
    let tmp = TempDir::new().unwrap();
    let cat = embedded(&tmp).await;
    let bound = skade_katalog_grpc::spawn(cat, "127.0.0.1:0".parse().unwrap())
        .await
        .expect("grpc bound");
    let mut client = connect(bound).await;

    // create namespace
    client
        .create_namespace(pb::CreateNamespaceReq {
            levels: vec!["db".into()],
            properties: HashMap::new(),
        })
        .await
        .expect("create_namespace");

    // create table
    let created = client
        .create_table(pb::CreateTableReq {
            namespace: vec!["db".into()],
            name: "t".into(),
            schema_json: serde_json::to_string(&schema()).unwrap(),
            location: String::new(),
        })
        .await
        .expect("create_table")
        .into_inner();
    let loc_before = created.metadata_location.clone();
    assert!(!loc_before.is_empty(), "create returns a metadata location");

    // load table
    let loaded = client
        .load_table(pb::TableIdentMsg { namespace: vec!["db".into()], name: "t".into() })
        .await
        .expect("load_table")
        .into_inner();
    assert_eq!(loaded.metadata_location, loc_before);

    // commit a property update through the gRPC commit_table path
    let update = TableUpdate::SetProperties {
        updates: HashMap::from([("answer".to_string(), "42".to_string())]),
    };
    let committed = client
        .commit_table(pb::CommitTableReq {
            namespace: vec!["db".into()],
            name: "t".into(),
            requirements_json: vec![],
            updates_json: vec![serde_json::to_string(&update).unwrap()],
        })
        .await
        .expect("commit_table")
        .into_inner();
    assert_ne!(committed.metadata_location, loc_before, "commit advances the pointer");

    // a fresh load sees the persisted property
    let reloaded = client
        .load_table(pb::TableIdentMsg { namespace: vec!["db".into()], name: "t".into() })
        .await
        .expect("reload")
        .into_inner();
    assert_eq!(reloaded.metadata_location, committed.metadata_location);
    let meta: serde_json::Value = serde_json::from_slice(&reloaded.metadata_json).unwrap();
    assert_eq!(
        meta["properties"]["answer"].as_str(),
        Some("42"),
        "gRPC load sees the committed property"
    );
    matrix!("commit_table.round_trip", true, "gRPC create+load+commit persisted");
}

#[tokio::test]
async fn grpc_commit_conflicts_on_stale_requirement() {
    let tmp = TempDir::new().unwrap();
    let cat = embedded(&tmp).await;
    let bound = skade_katalog_grpc::spawn(cat, "127.0.0.1:0".parse().unwrap())
        .await
        .expect("grpc bound");
    let mut client = connect(bound).await;

    client
        .create_namespace(pb::CreateNamespaceReq { levels: vec!["db".into()], properties: HashMap::new() })
        .await
        .expect("create_namespace");
    client
        .create_table(pb::CreateTableReq {
            namespace: vec!["db".into()],
            name: "t".into(),
            schema_json: serde_json::to_string(&schema()).unwrap(),
            location: String::new(),
        })
        .await
        .expect("create_table");

    // A requirement asserting a UUID the table does not have → CatalogCommitConflicts.
    let bad = TableRequirement::UuidMatch { uuid: uuid::Uuid::nil() };
    let status = client
        .commit_table(pb::CommitTableReq {
            namespace: vec!["db".into()],
            name: "t".into(),
            requirements_json: vec![serde_json::to_string(&bad).unwrap()],
            updates_json: vec![],
        })
        .await
        .expect_err("stale requirement must be rejected");
    assert_eq!(
        status.code(),
        tonic::Code::Aborted,
        "stale assert-table-uuid must map to gRPC ABORTED, got {status:?}"
    );
    matrix!("commit_table.conflict", status.code() == tonic::Code::Aborted, "stale uuid → ABORTED");
}