apunkt/omnigraph
1
0
Fork 0
mirror of https://github.com/ModernRelay/omnigraph.git synced 2026-06-21 02:28:07 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions 5

feat(server)!: cluster-only server — remove single-graph serving (RFC-011) (#250)

Browse source 
omnigraph-server boots only from --cluster; all HTTP is /graphs/<id>/…; flat single-graph routes and the omnigraph.yaml server boot are removed. GraphRouting/ServerConfigMode collapse to multi-only; openapi.json regenerated to the nested shape; ~100 server route tests migrated; parity/system_local boot from a converged cluster. Gate green (1410 tests).
This commit is contained in:
Andrew Altshuler 2026-06-15 20:17:25 +03:00 • committed by GitHub
parent b183db078f
commit 8b01c6e547
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
20 changed files with 988 additions and 1492 deletions
Download patch file Download diff file Expand all files Collapse all files

9
crates/omnigraph-cli/tests/parity_matrix.rs
View file

@ -37,9 +37,12 @@ struct Parity {
fn parity() -> Parity {
let (temp, local, remote) = twin_graphs();
let (local_cfg, server_cfg) = parity_configs(temp.path(), &local, &remote);
let server = spawn_server_with_config_env(
&server_cfg,
// RFC-011 cluster-only: the remote arm is served from a converged
// cluster directory (one graph, id `parity`), seeded with the same
// fixture data as the local twin.
let (local_cfg, cluster_dir) = parity_configs(temp.path(), &local, &remote);
let server = spawn_server_with_cluster_env(
&cluster_dir,
&[(
"OMNIGRAPH_SERVER_BEARER_TOKENS_JSON",
r#"{"act-parity":"parity-tok"}"#,

158
crates/omnigraph-cli/tests/support/mod.rs
View file

@ -339,6 +339,63 @@ impl SystemGraph {
}
}
/// A converged cluster directory the server can boot from (`--cluster`),
/// serving one graph seeded with the standard fixture. Holds the temp dir
/// alive for the test's lifetime.
pub struct ClusterFixture {
_temp: TempDir,
dir: PathBuf,
}
impl ClusterFixture {
pub fn path(&self) -> &Path {
&self.dir
}
}
/// Build a converged cluster (RFC-011 cluster-only serving) with a single
/// graph `graph_id`, seeded with the `test.jsonl` fixture so reads return
/// data. When `policy_yaml` is `Some`, the bundle is bound to the graph
/// scope. The server boots from the returned path via `--cluster`.
pub fn converged_loaded_cluster(graph_id: &str, policy_yaml: Option<&str>) -> ClusterFixture {
let temp = tempdir().unwrap();
let dir = temp.path().to_path_buf();
fs::copy(fixture("test.pg"), dir.join("graph.pg")).unwrap();
let policy_block = match policy_yaml {
Some(source) => {
fs::write(dir.join("graph.policy.yaml"), source).unwrap();
format!(
"policies:\n graph:\n file: ./graph.policy.yaml\n applies_to: [{graph_id}]\n"
)
}
None => String::new(),
};
fs::write(
dir.join("cluster.yaml"),
format!(
"version: 1\nmetadata:\n name: sys\nstate:\n backend: cluster\n lock: true\ngraphs:\n {graph_id}:\n schema: ./graph.pg\n{policy_block}"
),
)
.unwrap();
output_success(cli().arg("cluster").arg("import").arg("--config").arg(&dir));
output_success(cli().arg("cluster").arg("apply").arg("--config").arg(&dir));
let served_root = dir.join("graphs").join(format!("{graph_id}.omni"));
output_success(
cli()
.arg("load")
.arg("--data")
.arg(fixture("test.jsonl"))
.arg("--mode")
.arg("overwrite")
.arg(&served_root),
);
ClusterFixture { _temp: temp, dir }
}
// ---- helpers moved from the monolithic tests/cli.rs ----
#[allow(unused_imports)]
use lance::Dataset;
@ -788,29 +845,104 @@ rules:
.to_string()
}
/// Per-arm config files carrying the same policy. Both arms address the
/// graph by positional URI, so the TOP-LEVEL policy.file applies on each
/// side (single-graph semantics).
pub fn parity_configs(root: &Path, _local_graph: &Path, remote_graph: &Path) -> (PathBuf, PathBuf) {
/// The graph id the parity cluster serves the remote arm under. The
/// remote arm addresses it with `--graph PARITY_GRAPH_ID` (RFC-011: the
/// server is cluster-only, so a graph selector is required).
pub const PARITY_GRAPH_ID: &str = "parity";
/// Build both arms' configuration (RFC-011 cluster-only server).
///
/// * Local arm: a `--config` file carrying the TOP-LEVEL `policy.file`
/// (single-graph embedded semantics), used as-is by `run_both_with_config`.
/// * Remote arm: a converged cluster directory whose single graph (id
/// `parity`) carries the SAME Cedar bundle (bound to the graph scope).
/// The cluster's derived graph root (`<dir>/graphs/parity.omni`) is
/// seeded with the SAME fixture data as the local twin so the two arms
/// compare like-for-like.
///
/// `local_graph` is overwritten with a byte-for-byte copy of the cluster's
/// seeded served graph so identity-bearing values that are NOT scrubbed
/// (e.g. `graph_commit_id`, edge `id`s in export) match across the arms —
/// the served graph is the source of truth and the local twin mirrors it.
///
/// Returns `(local_config_path, cluster_dir)`. The caller spawns the
/// server with `--cluster <cluster_dir>`.
pub fn parity_configs(root: &Path, local_graph: &Path, _remote_graph: &Path) -> (PathBuf, PathBuf) {
let policy = root.join("parity.policy.yaml");
fs::write(&policy, parity_policy_yaml()).unwrap();
// Local arm config: top-level single-graph policy.
let local_cfg = root.join("local.omnigraph.yaml");
fs::write(
&local_cfg,
format!("policy:\n file: {}\n", policy.display()),
)
.unwrap();
let server_cfg = root.join("server.omnigraph.yaml");
// Remote arm: a cluster directory the server boots from. One graph
// (`parity`), schema = the shared fixture, policy bound to the graph.
let cluster_dir = root.join("parity-cluster");
fs::create_dir_all(&cluster_dir).unwrap();
fs::copy(fixture("test.pg"), cluster_dir.join("parity.pg")).unwrap();
fs::copy(&policy, cluster_dir.join("parity.policy.yaml")).unwrap();
fs::write(
&server_cfg,
cluster_dir.join("cluster.yaml"),
format!(
"server:\n graph: parity\ngraphs:\n parity:\n uri: {}\n policy:\n file: {}\n",
remote_graph.display(),
policy.display()
r#"version: 1
metadata:
name: parity
state:
backend: cluster
lock: true
graphs:
{PARITY_GRAPH_ID}:
schema: ./parity.pg
policies:
parity:
file: ./parity.policy.yaml
applies_to: [{PARITY_GRAPH_ID}]
"#
),
)
.unwrap();
(local_cfg, server_cfg)
// Converge the cluster (creates the empty graph at the derived root),
// then seed it with the same fixture data the local twin holds.
output_success(
cli()
.arg("cluster")
.arg("import")
.arg("--config")
.arg(&cluster_dir),
);
output_success(
cli()
.arg("cluster")
.arg("apply")
.arg("--config")
.arg(&cluster_dir),
);
let served_root = cluster_dir
.join("graphs")
.join(format!("{PARITY_GRAPH_ID}.omni"));
output_success(
cli()
.arg("load")
.arg("--data")
.arg(fixture("test.jsonl"))
.arg("--mode")
.arg("overwrite")
.arg(&served_root),
);
// Mirror the seeded served graph into the local twin so both arms hold
// identical ULIDs / commit ids (the served graph is authoritative).
if local_graph.exists() {
fs::remove_dir_all(local_graph).unwrap();
}
copy_dir(&served_root, local_graph);
(local_cfg, cluster_dir)
}
/// Run one CLI invocation per arm with identical verb args: locally against
@ -853,7 +985,11 @@ pub fn run_both_with_config(
.env("OMNIGRAPH_BEARER_TOKEN", PARITY_TOKEN)
.args(args)
.arg("--server")
.arg(server_url);
.arg(server_url)
// RFC-011: the parity server is cluster-only (multi-graph), so the
// remote arm must name the graph it addresses.
.arg("--graph")
.arg(PARITY_GRAPH_ID);
let remote_out = remote.output().unwrap();
(local_out, remote_out)
}

59
crates/omnigraph-cli/tests/system_local.rs
View file

@ -2319,9 +2319,12 @@ fn cluster_server_boot_ignores_local_config_in_cwd() {
/// 3), and `logout` revokes.
#[test]
fn local_cli_keyed_credentials_authenticate_url_matched_server() {
let graph = SystemGraph::loaded();
let server = spawn_server_with_env(
graph.path(),
// RFC-011 cluster-only: the server boots from a converged cluster
// serving the fixture graph under id `local`; tokens-only boot is
// default-deny, which still permits `read`.
let cluster = converged_loaded_cluster("local", None);
let server = spawn_server_with_cluster_env(
cluster.path(),
&[("OMNIGRAPH_SERVER_BEARER_TOKEN", "secret-tok")],
);
let operator_home = tempfile::tempdir().unwrap();
@ -2344,6 +2347,8 @@ fn local_cli_keyed_credentials_authenticate_url_matched_server() {
.arg("read")
.arg("--server")
.arg(&server.base_url)
.arg("--graph")
.arg("local")
.arg("--query")
.arg(fixture("test.gq"))
.arg("get_person")
@ -2432,26 +2437,40 @@ fn local_cli_keyed_credentials_authenticate_url_matched_server() {
/// stored queries) end to end, with the keyed credential from PR 2.
#[test]
fn local_cli_operator_alias_and_server_flag_invoke_stored_query() {
let graph = SystemGraph::loaded();
graph.write_query(
"stored-find-person.gq",
// RFC-011 cluster-only: build a converged cluster serving graph `local`
// with a stored query `find_person` and a per-graph policy granting the
// operator invoke_query + read (invoke_query is policy-gated — anti-probing
// 404 without the grant).
let cluster = tempfile::tempdir().unwrap();
fs::copy(fixture("test.pg"), cluster.path().join("local.pg")).unwrap();
fs::write(
cluster.path().join("find-person.gq"),
"query find_person($name: String) { match { $p: Person { name: $name } } return { $p.name } }",
);
// invoke_query is policy-gated (anti-probing 404 without the grant),
// so the server gets a per-graph bundle granting it to the operator.
graph.write_file(
"graph.policy.yaml",
)
.unwrap();
fs::write(
cluster.path().join("graph.policy.yaml"),
"version: 1\ngroups:\n ops: [\"act-op\"]\nprotected_branches: [main]\nrules:\n - id: allow-invoke\n allow:\n actors: { group: ops }\n actions: [invoke_query]\n - id: allow-read\n allow:\n actors: { group: ops }\n actions: [read]\n branch_scope: any\n",
)
.unwrap();
fs::write(
cluster.path().join("cluster.yaml"),
"version: 1\nmetadata:\n name: alias-sys\nstate:\n backend: cluster\n lock: true\ngraphs:\n local:\n schema: ./local.pg\n queries:\n find_person:\n file: ./find-person.gq\npolicies:\n graph:\n file: ./graph.policy.yaml\n applies_to: [local]\n",
)
.unwrap();
output_success(cli().arg("cluster").arg("import").arg("--config").arg(cluster.path()));
output_success(cli().arg("cluster").arg("apply").arg("--config").arg(cluster.path()));
output_success(
cli()
.arg("load")
.arg("--data")
.arg(fixture("test.jsonl"))
.arg("--mode")
.arg("overwrite")
.arg(cluster.path().join("graphs").join("local.omni")),
);
let config = graph.write_config(
"omnigraph-server.yaml",
&format!(
"graphs:\n local:\n uri: {}\n policy:\n file: ./graph.policy.yaml\n queries:\n find_person:\n file: ./stored-find-person.gq\n",
yaml_string(&graph.path().to_string_lossy())
),
);
let server = spawn_server_with_config_env(
&config,
let server = spawn_server_with_cluster_env(
cluster.path(),
&[(
"OMNIGRAPH_SERVER_BEARER_TOKENS_JSON",
r#"{"act-op":"srv-tok"}"#,

98
crates/omnigraph-server/src/handlers.rs
View file

@ -51,16 +51,7 @@ pub(crate) async fn server_graphs_list(
State(state): State<AppState>,
actor: Option<Extension<ResolvedActor>>,
) -> std::result::Result<Json<GraphListResponse>, ApiError> {
// 405 in single mode — there's no registry to enumerate, and the
// legacy URL surface didn't expose this endpoint.
let registry = match state.routing() {
GraphRouting::Single { .. } => {
return Err(ApiError::method_not_allowed(
"GET /graphs is only available in multi-graph mode",
));
}
GraphRouting::Multi { registry, .. } => registry,
};
let registry = &state.routing().registry;
// Server-level Cedar gate. `state.server_policy` is loaded from
// `server.policy.file` in `omnigraph.yaml` at startup. When no
@ -93,17 +84,15 @@ pub(crate) async fn server_graphs_list(
}
pub(crate) async fn server_openapi(State(state): State<AppState>) -> Json<utoipa::openapi::OpenApi> {
let mut doc = ApiDoc::openapi();
// `served_openapi` is the single nesting source — the protected
// routes always live under `/graphs/{graph_id}/...` (public/management
// paths `/healthz`, `/graphs` stay flat). Building from it here means
// the runtime spec and the committed `openapi.json` share one nesting
// pass and can't drift.
let mut doc = crate::served_openapi();
if !state.requires_bearer_auth() {
strip_security(&mut doc);
}
// MR-668: in multi mode, the protected routes live under
// `/graphs/{graph_id}/...`. Rewrite the doc so the spec matches
// the routes the router actually serves. Public paths (`/healthz`)
// stay flat in both modes.
if matches!(state.routing(), GraphRouting::Multi { .. }) {
nest_paths_under_cluster_prefix(&mut doc);
}
Json(doc)
}
@ -248,16 +237,11 @@ pub(crate) async fn require_bearer_auth(
Ok(next.run(request).await)
}
/// Routing middleware (MR-668). Resolves the active graph for the
/// request and injects `Arc<GraphHandle>` as an extension so handlers can
/// extract it via `Extension<Arc<GraphHandle>>`.
/// Routing middleware (RFC-011 cluster-only). Resolves the active graph
/// for the request and injects `Arc<GraphHandle>` as an extension so
/// handlers can extract it via `Extension<Arc<GraphHandle>>`.
///
/// **Single mode**: the routing field holds the single handle directly.
/// Routes are flat; every request resolves to that handle, regardless
/// of the URI path. No registry walk, no sentinel key, no
/// programmer-error guard.
///
/// **Multi mode**: routes are nested under `/graphs/{graph_id}/...`. The
/// Routes are always nested under `/graphs/{graph_id}/...`. The
/// middleware extracts `{graph_id}` from the URI path and looks it up in
/// the registry. Returns 404 if the graph is not registered.
///
@ -268,39 +252,33 @@ pub(crate) async fn resolve_graph_handle(
mut request: Request,
next: Next,
) -> std::result::Result<Response, ApiError> {
let handle = match &state.routing {
GraphRouting::Single { handle } => Arc::clone(handle),
GraphRouting::Multi { registry, .. } => {
// `Router::nest("/graphs/{graph_id}", inner)` rewrites
// `request.uri().path()` to the inner suffix (e.g. `/snapshot`).
// The pre-rewrite URI is preserved in the `OriginalUri`
// request extension by axum's router; we read from there to
// extract `{graph_id}`. Fall back to the current URI only if
// the extension is missing, which shouldn't happen for
// nested routes but is safe defensive code.
let original_path: String = request
.extensions()
.get::<OriginalUri>()
.map(|OriginalUri(uri)| uri.path().to_string())
.unwrap_or_else(|| request.uri().path().to_string());
let graph_id_str = original_path
.strip_prefix("/graphs/")
.and_then(|rest| rest.split('/').next())
.filter(|s| !s.is_empty())
.ok_or_else(|| {
ApiError::bad_request(
"cluster route missing /graphs/{graph_id} prefix".to_string(),
)
})?;
let graph_id = GraphId::try_from(graph_id_str.to_string())
.map_err(|err| ApiError::bad_request(err.to_string()))?;
let key = GraphKey::cluster(graph_id.clone());
match registry.get(&key) {
RegistryLookup::Ready(handle) => handle,
RegistryLookup::Gone => {
return Err(ApiError::not_found(format!("graph '{graph_id}' not found")));
}
}
let registry = &state.routing.registry;
// `Router::nest("/graphs/{graph_id}", inner)` rewrites
// `request.uri().path()` to the inner suffix (e.g. `/snapshot`).
// The pre-rewrite URI is preserved in the `OriginalUri`
// request extension by axum's router; we read from there to
// extract `{graph_id}`. Fall back to the current URI only if
// the extension is missing, which shouldn't happen for
// nested routes but is safe defensive code.
let original_path: String = request
.extensions()
.get::<OriginalUri>()
.map(|OriginalUri(uri)| uri.path().to_string())
.unwrap_or_else(|| request.uri().path().to_string());
let graph_id_str = original_path
.strip_prefix("/graphs/")
.and_then(|rest| rest.split('/').next())
.filter(|s| !s.is_empty())
.ok_or_else(|| {
ApiError::bad_request("cluster route missing /graphs/{graph_id} prefix".to_string())
})?;
let graph_id = GraphId::try_from(graph_id_str.to_string())
.map_err(|err| ApiError::bad_request(err.to_string()))?;
let key = GraphKey::cluster(graph_id.clone());
let handle = match registry.get(&key) {
RegistryLookup::Ready(handle) => handle,
RegistryLookup::Gone => {
return Err(ApiError::not_found(format!("graph '{graph_id}' not found")));
}
};

206
crates/omnigraph-server/src/lib.rs
View file

@ -1,7 +1,7 @@
pub mod api;
mod handlers;
mod settings;
pub use settings::{load_server_settings, classify_server_runtime_state, server_config_is_multi, ServerRuntimeState};
pub use settings::{load_server_settings, classify_server_runtime_state, ServerRuntimeState};
use settings::*;
use handlers::*;
pub mod auth;
@ -122,6 +122,20 @@ fn hash_bearer_token(token: &str) -> BearerTokenHash {
)]
pub struct ApiDoc;
/// The canonical served OpenAPI shape (RFC-011 cluster-only): the static
/// `ApiDoc` with every protected path nested under `/graphs/{graph_id}/…`
/// and `cluster_`-prefixed operation ids. `/healthz` and `/graphs` stay
/// flat. This is the single source of nesting — both the runtime
/// `server_openapi` handler and the committed `openapi.json` derive from
/// it, so the published spec can never describe routes the server does
/// not serve. The handler additionally strips security in open mode; the
/// committed spec retains it.
pub fn served_openapi() -> utoipa::openapi::OpenApi {
let mut doc = ApiDoc::openapi();
handlers::nest_paths_under_cluster_prefix(&mut doc);
doc
}
struct SecurityAddon;
impl utoipa::Modify for SecurityAddon {
@ -143,11 +157,10 @@ const SERVER_SOURCE_VERSION: Option<&str> = option_env!("OMNIGRAPH_SOURCE_VERSIO
#[derive(Debug, Clone)]
pub struct ServerConfig {
/// Server topology + the graphs to open at startup. Single-mode
/// invocations (`omnigraph-server <URI>` or `--target <name>`)
/// produce `ServerConfigMode::Single`; multi-mode invocations
/// (`--config omnigraph.yaml` with a non-empty `graphs:` map and
/// no single-mode selector) produce `ServerConfigMode::Multi`.
/// Server topology + the graphs to open at startup. RFC-011
/// cluster-only: the server always boots from a cluster
/// (`--cluster <dir | s3://…>`) and serves N graphs under cluster
/// routes.
pub mode: ServerConfigMode,
pub bind: String,
/// Operator opt-in for fully-unauthenticated dev mode (MR-723).
@ -161,41 +174,25 @@ pub struct ServerConfig {
pub allow_unauthenticated: bool,
}
/// What `load_server_settings` produces after applying the four-rule
/// mode inference matrix (MR-668 decision 2).
/// What `load_server_settings` produces. RFC-011 cluster-only: the
/// server always boots from a cluster's applied revision into a
/// multi-graph deployment (N ≥ 1 graphs).
#[derive(Debug, Clone)]
pub enum ServerConfigMode {
/// Legacy invocation — one graph at the given URI. Either:
/// * `omnigraph-server <URI>` (CLI positional), or
/// * `omnigraph-server --target <name> --config omnigraph.yaml`, or
/// * `omnigraph-server --config omnigraph.yaml` with `server.graph`
/// set to a named target.
Single {
uri: String,
/// Cedar graph resource id for the single graph. A named selection
/// uses the graph name; an anonymous URI uses the normalized URI to
/// preserve legacy single-graph policy identity.
graph_id: String,
/// Top-level `policy.file` (single-graph Cedar policy).
policy_file: Option<PathBuf>,
/// Top-level stored-query registry, loaded and identity-checked
/// at settings-build time; type-checked against the schema when
/// the engine opens.
queries: QueryRegistry,
},
/// Multi-graph invocation — `--config omnigraph.yaml` with a
/// non-empty `graphs:` map and no single-mode selector.
/// Cluster boot — `--cluster <dir | s3://…>` resolves the applied
/// revision into per-graph startup configs plus an optional
/// server-level policy.
Multi {
/// Per-graph startup configs, sorted by graph id (BTreeMap
/// iteration order). The parallel-open loop iterates this.
graphs: Vec<GraphStartupConfig>,
/// Path to the config file the server was started from. Kept on
/// the mode so future runtime mutation (deferred — see release
/// notes) can locate the source of truth without re-parsing CLI
/// args.
/// The cluster boot source (config directory or storage root).
/// Kept on the mode so future runtime mutation (deferred — see
/// release notes) can locate the source of truth without
/// re-parsing CLI args.
config_path: PathBuf,
/// `server.policy.file` (server-level Cedar policy for the
/// management endpoints). Wired into `GET /graphs` authorization.
/// Server-level Cedar policy for the management endpoints
/// (`GET /graphs`). Wired into `GET /graphs` authorization.
server_policy: Option<PolicySource>,
},
}
@ -224,36 +221,25 @@ pub struct GraphStartupConfig {
pub queries: QueryRegistry,
}
/// Runtime routing for the server. Single mode = legacy
/// `omnigraph-server <URI>` invocation, one graph, flat HTTP routes.
/// Multi mode = `--config omnigraph.yaml` with a non-empty `graphs:`
/// map, N graphs, cluster routes (`/graphs/{graph_id}/...`). Mode is
/// determined at startup by `load_server_settings`.
/// Runtime routing for the server (RFC-011 cluster-only). Every
/// deployment serves cluster routes (`/graphs/{graph_id}/...`) backed by
/// a registry of N graphs (N ≥ 1). The single-graph convenience
/// constructors build a one-graph registry keyed by `default`; the
/// cluster boot path builds an N-graph registry. There is no longer a
/// flat-route mode.
///
/// In single mode the handle lives here directly — there is no
/// registry, no sentinel key, no walk-and-assert. In multi mode the
/// registry carries N handles and the middleware dispatches on the
/// URL's `{graph_id}` segment.
/// `config_path` is the boot source (the cluster directory or storage
/// root); preserved here so future runtime mutation (deferred) can find
/// the source of truth without re-parsing CLI args. The server treats
/// the source as operator-owned and never writes it.
///
/// Both modes share the same handler bodies — the routing middleware
/// All handler bodies are mode-agnostic — the routing middleware
/// (`resolve_graph_handle`) injects `Arc<GraphHandle>` as a request
/// extension so handlers never see the routing discriminator.
/// extension by looking up the `{graph_id}` URL segment in the registry.
#[derive(Clone)]
pub enum GraphRouting {
/// Single-graph deployment: one handle, flat routes (`/snapshot`,
/// `/read`, …). The `handle.uri` field carries the URI the engine
/// was opened from. Backward compatible with v0.6.0 deployments.
Single { handle: Arc<GraphHandle> },
/// Multi-graph deployment: many handles, cluster routes
/// (`/graphs/{graph_id}/...`). `config_path` is the `omnigraph.yaml`
/// the server reads at startup; preserved here so future runtime
/// mutation (deferred) can find the source of truth without
/// re-parsing CLI args. The server treats the file as
/// operator-owned and never writes it.
Multi {
registry: Arc<GraphRegistry>,
config_path: Option<PathBuf>,
},
pub struct GraphRouting {
pub registry: Arc<GraphRegistry>,
pub config_path: Option<PathBuf>,
}
#[derive(Clone)]
@ -499,11 +485,13 @@ impl AppState {
))
}
/// Single-mode shared construction: wraps the bare engine + per-graph
/// policy in a `GraphHandle` carried directly by `GraphRouting::Single`.
/// Per-graph policy enforcement on the engine (MR-722) is re-applied
/// via `Omnigraph::with_policy` so HTTP and engine layers can never
/// diverge.
/// Single-graph convenience construction (RFC-011 cluster-only):
/// wraps the bare engine + per-graph policy in a `GraphHandle` keyed
/// by `default`, then builds a one-graph registry so the deployment
/// serves the same `/graphs/{graph_id}/...` cluster routes as any
/// other. Per-graph policy enforcement on the engine (MR-722) is
/// re-applied via `Omnigraph::with_policy` so HTTP and engine layers
/// can never diverge.
fn build_single_mode(
uri: String,
db: Omnigraph,
@ -522,18 +510,13 @@ impl AppState {
} else {
db
};
// `GraphHandle.key` is required by the struct, but in single
// mode it is never a registry key (there's no registry) and
// never compared against user input (routes are flat, no
// `{graph_id}` parameter). The label appears only in tracing
// output from `resolve_graph_handle`. The literal below is a
// log label, not a routing key — when the future cluster
// catalog ships, single mode may carry the catalog-assigned
// id here instead.
// The convenience constructors address the single graph by the
// reserved id `default` — both the registry key and the URL
// segment (`/graphs/default/...`).
let uri = normalize_root_uri(&uri).unwrap_or(uri);
let key = GraphKey::cluster(
GraphId::try_from("default").expect("'default' is a valid GraphId log label"),
);
let graph_id =
GraphId::try_from("default").expect("'default' is a valid GraphId");
let key = GraphKey::cluster(graph_id);
let handle = Arc::new(GraphHandle {
key,
uri,
@ -541,8 +524,15 @@ impl AppState {
policy: policy_engine,
queries,
});
let registry = Arc::new(
GraphRegistry::from_handles(vec![handle])
.expect("a single handle never collides on graph id"),
);
Self {
routing: GraphRouting::Single { handle },
routing: GraphRouting {
registry,
config_path: None,
},
workload,
bearer_tokens,
server_policy: None,
@ -566,7 +556,7 @@ impl AppState {
let bearer_tokens = hash_bearer_tokens(bearer_tokens);
let registry = Arc::new(GraphRegistry::from_handles(handles)?);
Ok(Self {
routing: GraphRouting::Multi {
routing: GraphRouting {
registry,
config_path,
},
@ -578,9 +568,7 @@ impl AppState {
/// Runtime routing accessor. Handlers don't typically inspect this —
/// they extract `Arc<GraphHandle>` via the routing middleware — but
/// `build_app` matches on it to decide flat vs nested route
/// mounting, and a handful of management endpoints (`GET /graphs`,
/// the OpenAPI cluster rewrite) match on the discriminant.
/// `server_graphs_list` reads the registry through it.
pub fn routing(&self) -> &GraphRouting {
&self.routing
}
@ -594,13 +582,9 @@ impl AppState {
}
// Any per-graph policy also requires auth — otherwise the
// policy gate would receive unauthenticated requests. Reading
// from `routing` is O(1) in both arms: single mode is a direct
// `handle.policy.is_some()` check, multi mode reads the
// cached `any_per_graph_policy` flag on the registry snapshot.
match &self.routing {
GraphRouting::Single { handle } => handle.policy.is_some(),
GraphRouting::Multi { registry, .. } => registry.snapshot_ref().any_per_graph_policy,
}
// the cached `any_per_graph_policy` flag off the registry
// snapshot is O(1).
self.routing.registry.snapshot_ref().any_per_graph_policy
}
fn authenticate_bearer_token(&self, provided_token: &str) -> Option<ResolvedActor> {
@ -972,13 +956,9 @@ pub fn build_app(state: AppState) -> Router {
// Management endpoints (`GET /graphs`) live alongside the per-graph
// router. They go through bearer auth but NOT through
// `resolve_graph_handle` — they operate on the registry directly.
// The endpoint is mounted in both modes; in single mode the handler
// returns 405 so clients see "resource exists, wrong context"
// rather than 404 "no such resource."
//
// Runtime add/remove (`POST /graphs`, `DELETE /graphs/{id}`) is not
// exposed in v0.6.0 — operators add graphs by editing
// `omnigraph.yaml` and restarting.
// exposed — operators run `cluster apply` and restart.
let management = Router::new()
.route("/graphs", get(server_graphs_list))
.route_layer(middleware::from_fn_with_state(
@ -986,15 +966,11 @@ pub fn build_app(state: AppState) -> Router {
require_bearer_auth,
));
// Mount the protected routes differently per mode:
// * Single → flat routes (legacy: `/snapshot`, `/read`, etc.)
// * Multi → nested under `/graphs/{graph_id}/...`
let protected: Router<AppState> = match state.routing() {
GraphRouting::Single { .. } => per_graph_protected.merge(management),
GraphRouting::Multi { .. } => Router::new()
.nest("/graphs/{graph_id}", per_graph_protected)
.merge(management),
};
// RFC-011 cluster-only: per-graph routes always nest under
// `/graphs/{graph_id}/...`; there are no flat single-graph routes.
let protected: Router<AppState> = Router::new()
.nest("/graphs/{graph_id}", per_graph_protected)
.merge(management);
Router::new()
.route("/healthz", get(server_health))
@ -1015,7 +991,6 @@ pub async fn serve(config: ServerConfig) -> Result<()> {
// policy OR any per-graph policy file. Mirrors the
// `requires_bearer_auth` semantics on AppState.
let has_policy_configured = match &config.mode {
ServerConfigMode::Single { policy_file, .. } => policy_file.is_some(),
ServerConfigMode::Multi {
graphs,
server_policy,
@ -1043,29 +1018,6 @@ pub async fn serve(config: ServerConfig) -> Result<()> {
let bind = config.bind.clone();
let state = match config.mode {
ServerConfigMode::Single {
uri,
graph_id,
policy_file,
queries,
} => {
let uri_for_log = uri.clone();
info!(
uri = %uri_for_log,
graph_id = %graph_id,
bind = %bind,
mode = "single",
"serving omnigraph"
);
AppState::open_single_with_queries_for_graph_id(
uri,
tokens,
policy_file.as_ref(),
queries,
Some(graph_id),
)
.await?
}
ServerConfigMode::Multi {
graphs,
config_path,
@ -1073,7 +1025,7 @@ pub async fn serve(config: ServerConfig) -> Result<()> {
} => {
info!(
bind = %bind,
mode = "multi",
mode = "cluster",
graph_count = graphs.len(),
config = %config_path.display(),
"serving omnigraph"

19
crates/omnigraph-server/src/main.rs
View file

@ -8,16 +8,10 @@ use omnigraph_server::{ServerConfig, init_tracing, load_server_settings, serve};
#[command(name = "omnigraph-server")]
#[command(about = "HTTP server for the Omnigraph graph database")]
struct Cli {
/// Graph URI
uri: Option<String>,
#[arg(long)]
target: Option<String>,
#[arg(long)]
config: Option<PathBuf>,
/// Boot from a cluster: either a config directory (storage resolved
/// through cluster.yaml) or a storage-root URI directly
/// (s3://bucket/prefix — config-free serving from the bucket).
/// Exclusive: cannot combine with <URI>, --target, or --config.
/// The server's only boot source (RFC-011 cluster-only).
#[arg(long)]
cluster: Option<PathBuf>,
#[arg(long)]
@ -36,14 +30,7 @@ async fn main() -> Result<()> {
init_tracing();
let cli = Cli::parse();
let settings: ServerConfig = load_server_settings(
cli.config.as_ref(),
cli.cluster.as_ref(),
cli.uri,
cli.target,
cli.bind,
cli.unauthenticated,
)
.await?;
let settings: ServerConfig =
load_server_settings(cli.cluster.as_ref(), cli.bind, cli.unauthenticated).await?;
serve(settings).await
}

364
crates/omnigraph-server/src/settings.rs
View file

@ -122,162 +122,24 @@ pub(crate) async fn load_cluster_settings(
})
}
/// RFC-011 cluster-only boot: the server serves exclusively from a
/// cluster's applied revision (`--cluster <dir | s3://…>`). The legacy
/// omnigraph.yaml / `--target` / positional-URI single-graph boot paths
/// were removed — a deployment serves from exactly one source.
pub async fn load_server_settings(
config_path: Option<&PathBuf>,
cli_cluster: Option<&PathBuf>,
cli_uri: Option<String>,
cli_target: Option<String>,
cli_bind: Option<String>,
cli_allow_unauthenticated: bool,
) -> Result<ServerConfig> {
// Rule 0 (RFC-005): --cluster is an exclusive boot source. It is checked
// before anything reads omnigraph.yaml — in cluster mode that file is
// never opened, not even the implicit current-directory search.
if let Some(cluster_dir) = cli_cluster {
if cli_uri.is_some() || cli_target.is_some() || config_path.is_some() {
bail!(
"--cluster is an exclusive boot source; it cannot combine with a graph URI, --target, or --config (axiom 15: a deployment serves from one source)"
);
}
return load_cluster_settings(cluster_dir, cli_bind, cli_allow_unauthenticated).await;
}
let config = load_config(config_path)?;
let bind = cli_bind.unwrap_or_else(|| config.server_bind().to_string());
// Either `--unauthenticated` or `OMNIGRAPH_UNAUTHENTICATED=1` flips
// this. Treat any non-empty, non-"0"/"false" string as truthy —
// standard 12-factor "any value is true" reading of the env var.
let env_unauth = std::env::var("OMNIGRAPH_UNAUTHENTICATED")
.ok()
.map(|v| {
let trimmed = v.trim();
!trimmed.is_empty() && trimmed != "0" && !trimmed.eq_ignore_ascii_case("false")
})
.unwrap_or(false);
let allow_unauthenticated = cli_allow_unauthenticated || env_unauth;
// MR-668 decision 2 — four-rule mode inference matrix.
//
// 1. CLI `<URI>` positional → Single (URI = the value)
// 2. CLI `--target <name>` → Single (URI = graphs.<name>.uri)
// 3. `server.graph` in config → Single (URI = graphs.<server.graph>.uri)
// 4. `--config` + non-empty `graphs:` + no single-mode selector
// → Multi (every entry in `graphs:`)
// 5. otherwise → error with migration hint
//
// Rules 1-3 are mutually compatible (CLI URI wins over `--target`
// wins over `server.graph`), reusing the existing
// `resolve_target_uri` precedence.
let has_cli_uri = cli_uri.is_some();
let has_cli_target = cli_target.is_some();
let has_server_graph = config.server_graph_name().is_some();
let has_graphs_map = !config.graphs.is_empty();
let has_explicit_config = config_path.is_some();
let mode = if has_cli_uri || has_cli_target || has_server_graph {
// Rules 1, 2, or 3 → Single mode.
let raw_uri = config.resolve_target_uri(
cli_uri,
cli_target.as_deref(),
config.server_graph_name(),
)?;
let uri = normalize_root_uri(&raw_uri).wrap_err_with(|| {
format!("normalize single-graph URI '{raw_uri}' from server settings")
})?;
// Config follows graph IDENTITY, not mode: a bare URI is anonymous
// (top-level config); a graph chosen by name uses its per-graph
// `graphs.<name>.{policy,queries}`. `resolve_target_uri` already
// errored on an unknown name, so a `Some(name)` here is a known graph.
let selected: Option<&str> = if has_cli_uri {
None
} else {
cli_target.as_deref().or_else(|| config.server_graph_name())
};
// A named selection must not leave a populated top-level block
// silently unused — refuse boot and point at the per-graph block. The
// same rule the CLI selection gate enforces, shared via one helper so
// the boot check and `omnigraph queries validate`/`list` can't drift.
config.ensure_top_level_blocks_honored(selected)?;
// Load + identity-check now (no engine needed); the schema
// type-check happens when the engine opens.
let policy_file = config.resolve_policy_file_for(selected);
let queries = QueryRegistry::load(&config, config.query_entries_for(selected))
.map_err(|errs| color_eyre::eyre::eyre!(format_registry_load_errors(&uri, &errs)))?;
let graph_id = graph_resource_id_for_selection(selected, &uri);
ServerConfigMode::Single {
uri,
graph_id,
policy_file,
queries,
}
} else if has_explicit_config && has_graphs_map {
// Multi mode: every graph uses its per-graph block; top-level
// policy/queries are never honored, so a populated one is an error.
let unhonored = config.populated_top_level_blocks();
if !unhonored.is_empty() {
bail!(
"multi-graph mode: top-level {} {} not honored — each graph uses its own \
`graphs.<graph_id>.…` block. Move per-graph rules there (and any \
`graph_list` policy to `server.policy.file`).",
unhonored.join(" and "),
if unhonored.len() == 1 { "is" } else { "are" },
);
}
// Rule 4 → Multi mode. Build a startup config per graph.
let mut graphs = Vec::with_capacity(config.graphs.len());
for (name, target) in &config.graphs {
// Validate the graph id can construct a `GraphId` newtype.
// Doing this here (not at registry insert) so a malformed
// omnigraph.yaml fails at startup with a clear error.
GraphId::try_from(name.clone()).map_err(|err| {
color_eyre::eyre::eyre!("invalid graph id '{name}' in omnigraph.yaml: {err}")
})?;
let raw_uri = config.resolve_uri_value(&target.uri);
let uri = normalize_root_uri(&raw_uri).wrap_err_with(|| {
format!("normalize URI '{raw_uri}' for graph '{name}' in omnigraph.yaml")
})?;
// Per-graph `queries:`, selected through the shared
// `query_entries_for` so server and CLI resolve identically.
// Load + identity-check now; the schema type-check happens
// when this graph's engine opens.
let queries = QueryRegistry::load(&config, config.query_entries_for(Some(name.as_str())))
.map_err(|errs| color_eyre::eyre::eyre!(format_registry_load_errors(name, &errs)))?;
graphs.push(GraphStartupConfig {
graph_id: name.clone(),
uri,
policy: config.resolve_target_policy_file(name).map(PolicySource::File),
queries,
});
}
let config_path = config_path
.cloned()
.expect("has_explicit_config implies config_path is Some");
let server_policy = config.resolve_server_policy_file().map(PolicySource::File);
ServerConfigMode::Multi {
graphs,
config_path,
server_policy,
}
} else {
// Rule 5 → error with migration hint.
let Some(cluster_dir) = cli_cluster else {
bail!(
"no graph to serve: pass a URI (`omnigraph-server <URI>`), select a target \
(`--target <name> --config omnigraph.yaml`), set `server.graph: <name>` in \
omnigraph.yaml, or for multi-graph mode add a `graphs:` map to the config \
file referenced by `--config`."
"omnigraph-server boots from a cluster: pass --cluster <dir|s3://…> \
(the cluster's applied revision is the deployment artifact). The legacy \
single-graph boot (positional <URI>, --target, --config omnigraph.yaml) \
was removed in RFC-011."
);
};
Ok(ServerConfig {
mode,
bind,
allow_unauthenticated,
})
}
/// Whether the loaded config will run the server in multi-graph mode.
/// Useful for the test that constructs `ServerConfig` directly.
pub fn server_config_is_multi(config: &ServerConfig) -> bool {
matches!(config.mode, ServerConfigMode::Multi { .. })
load_cluster_settings(cluster_dir, cli_bind, cli_allow_unauthenticated).await
}
/// MR-723 server runtime state, classified from the three-state matrix
@ -417,8 +279,8 @@ pub(crate) fn server_bearer_tokens_from_env() -> Result<Vec<(String, String)>> {
mod tests {
use super::{
GraphStartupConfig, ServerConfig, ServerConfigMode, ServerRuntimeState,
classify_server_runtime_state, hash_bearer_token, load_server_settings,
normalize_bearer_token, parse_bearer_tokens_json, serve, server_bearer_tokens_from_env,
classify_server_runtime_state, hash_bearer_token, normalize_bearer_token,
parse_bearer_tokens_json, serve, server_bearer_tokens_from_env,
};
use serial_test::serial;
use std::env;
@ -577,108 +439,15 @@ mod tests {
}
#[tokio::test]
async fn server_settings_load_from_yaml_config() {
let temp = tempdir().unwrap();
let config = temp.path().join("omnigraph.yaml");
fs::write(
&config,
r#"
graphs:
local:
uri: /tmp/demo.omni
server:
graph: local
bind: 0.0.0.0:9090
"#,
)
.unwrap();
let settings = load_server_settings(Some(&config), None, None, None, None, false).await.unwrap();
match &settings.mode {
ServerConfigMode::Single { uri, graph_id, .. } => {
assert_eq!(uri, "/tmp/demo.omni");
assert_eq!(graph_id, "local");
}
ServerConfigMode::Multi { .. } => panic!("expected Single mode, got Multi"),
}
assert_eq!(settings.bind, "0.0.0.0:9090");
}
#[tokio::test]
async fn server_settings_cli_flags_override_yaml_config() {
let temp = tempdir().unwrap();
let config = temp.path().join("omnigraph.yaml");
fs::write(
&config,
r#"
graphs:
local:
uri: /tmp/demo.omni
server:
graph: local
bind: 127.0.0.1:8080
"#,
)
.unwrap();
let settings = load_server_settings(
Some(&config),
None,
Some("/tmp/override.omni".to_string()),
None,
Some("0.0.0.0:9999".to_string()),
false,
)
.await
.unwrap();
match &settings.mode {
ServerConfigMode::Single { uri, graph_id, .. } => {
assert_eq!(uri, "/tmp/override.omni");
assert_eq!(graph_id, "/tmp/override.omni");
}
ServerConfigMode::Multi { .. } => panic!("expected Single mode, got Multi"),
}
assert_eq!(settings.bind, "0.0.0.0:9999");
}
#[tokio::test]
async fn server_settings_can_resolve_named_target() {
let temp = tempdir().unwrap();
let config = temp.path().join("omnigraph.yaml");
fs::write(
&config,
r#"
graphs:
local:
uri: ./demo.omni
dev:
uri: http://127.0.0.1:8080
server:
graph: local
bind: 127.0.0.1:8080
"#,
)
.unwrap();
let settings =
load_server_settings(Some(&config), None, None, Some("dev".to_string()), None, false)
.await
.unwrap();
match &settings.mode {
ServerConfigMode::Single { uri, graph_id, .. } => {
assert_eq!(uri, "http://127.0.0.1:8080");
assert_eq!(graph_id, "dev");
}
ServerConfigMode::Multi { .. } => panic!("expected Single mode, got Multi"),
}
}
#[tokio::test]
async fn server_settings_require_uri_from_cli_or_config() {
let error = load_server_settings(None, None, None, None, None, false).await.unwrap_err();
async fn server_settings_require_cluster_boot_source() {
// RFC-011 cluster-only: with no --cluster the server refuses to
// start and names the cluster-required remedy.
let error = super::load_server_settings(None, None, false)
.await
.unwrap_err();
assert!(
error.to_string().contains("no graph to serve"),
"expected mode-inference error, got: {error}",
error.to_string().contains("boots from a cluster"),
"expected cluster-required error, got: {error}",
);
}
@ -788,17 +557,21 @@ server:
]);
let temp = tempdir().unwrap();
// Graph path doesn't need to exist — classifier fires before
// `AppState::open_with_bearer_tokens_and_policy`.
// any engine open.
let config = ServerConfig {
mode: ServerConfigMode::Single {
uri: temp
.path()
.join("graph.omni")
.to_string_lossy()
.into_owned(),
graph_id: "default".to_string(),
policy_file: None,
queries: crate::queries::QueryRegistry::default(),
mode: ServerConfigMode::Multi {
graphs: vec![GraphStartupConfig {
graph_id: "default".to_string(),
uri: temp
.path()
.join("graph.omni")
.to_string_lossy()
.into_owned(),
policy: None,
queries: crate::queries::QueryRegistry::default(),
}],
config_path: temp.path().join("cluster"),
server_policy: None,
},
bind: "127.0.0.1:0".to_string(),
allow_unauthenticated: false,
@ -813,75 +586,6 @@ server:
);
}
#[tokio::test]
#[serial]
async fn unauthenticated_env_var_classification() {
// MR-723 PR A: closes the gap where the env-var read path inside
// `load_server_settings` was structurally implemented but not
// exercised by any test. Three properties to pin, all in one
// sequential test because `cargo test` runs the mod test suite
// in parallel and `OMNIGRAPH_UNAUTHENTICATED` is process-global
// — interleaving with another test that sets the same env var
// (concurrent classifier tests, even the bearer-token suite
// sharing `EnvGuard`) corrupts the read. Sequential within one
// test fn is the simplest race-free shape.
let temp = tempdir().unwrap();
let config_path = temp.path().join("omnigraph.yaml");
fs::write(
&config_path,
r#"
graphs:
local:
uri: /tmp/demo-unauth.omni
server:
graph: local
"#,
)
.unwrap();
// Truthy values flip Open mode on, even with CLI flag off.
for value in ["1", "true", "yes", "TRUE", "anything"] {
let _guard = EnvGuard::set(&[("OMNIGRAPH_UNAUTHENTICATED", Some(value))]);
let settings = load_server_settings(Some(&config_path), None, None, None, None, false).await
.expect("settings load should succeed");
assert!(
settings.allow_unauthenticated,
"OMNIGRAPH_UNAUTHENTICATED={value:?} should enable Open mode",
);
}
// Falsy values keep refusal behavior, even with CLI flag off.
for value in ["0", "false", "FALSE", ""] {
let _guard = EnvGuard::set(&[("OMNIGRAPH_UNAUTHENTICATED", Some(value))]);
let settings = load_server_settings(Some(&config_path), None, None, None, None, false).await
.expect("settings load should succeed");
assert!(
!settings.allow_unauthenticated,
"OMNIGRAPH_UNAUTHENTICATED={value:?} should NOT enable Open mode",
);
}
// Unset env var: also false.
let _guard = EnvGuard::set(&[("OMNIGRAPH_UNAUTHENTICATED", None)]);
let settings = load_server_settings(Some(&config_path), None, None, None, None, false).await
.expect("settings load should succeed");
assert!(
!settings.allow_unauthenticated,
"OMNIGRAPH_UNAUTHENTICATED unset should NOT enable Open mode",
);
drop(_guard);
// CLI flag wins even when env is falsy — `serve()` honors the
// OR of both inputs.
let _guard = EnvGuard::set(&[("OMNIGRAPH_UNAUTHENTICATED", Some("0"))]);
let settings = load_server_settings(Some(&config_path), None, None, None, None, true).await
.expect("settings load should succeed");
assert!(
settings.allow_unauthenticated,
"--unauthenticated CLI flag should win even when env is falsy",
);
}
#[test]
fn classify_policy_enabled_requires_tokens() {
// State 3: tokens + policy → PolicyEnabled, regardless of the

68
crates/omnigraph-server/tests/auth_policy.rs
View file

@ -50,7 +50,7 @@ async fn protected_routes_require_bearer_token() {
let (status, body) = json_response(
&app,
Request::builder()
.uri("/branches")
.uri(g("/branches"))
.method(Method::GET)
.body(Body::empty())
.unwrap(),
@ -85,7 +85,7 @@ async fn protected_routes_accept_valid_bearer_token_while_healthz_stays_open() {
let (status, body) = json_response(
&app,
Request::builder()
.uri("/branches")
.uri(g("/branches"))
.method(Method::GET)
.header("authorization", "Bearer demo-token")
.body(Body::empty())
@ -108,7 +108,7 @@ async fn protected_routes_accept_any_configured_team_bearer_token() {
let (status, body) = json_response(
&app,
Request::builder()
.uri("/branches")
.uri(g("/branches"))
.method(Method::GET)
.header("authorization", "Bearer token-two")
.body(Body::empty())
@ -158,7 +158,7 @@ rules:
let (ok_status, _) = json_response(
&app,
Request::builder()
.uri("/snapshot?branch=main")
.uri(g("/snapshot?branch=main"))
.method(Method::GET)
.header("authorization", "Bearer token-a")
.body(Body::empty())
@ -172,7 +172,7 @@ rules:
let (denied_status, denied_body) = json_response(
&app,
Request::builder()
.uri("/snapshot?branch=main")
.uri(g("/snapshot?branch=main"))
.method(Method::GET)
.header("authorization", "Bearer token-b")
.body(Body::empty())
@ -190,7 +190,7 @@ rules:
let (bad_status, _) = json_response(
&app,
Request::builder()
.uri("/snapshot?branch=main")
.uri(g("/snapshot?branch=main"))
.method(Method::GET)
.header("authorization", "Bearer wrong-token")
.body(Body::empty())
@ -245,7 +245,7 @@ rules:
let (spoof_up_status, spoof_up_body) = json_response(
&app,
Request::builder()
.uri("/snapshot?branch=main")
.uri(g("/snapshot?branch=main"))
.method(Method::GET)
.header("authorization", "Bearer token-b")
.header("x-actor-id", "act-a")
@ -270,7 +270,7 @@ rules:
let (spoof_down_status, _) = json_response(
&app,
Request::builder()
.uri("/snapshot?branch=main")
.uri(g("/snapshot?branch=main"))
.method(Method::GET)
.header("authorization", "Bearer token-a")
.header("x-actor-id", "act-b")
@ -290,7 +290,7 @@ rules:
let (empty_spoof_status, _) = json_response(
&app,
Request::builder()
.uri("/snapshot?branch=main")
.uri(g("/snapshot?branch=main"))
.method(Method::GET)
.header("authorization", "Bearer token-b")
.header("x-actor-id", "")
@ -316,7 +316,7 @@ async fn policy_allows_read_but_distinguishes_401_from_403() {
let (missing_status, missing_body) = json_response(
&app,
Request::builder()
.uri("/snapshot?branch=main")
.uri(g("/snapshot?branch=main"))
.method(Method::GET)
.body(Body::empty())
.unwrap(),
@ -332,7 +332,7 @@ async fn policy_allows_read_but_distinguishes_401_from_403() {
let (snapshot_status, snapshot_body) = json_response(
&app,
Request::builder()
.uri("/snapshot?branch=main")
.uri(g("/snapshot?branch=main"))
.method(Method::GET)
.header("authorization", "Bearer team-token")
.body(Body::empty())
@ -350,7 +350,7 @@ async fn policy_allows_read_but_distinguishes_401_from_403() {
let (forbidden_status, forbidden_body) = json_response(
&app,
Request::builder()
.uri("/export")
.uri(g("/export"))
.method(Method::POST)
.header("authorization", "Bearer team-token")
.header("content-type", "application/json")
@ -369,7 +369,7 @@ async fn policy_allows_read_but_distinguishes_401_from_403() {
.clone()
.oneshot(
Request::builder()
.uri("/export")
.uri(g("/export"))
.method(Method::POST)
.header("authorization", "Bearer admin-token")
.header("content-type", "application/json")
@ -410,7 +410,7 @@ async fn policy_uses_resolved_branch_for_snapshot_reads() {
let (status, body) = json_response(
&app,
Request::builder()
.uri("/read")
.uri(g("/read"))
.method(Method::POST)
.header("authorization", "Bearer team-token")
.header("content-type", "application/json")
@ -458,7 +458,7 @@ async fn policy_blocks_change_on_protected_main_but_allows_unprotected_branch()
let (main_status, main_body) = json_response(
&app,
Request::builder()
.uri("/change")
.uri(g("/change"))
.method(Method::POST)
.header("authorization", "Bearer team-token")
.header("content-type", "application/json")
@ -482,7 +482,7 @@ async fn policy_blocks_change_on_protected_main_but_allows_unprotected_branch()
let (feature_status, feature_body) = json_response(
&app,
Request::builder()
.uri("/change")
.uri(g("/change"))
.method(Method::POST)
.header("authorization", "Bearer team-token")
.header("content-type", "application/json")
@ -533,7 +533,7 @@ async fn policy_blocks_non_admin_merge_to_main_and_allows_admin() {
let (deny_status, deny_body) = json_response(
&app,
Request::builder()
.uri("/branches/merge")
.uri(g("/branches/merge"))
.method(Method::POST)
.header("authorization", "Bearer team-token")
.header("content-type", "application/json")
@ -551,7 +551,7 @@ async fn policy_blocks_non_admin_merge_to_main_and_allows_admin() {
let (allow_status, allow_body) = json_response(
&app,
Request::builder()
.uri("/branches/merge")
.uri(g("/branches/merge"))
.method(Method::POST)
.header("authorization", "Bearer admin-token")
.header("content-type", "application/json")
@ -578,7 +578,7 @@ async fn authenticated_change_stamps_actor_on_commits() {
let (change_status, change_body) = json_response(
&app,
Request::builder()
.uri("/change")
.uri(g("/change"))
.method(Method::POST)
.header("authorization", "Bearer token-one")
.header("content-type", "application/json")
@ -592,7 +592,7 @@ async fn authenticated_change_stamps_actor_on_commits() {
let (commits_status, commits_body) = json_response(
&app,
Request::builder()
.uri("/commits?branch=main")
.uri(g("/commits?branch=main"))
.method(Method::GET)
.header("authorization", "Bearer token-one")
.body(Body::empty())
@ -623,7 +623,7 @@ async fn authenticated_branch_merge_stamps_merge_actor_on_head_commit() {
let (create_status, _) = json_response(
&app,
Request::builder()
.uri("/branches")
.uri(g("/branches"))
.method(Method::POST)
.header("authorization", "Bearer token-one")
.header("content-type", "application/json")
@ -642,7 +642,7 @@ async fn authenticated_branch_merge_stamps_merge_actor_on_head_commit() {
let (change_status, _) = json_response(
&app,
Request::builder()
.uri("/change")
.uri(g("/change"))
.method(Method::POST)
.header("authorization", "Bearer token-one")
.header("content-type", "application/json")
@ -659,7 +659,7 @@ async fn authenticated_branch_merge_stamps_merge_actor_on_head_commit() {
let (merge_status, merge_body) = json_response(
&app,
Request::builder()
.uri("/branches/merge")
.uri(g("/branches/merge"))
.method(Method::POST)
.header("authorization", "Bearer token-two")
.header("content-type", "application/json")
@ -673,7 +673,7 @@ async fn authenticated_branch_merge_stamps_merge_actor_on_head_commit() {
let (commit_status, commit_body) = json_response(
&app,
Request::builder()
.uri("/commits?branch=main")
.uri(g("/commits?branch=main"))
.method(Method::GET)
.header("authorization", "Bearer token-two")
.body(Body::empty())
@ -691,7 +691,6 @@ async fn authenticated_branch_merge_stamps_merge_actor_on_head_commit() {
#[tokio::test(flavor = "multi_thread")]
async fn engine_layer_policy_fires_via_direct_arc_omnigraph_from_new_single() {
use omnigraph_server::GraphRouting;
let temp = init_loaded_graph().await;
let graph = graph_path(temp.path());
let db = Omnigraph::open(graph.to_str().unwrap()).await.unwrap();
@ -717,9 +716,14 @@ async fn engine_layer_policy_fires_via_direct_arc_omnigraph_from_new_single() {
// embedded consumer holding `Arc<Omnigraph>` would. If `new_single`
// failed to apply `with_policy` to the engine, this `mutate_as`
// would succeed — the HTTP-layer is bypassed entirely.
let handle = match state.routing() {
GraphRouting::Single { handle } => Arc::clone(handle),
GraphRouting::Multi { .. } => panic!("expected single-mode routing"),
// RFC-011 cluster-only: the single-graph convenience constructor
// registers the graph under the reserved id `default`.
let key = omnigraph_server::GraphKey::cluster(
omnigraph_server::GraphId::try_from("default").unwrap(),
);
let handle = match state.routing().registry.get(&key) {
omnigraph_server::RegistryLookup::Ready(handle) => handle,
omnigraph_server::RegistryLookup::Gone => panic!("default graph must be registered"),
};
let engine = Arc::clone(&handle.engine);
@ -758,7 +762,7 @@ async fn oversized_request_body_returns_payload_too_large() {
.clone()
.oneshot(
Request::builder()
.uri("/read")
.uri(g("/read"))
.method(Method::POST)
.header("content-type", "application/json")
.body(Body::from(oversized))
@ -781,7 +785,7 @@ async fn default_deny_mode_allows_read_for_authenticated_actor() {
let (status, _body) = json_response(
&app,
Request::builder()
.uri("/snapshot")
.uri(g("/snapshot"))
.method(Method::GET)
.header(AUTHORIZATION, "Bearer demo-token")
.body(Body::empty())
@ -808,7 +812,7 @@ async fn default_deny_mode_rejects_change_with_forbidden() {
let (status, body) = json_response(
&app,
Request::builder()
.uri("/change")
.uri(g("/change"))
.method(Method::POST)
.header(AUTHORIZATION, "Bearer demo-token")
.header("content-type", "application/json")
@ -840,7 +844,7 @@ async fn default_deny_mode_rejects_schema_apply_with_forbidden() {
let (status, body) = json_response(
&app,
Request::builder()
.uri("/schema/apply")
.uri(g("/schema/apply"))
.method(Method::POST)
.header(AUTHORIZATION, "Bearer demo-token")
.header("content-type", "application/json")

470
crates/omnigraph-server/tests/boot_settings.rs
View file

@ -18,10 +18,7 @@ use support::*;
mod multi_graph_startup {
use super::*;
use omnigraph::storage::normalize_root_uri;
use omnigraph_server::{
GraphHandle, GraphId, GraphKey, GraphRegistry, InsertError, ServerConfig, ServerConfigMode,
load_server_settings,
};
use omnigraph_server::{GraphHandle, GraphId, GraphKey, GraphRegistry, InsertError};
use std::sync::Arc;
async fn build_multi_mode_app(graph_ids: &[&str]) -> (Vec<tempfile::TempDir>, Router) {
@ -280,10 +277,11 @@ mod multi_graph_startup {
);
}
/// Flat routes 404 in multi mode — the router only mounts under
/// `/graphs/{graph_id}/...` so `/snapshot` doesn't resolve.
/// RFC-011 cluster-only: flat per-graph routes never resolve — the
/// router only mounts under `/graphs/{graph_id}/...` so a root
/// `/snapshot` returns 404.
#[tokio::test(flavor = "multi_thread")]
async fn flat_routes_404_in_multi_mode() {
async fn flat_routes_404_at_root() {
let (_dirs, app) = build_multi_mode_app(&["alpha"]).await;
let resp = app
.oneshot(
@ -298,28 +296,6 @@ mod multi_graph_startup {
assert_eq!(resp.status(), StatusCode::NOT_FOUND);
}
/// `GraphId` validation runs at startup — a reserved name in
/// `omnigraph.yaml` produces a clear error rather than getting
/// rejected per-request.
#[tokio::test]
async fn load_server_settings_rejects_reserved_graph_id() {
let temp = tempfile::tempdir().unwrap();
let config_path = temp.path().join("omnigraph.yaml");
fs::write(
&config_path,
r#"
graphs:
policies:
uri: /tmp/g1.omni
"#,
)
.unwrap();
let err = load_server_settings(Some(&config_path), None, None, None, None, false).await.unwrap_err();
assert!(
err.to_string().contains("invalid graph id 'policies'"),
"expected reserved-name rejection, got: {err}"
);
}
#[tokio::test(flavor = "multi_thread")]
async fn registry_rejects_duplicate_normalized_graph_uris() {
@ -375,372 +351,6 @@ graphs:
assert_eq!(listed[0].uri, graph_uri);
}
// ── Four-rule mode inference matrix ───────────────────────────────
/// Rule 1: CLI positional URI → Single.
#[tokio::test]
async fn mode_inference_cli_uri_is_single() {
let settings = load_server_settings(
None,
None,
Some("/tmp/cli.omni".to_string()),
None,
None,
true, // allow unauth so we get past the runtime-state check
)
.await
.unwrap();
match settings.mode {
ServerConfigMode::Single { uri, .. } => assert_eq!(uri, "/tmp/cli.omni"),
ServerConfigMode::Multi { .. } => panic!("expected Single (rule 1), got Multi"),
}
}
/// Rule 2: --target picks one graph from `graphs:` map → Single.
#[tokio::test]
async fn mode_inference_cli_target_is_single() {
let temp = tempfile::tempdir().unwrap();
let config_path = temp.path().join("omnigraph.yaml");
fs::write(
&config_path,
r#"
graphs:
alpha:
uri: /tmp/alpha.omni
beta:
uri: /tmp/beta.omni
"#,
)
.unwrap();
let settings =
load_server_settings(Some(&config_path), None, None, Some("alpha".into()), None, true)
.await
.unwrap();
match settings.mode {
ServerConfigMode::Single { uri, .. } => assert_eq!(uri, "/tmp/alpha.omni"),
ServerConfigMode::Multi { .. } => panic!("expected Single (rule 2), got Multi"),
}
}
/// Rule 3: `server.graph` set → Single (target picked from config).
#[tokio::test]
async fn mode_inference_server_graph_is_single() {
let temp = tempfile::tempdir().unwrap();
let config_path = temp.path().join("omnigraph.yaml");
fs::write(
&config_path,
r#"
graphs:
alpha:
uri: /tmp/alpha.omni
beta:
uri: /tmp/beta.omni
server:
graph: beta
"#,
)
.unwrap();
let settings = load_server_settings(Some(&config_path), None, None, None, None, true).await.unwrap();
match settings.mode {
ServerConfigMode::Single { uri, .. } => assert_eq!(uri, "/tmp/beta.omni"),
ServerConfigMode::Multi { .. } => panic!("expected Single (rule 3), got Multi"),
}
}
/// Rule 4: `--config` + non-empty `graphs:` + no single-mode selector → Multi.
#[tokio::test]
async fn mode_inference_config_plus_graphs_is_multi() {
let temp = tempfile::tempdir().unwrap();
let config_path = temp.path().join("omnigraph.yaml");
fs::write(
&config_path,
r#"
graphs:
alpha:
uri: /tmp/alpha.omni
beta:
uri: /tmp/beta.omni
"#,
)
.unwrap();
let settings = load_server_settings(Some(&config_path), None, None, None, None, true).await.unwrap();
match settings.mode {
ServerConfigMode::Multi { graphs, .. } => {
let ids: Vec<&str> = graphs.iter().map(|g| g.graph_id.as_str()).collect();
// BTreeMap iteration order is alphabetical.
assert_eq!(ids, vec!["alpha", "beta"]);
}
ServerConfigMode::Single { .. } => panic!("expected Multi (rule 4), got Single"),
}
}
#[tokio::test]
async fn mode_inference_multi_rejects_top_level_policy_file() {
let temp = tempfile::tempdir().unwrap();
let config_path = temp.path().join("omnigraph.yaml");
fs::write(
&config_path,
r#"
policy:
file: ./policy.yaml
graphs:
alpha:
uri: /tmp/alpha.omni
"#,
)
.unwrap();
let err = load_server_settings(Some(&config_path), None, None, None, None, true).await.unwrap_err();
let msg = err.to_string();
assert!(
msg.contains("top-level") && msg.contains("policy.file") && msg.contains("not honored"),
"expected top-level-not-honored guidance, got: {msg}"
);
assert!(
msg.contains("graphs.<graph_id>"),
"expected per-graph migration guidance, got: {msg}"
);
assert!(
msg.contains("server.policy.file"),
"expected server policy migration guidance, got: {msg}"
);
}
#[tokio::test]
async fn mode_inference_multi_rejects_top_level_queries() {
// Symmetric to the policy guard: a top-level `queries:` block in
// multi-graph mode is not honored (each graph uses its own), so it
// is a loud error rather than a silent no-op.
let temp = tempfile::tempdir().unwrap();
let config_path = temp.path().join("omnigraph.yaml");
fs::write(
&config_path,
"queries:\n q:\n file: ./q.gq\ngraphs:\n alpha:\n uri: /tmp/alpha.omni\n",
)
.unwrap();
let err = load_server_settings(Some(&config_path), None, None, None, None, true).await.unwrap_err();
let msg = err.to_string();
assert!(
msg.contains("queries") && msg.contains("not honored"),
"top-level queries must be rejected in multi-graph mode: {msg}"
);
}
#[tokio::test]
async fn single_mode_named_graph_rejects_top_level_blocks() {
// Serving a graph by name (`--target`/`server.graph`) uses its
// per-graph block; a populated top-level block would be silently
// shadowed, so boot refuses and names the per-graph location.
let temp = tempfile::tempdir().unwrap();
let config_path = temp.path().join("omnigraph.yaml");
fs::write(
&config_path,
"policy:\n file: ./top.yaml\ngraphs:\n prod:\n uri: /tmp/prod.omni\n",
)
.unwrap();
let err =
load_server_settings(Some(&config_path), None, None, Some("prod".to_string()), None, true)
.await
.unwrap_err();
let msg = err.to_string();
assert!(
msg.contains("prod") && msg.contains("policy.file") && msg.contains("graphs.prod"),
"named single-mode + top-level policy must refuse, naming the graph: {msg}"
);
}
#[tokio::test]
async fn single_mode_named_graph_uses_per_graph_policy_and_queries() {
// The identity rule: `--target prod` attaches `graphs.prod`'s own
// policy + queries, not the top-level ones (which are absent here).
let temp = tempfile::tempdir().unwrap();
fs::write(
temp.path().join("prod.gq"),
"query pq() { match { $u: User } return { $u.name } }",
)
.unwrap();
let config_path = temp.path().join("omnigraph.yaml");
fs::write(
&config_path,
"graphs:\n prod:\n uri: /tmp/prod.omni\n policy:\n file: ./prod-policy.yaml\n \
queries:\n pq:\n file: ./prod.gq\n",
)
.unwrap();
let settings =
load_server_settings(Some(&config_path), None, None, Some("prod".to_string()), None, true)
.await
.unwrap();
match settings.mode {
ServerConfigMode::Single {
graph_id,
policy_file,
queries,
..
} => {
assert_eq!(graph_id, "prod", "named single-mode keeps graph identity");
assert!(
policy_file
.as_ref()
.is_some_and(|p| p.ends_with("prod-policy.yaml")),
"per-graph policy attached: {policy_file:?}"
);
assert!(queries.lookup("pq").is_some(), "per-graph query attached");
}
other => panic!("expected Single mode, got {other:?}"),
}
}
#[tokio::test]
async fn mode_inference_normalizes_multi_graph_uris() {
let temp = tempfile::tempdir().unwrap();
let graph = temp.path().join("alpha.omni");
let config_path = temp.path().join("omnigraph.yaml");
fs::write(
&config_path,
format!(
r#"
graphs:
alpha:
uri: file://{}/
"#,
graph.display()
),
)
.unwrap();
let settings = load_server_settings(Some(&config_path), None, None, None, None, true).await.unwrap();
match settings.mode {
ServerConfigMode::Multi { graphs, .. } => {
assert_eq!(graphs[0].uri, graph.to_string_lossy());
}
ServerConfigMode::Single { .. } => panic!("expected Multi"),
}
}
/// Rule 5: nothing → error with migration hint.
#[tokio::test]
async fn mode_inference_no_inputs_errors_with_migration_hint() {
let err = load_server_settings(None, None, None, None, None, true).await.unwrap_err();
let msg = err.to_string();
assert!(
msg.contains("no graph to serve"),
"expected migration-hint error, got: {msg}"
);
}
/// Rule 4 sub-case: `--config` with empty `graphs:` map and no
/// single-mode selector → rule 5 fires (no graph to serve).
#[tokio::test]
async fn mode_inference_empty_graphs_map_errors() {
let temp = tempfile::tempdir().unwrap();
let config_path = temp.path().join("omnigraph.yaml");
fs::write(&config_path, "server:\n bind: 127.0.0.1:8080\n").unwrap();
let err = load_server_settings(Some(&config_path), None, None, None, None, true).await.unwrap_err();
assert!(err.to_string().contains("no graph to serve"));
}
/// `--config` + `<URI>` together: URI wins → Single (the CLI URI
/// takes precedence over the config's graphs map).
#[tokio::test]
async fn mode_inference_cli_uri_overrides_graphs_map() {
let temp = tempfile::tempdir().unwrap();
let config_path = temp.path().join("omnigraph.yaml");
fs::write(
&config_path,
r#"
graphs:
alpha:
uri: /tmp/alpha.omni
"#,
)
.unwrap();
let settings = load_server_settings(
Some(&config_path),
None,
Some("/tmp/cli-override.omni".to_string()),
None,
None,
true,
)
.await
.unwrap();
match settings.mode {
ServerConfigMode::Single { uri, .. } => {
assert_eq!(
uri, "/tmp/cli-override.omni",
"CLI URI must win over graphs: map"
);
}
ServerConfigMode::Multi { .. } => {
panic!("expected Single (CLI URI wins), got Multi")
}
}
}
/// Per-graph `policy.file` is resolved relative to the config base_dir.
#[tokio::test]
async fn per_graph_policy_file_is_resolved_relative_to_base_dir() {
let temp = tempfile::tempdir().unwrap();
let config_path = temp.path().join("omnigraph.yaml");
fs::write(
&config_path,
r#"
graphs:
alpha:
uri: /tmp/alpha.omni
policy:
file: ./policies/alpha.yaml
beta:
uri: /tmp/beta.omni
"#,
)
.unwrap();
let settings = load_server_settings(Some(&config_path), None, None, None, None, true).await.unwrap();
let graphs = match settings.mode {
ServerConfigMode::Multi { graphs, .. } => graphs,
_ => panic!("expected Multi"),
};
// graphs is BTreeMap-iter order (alphabetical).
let alpha = &graphs[0];
let beta = &graphs[1];
assert_eq!(alpha.graph_id, "alpha");
let omnigraph_server::PolicySource::File(alpha_policy) =
alpha.policy.as_ref().unwrap()
else {
panic!("yaml-configured policy must stay file-based");
};
assert_eq!(alpha_policy, &temp.path().join("policies/alpha.yaml"));
assert_eq!(beta.graph_id, "beta");
assert!(beta.policy.is_none());
}
/// `server.policy.file` resolves alongside the graphs map.
#[tokio::test]
async fn server_policy_file_is_resolved_relative_to_base_dir() {
let temp = tempfile::tempdir().unwrap();
let config_path = temp.path().join("omnigraph.yaml");
fs::write(
&config_path,
r#"
server:
policy:
file: ./server-policy.yaml
graphs:
alpha:
uri: /tmp/alpha.omni
"#,
)
.unwrap();
let settings = load_server_settings(Some(&config_path), None, None, None, None, true).await.unwrap();
match settings.mode {
ServerConfigMode::Multi { server_policy, .. } => {
let omnigraph_server::PolicySource::File(path) = server_policy.unwrap() else {
panic!("yaml-configured server policy must stay file-based");
};
assert_eq!(path, temp.path().join("server-policy.yaml"));
}
_ => panic!("expected Multi"),
}
}
/// `GET /graphs` must NOT leak the registry in Open mode without
/// an explicit server policy. Operators who pass `--unauthenticated`
/// opted into trusting the network for graph DATA, not for leaking
@ -786,28 +396,6 @@ graphs:
);
}
/// `GET /graphs` returns 405 in single mode (resource exists in the
/// API surface, just not operational without a `graphs:` map).
#[tokio::test(flavor = "multi_thread")]
async fn get_graphs_returns_405_in_single_mode() {
let temp = init_loaded_graph().await;
let graph = graph_path(temp.path());
let state = AppState::open(graph.to_string_lossy().to_string())
.await
.unwrap();
let app = build_app(state);
let resp = app
.oneshot(
Request::builder()
.method(Method::GET)
.uri("/graphs")
.body(Body::empty())
.unwrap(),
)
.await
.unwrap();
assert_eq!(resp.status(), StatusCode::METHOD_NOT_ALLOWED);
}
/// `GET /graphs` requires bearer auth when tokens are configured.
#[tokio::test(flavor = "multi_thread")]
@ -971,52 +559,4 @@ rules:
);
}
/// Loads an `omnigraph.yaml` with two graphs and verifies multi-mode
/// inference plus graph entry resolution. Cluster-route dispatch is
/// covered by the route tests above.
#[tokio::test(flavor = "multi_thread")]
async fn server_settings_load_multi_graph_config_entries() {
let cfg_dir = tempfile::tempdir().unwrap();
// Real graph storage dirs (the URIs in the config must point to
// a graph init-able location).
let alpha_dir = cfg_dir.path().join("alpha.omni");
let beta_dir = cfg_dir.path().join("beta.omni");
let schema = fs::read_to_string(fixture("test.pg")).unwrap();
Omnigraph::init(alpha_dir.to_str().unwrap(), &schema)
.await
.unwrap();
Omnigraph::init(beta_dir.to_str().unwrap(), &schema)
.await
.unwrap();
let config_path = cfg_dir.path().join("omnigraph.yaml");
fs::write(
&config_path,
format!(
r#"
graphs:
alpha:
uri: {alpha}
beta:
uri: {beta}
"#,
alpha = alpha_dir.display(),
beta = beta_dir.display(),
),
)
.unwrap();
let settings: ServerConfig =
load_server_settings(Some(&config_path), None, None, None, None, true).await.unwrap();
assert!(matches!(settings.mode, ServerConfigMode::Multi { .. }));
match settings.mode {
ServerConfigMode::Multi { graphs, .. } => {
assert_eq!(graphs.len(), 2);
let ids: Vec<&str> = graphs.iter().map(|g| g.graph_id.as_str()).collect();
assert_eq!(ids, vec!["alpha", "beta"]);
}
_ => unreachable!(),
}
}
}

84
crates/omnigraph-server/tests/data_routes.rs
View file

@ -63,7 +63,7 @@ async fn export_route_returns_jsonl_for_branch_snapshot() {
.clone()
.oneshot(
Request::builder()
.uri("/export")
.uri(g("/export"))
.method(Method::POST)
.header("content-type", "application/json")
.header("authorization", format!("Bearer {}", token))
@ -99,7 +99,7 @@ async fn snapshot_route_returns_manifest_dataset_version() {
let (snapshot_status, snapshot_body) = json_response(
&app,
Request::builder()
.uri("/snapshot?branch=main")
.uri(g("/snapshot?branch=main"))
.method(Method::GET)
.body(Body::empty())
.unwrap(),
@ -131,7 +131,7 @@ async fn ingest_creates_branch_returns_metadata_and_stamps_actor() {
let (status, body) = json_response(
&app,
Request::builder()
.uri("/ingest")
.uri(g("/ingest"))
.method(Method::POST)
.header("authorization", "Bearer token-one")
.header("content-type", "application/json")
@ -195,7 +195,7 @@ async fn ingest_existing_branch_skips_branch_create_policy_check() {
let (status, body) = json_response(
&app,
Request::builder()
.uri("/ingest")
.uri(g("/ingest"))
.method(Method::POST)
.header("authorization", "Bearer team-token")
.header("content-type", "application/json")
@ -223,7 +223,7 @@ async fn ingest_without_from_returns_404_for_missing_branch_and_creates_nothing(
let (status, body) = json_response(
&app,
Request::builder()
.uri("/ingest")
.uri(g("/ingest"))
.method(Method::POST)
.header("content-type", "application/json")
.body(Body::from(serde_json::to_vec(&ingest).unwrap()))
@ -264,7 +264,7 @@ async fn ingest_without_from_loads_into_existing_branch() {
let (status, body) = json_response(
&app,
Request::builder()
.uri("/ingest")
.uri(g("/ingest"))
.method(Method::POST)
.header("content-type", "application/json")
.body(Body::from(serde_json::to_vec(&ingest).unwrap()))
@ -294,7 +294,7 @@ async fn ingest_denies_missing_branch_without_branch_create_permission() {
let (status, body) = json_response(
&app,
Request::builder()
.uri("/ingest")
.uri(g("/ingest"))
.method(Method::POST)
.header("authorization", "Bearer team-token")
.header("content-type", "application/json")
@ -327,7 +327,7 @@ async fn ingest_denies_when_actor_lacks_change_permission() {
let (status, body) = json_response(
&app,
Request::builder()
.uri("/ingest")
.uri(g("/ingest"))
.method(Method::POST)
.header("authorization", "Bearer team-token")
.header("content-type", "application/json")
@ -357,7 +357,7 @@ async fn ingest_rejects_payloads_over_32_mib() {
.clone()
.oneshot(
Request::builder()
.uri("/ingest")
.uri(g("/ingest"))
.method(Method::POST)
.header("content-type", "application/json")
.body(Body::from(serde_json::to_vec(&oversize).unwrap()))
@ -419,7 +419,7 @@ async fn branch_merge_conflict_response_includes_structured_conflicts() {
let (status, body) = json_response(
&app,
Request::builder()
.uri("/branches/merge")
.uri(g("/branches/merge"))
.method(Method::POST)
.header("content-type", "application/json")
.body(Body::from(serde_json::to_vec(&merge).unwrap()))
@ -451,7 +451,7 @@ async fn repeated_read_after_change_sees_updated_state_from_same_app() {
let (change_status, change_body) = json_response(
&app,
Request::builder()
.uri("/change")
.uri(g("/change"))
.method(Method::POST)
.header("content-type", "application/json")
.body(Body::from(serde_json::to_vec(&change).unwrap()))
@ -471,7 +471,7 @@ async fn repeated_read_after_change_sees_updated_state_from_same_app() {
let (read_status, read_body) = json_response(
&app,
Request::builder()
.uri("/read")
.uri(g("/read"))
.method(Method::POST)
.header("content-type", "application/json")
.body(Body::from(serde_json::to_vec(&read).unwrap()))
@ -497,7 +497,7 @@ async fn query_endpoint_runs_inline_read() {
let (status, body) = json_response(
&app,
Request::builder()
.uri("/query")
.uri(g("/query"))
.method(Method::POST)
.header("content-type", "application/json")
.body(Body::from(serde_json::to_vec(&query).unwrap()))
@ -524,7 +524,7 @@ async fn query_endpoint_rejects_mutation_with_400() {
let (status, body) = json_response(
&app,
Request::builder()
.uri("/query")
.uri(g("/query"))
.method(Method::POST)
.header("content-type", "application/json")
.body(Body::from(serde_json::to_vec(&query).unwrap()))
@ -555,7 +555,7 @@ async fn mutate_endpoint_runs_inline_mutation() {
.clone()
.oneshot(
Request::builder()
.uri("/mutate")
.uri(g("/mutate"))
.method(Method::POST)
.header("content-type", "application/json")
.body(Body::from(serde_json::to_vec(&request).unwrap()))
@ -595,7 +595,7 @@ async fn change_endpoint_emits_deprecation_headers() {
.clone()
.oneshot(
Request::builder()
.uri("/change")
.uri(g("/change"))
.method(Method::POST)
.header("content-type", "application/json")
.body(Body::from(serde_json::to_vec(&request).unwrap()))
@ -635,7 +635,7 @@ async fn load_endpoint_loads_into_existing_branch() {
.clone()
.oneshot(
Request::builder()
.uri("/load")
.uri(g("/load"))
.method(Method::POST)
.header("content-type", "application/json")
.body(Body::from(serde_json::to_vec(&request).unwrap()))
@ -672,7 +672,7 @@ async fn ingest_endpoint_emits_deprecation_headers() {
.clone()
.oneshot(
Request::builder()
.uri("/ingest")
.uri(g("/ingest"))
.method(Method::POST)
.header("content-type", "application/json")
.body(Body::from(serde_json::to_vec(&request).unwrap()))
@ -714,7 +714,7 @@ async fn read_endpoint_emits_deprecation_headers() {
.clone()
.oneshot(
Request::builder()
.uri("/read")
.uri(g("/read"))
.method(Method::POST)
.header("content-type", "application/json")
.body(Body::from(serde_json::to_vec(&request).unwrap()))
@ -757,7 +757,7 @@ async fn query_endpoint_does_not_emit_deprecation_headers() {
.clone()
.oneshot(
Request::builder()
.uri("/query")
.uri(g("/query"))
.method(Method::POST)
.header("content-type", "application/json")
.body(Body::from(serde_json::to_vec(&request).unwrap()))
@ -789,7 +789,7 @@ async fn change_endpoint_accepts_legacy_field_names() {
let (status, body) = json_response(
&app,
Request::builder()
.uri("/change")
.uri(g("/change"))
.method(Method::POST)
.header("content-type", "application/json")
.body(Body::from(serde_json::to_vec(&legacy_body).unwrap()))
@ -808,7 +808,7 @@ async fn change_endpoint_accepts_legacy_field_names() {
let (status, body) = json_response(
&app,
Request::builder()
.uri("/change")
.uri(g("/change"))
.method(Method::POST)
.header("content-type", "application/json")
.body(Body::from(serde_json::to_vec(&canonical_body).unwrap()))
@ -826,7 +826,7 @@ async fn remote_branch_list_create_merge_flow_works() {
let (list_status, list_body) = json_response(
&app,
Request::builder()
.uri("/branches")
.uri(g("/branches"))
.method(Method::GET)
.body(Body::empty())
.unwrap(),
@ -842,7 +842,7 @@ async fn remote_branch_list_create_merge_flow_works() {
let (create_status, create_body) = json_response(
&app,
Request::builder()
.uri("/branches")
.uri(g("/branches"))
.method(Method::POST)
.header("content-type", "application/json")
.body(Body::from(serde_json::to_vec(&create).unwrap()))
@ -856,7 +856,7 @@ async fn remote_branch_list_create_merge_flow_works() {
let (list_status, list_body) = json_response(
&app,
Request::builder()
.uri("/branches")
.uri(g("/branches"))
.method(Method::GET)
.body(Body::empty())
.unwrap(),
@ -874,7 +874,7 @@ async fn remote_branch_list_create_merge_flow_works() {
let (change_status, change_body) = json_response(
&app,
Request::builder()
.uri("/change")
.uri(g("/change"))
.method(Method::POST)
.header("content-type", "application/json")
.body(Body::from(serde_json::to_vec(&change).unwrap()))
@ -895,7 +895,7 @@ async fn remote_branch_list_create_merge_flow_works() {
let (read_status, read_body) = json_response(
&app,
Request::builder()
.uri("/read")
.uri(g("/read"))
.method(Method::POST)
.header("content-type", "application/json")
.body(Body::from(serde_json::to_vec(&read_main_before).unwrap()))
@ -912,7 +912,7 @@ async fn remote_branch_list_create_merge_flow_works() {
let (merge_status, merge_body) = json_response(
&app,
Request::builder()
.uri("/branches/merge")
.uri(g("/branches/merge"))
.method(Method::POST)
.header("content-type", "application/json")
.body(Body::from(serde_json::to_vec(&merge).unwrap()))
@ -934,7 +934,7 @@ async fn remote_branch_list_create_merge_flow_works() {
let (read_status, read_body) = json_response(
&app,
Request::builder()
.uri("/read")
.uri(g("/read"))
.method(Method::POST)
.header("content-type", "application/json")
.body(Body::from(serde_json::to_vec(&read_main_after).unwrap()))
@ -957,7 +957,7 @@ async fn remote_branch_delete_flow_works() {
let (create_status, _) = json_response(
&app,
Request::builder()
.uri("/branches")
.uri(g("/branches"))
.method(Method::POST)
.header("content-type", "application/json")
.body(Body::from(serde_json::to_vec(&create).unwrap()))
@ -969,7 +969,7 @@ async fn remote_branch_delete_flow_works() {
let (delete_status, delete_body) = json_response(
&app,
Request::builder()
.uri("/branches/feature")
.uri(g("/branches/feature"))
.method(Method::DELETE)
.body(Body::empty())
.unwrap(),
@ -981,7 +981,7 @@ async fn remote_branch_delete_flow_works() {
let (list_status, list_body) = json_response(
&app,
Request::builder()
.uri("/branches")
.uri(g("/branches"))
.method(Method::GET)
.body(Body::empty())
.unwrap(),
@ -1009,7 +1009,7 @@ async fn branch_delete_denies_without_policy_permission() {
let (status, body) = json_response(
&app,
Request::builder()
.uri("/branches/feature")
.uri(g("/branches/feature"))
.method(Method::DELETE)
.header("authorization", "Bearer token-team")
.body(Body::empty())
@ -1081,7 +1081,7 @@ query vector_search_string($q: String) {
let (status, body) = json_response(
&app,
Request::builder()
.uri("/read")
.uri(g("/read"))
.method(Method::POST)
.header("content-type", "application/json")
.body(Body::from(serde_json::to_vec(&read).unwrap()))
@ -1134,7 +1134,7 @@ async fn change_conflict_returns_manifest_conflict_409() {
let (status, body) = json_response(
&app,
Request::builder()
.uri("/change")
.uri(g("/change"))
.method(Method::POST)
.header("content-type", "application/json")
.body(Body::from(
@ -1206,7 +1206,7 @@ async fn change_concurrent_inserts_same_key_serialize_without_409() {
})
.unwrap();
let req = Request::builder()
.uri("/change")
.uri(g("/change"))
.method(Method::POST)
.header("content-type", "application/json")
.body(Body::from(body))
@ -1238,7 +1238,7 @@ async fn change_concurrent_inserts_same_key_serialize_without_409() {
let (snapshot_status, snapshot_body) = json_response(
&app,
Request::builder()
.uri("/snapshot?branch=main")
.uri(g("/snapshot?branch=main"))
.method(Method::GET)
.body(Body::empty())
.unwrap(),
@ -1319,7 +1319,7 @@ async fn change_concurrent_updates_same_key_serialize_via_publisher_cas() {
})
.unwrap();
let req = Request::builder()
.uri("/change")
.uri(g("/change"))
.method(Method::POST)
.header("content-type", "application/json")
.body(Body::from(body))
@ -1428,7 +1428,7 @@ query insert_c($name: String) {
})
.unwrap();
let req = Request::builder()
.uri("/change")
.uri(g("/change"))
.method(Method::POST)
.header("content-type", "application/json")
.body(Body::from(body))
@ -1445,7 +1445,7 @@ query insert_c($name: String) {
})
.unwrap();
let req = Request::builder()
.uri("/change")
.uri(g("/change"))
.method(Method::POST)
.header("content-type", "application/json")
.body(Body::from(body))
@ -1474,7 +1474,7 @@ query insert_c($name: String) {
let (status, body) = json_response(
&app,
Request::builder()
.uri("/snapshot?branch=main")
.uri(g("/snapshot?branch=main"))
.method(Method::GET)
.body(Body::empty())
.unwrap(),
@ -1582,7 +1582,7 @@ async fn ingest_per_actor_admission_cap_returns_429() {
})
.unwrap();
let req = Request::builder()
.uri("/ingest")
.uri(g("/ingest"))
.method(Method::POST)
.header("authorization", "Bearer flooder-token")
.header("content-type", "application/json")

34
crates/omnigraph-server/tests/multi_graph.rs
View file

@ -245,7 +245,7 @@ async fn concurrent_branch_ops_morphological_matrix() {
.clone()
.oneshot(
Request::builder()
.uri("/branches")
.uri(g("/branches"))
.method(Method::GET)
.body(Body::empty())
.unwrap(),
@ -366,7 +366,7 @@ async fn concurrent_branch_ops_morphological_matrix() {
.clone()
.oneshot(
Request::builder()
.uri("/snapshot?branch=main")
.uri(g("/snapshot?branch=main"))
.method(Method::GET)
.body(Body::empty())
.unwrap(),
@ -540,31 +540,15 @@ graphs:
#[tokio::test]
async fn cluster_boot_refusals() {
// Mutual exclusion with --config / URI.
// RFC-011 cluster-only: with no --cluster, boot refuses with the
// cluster-required remedy.
let err = omnigraph_server::load_server_settings(None, None, true)
.await
.unwrap_err();
assert!(err.to_string().contains("boots from a cluster"), "{err}");
let temp = converged_cluster_dir("").await;
let dir = temp.path().to_path_buf();
let err = omnigraph_server::load_server_settings(
Some(&dir.join("omnigraph.yaml")),
Some(&dir),
None,
None,
None,
true,
)
.await
.unwrap_err();
assert!(err.to_string().contains("exclusive boot source"), "{err}");
let err = omnigraph_server::load_server_settings(
None,
Some(&dir),
Some("file:///tmp/x.omni".to_string()),
None,
None,
true,
)
.await
.unwrap_err();
assert!(err.to_string().contains("exclusive boot source"), "{err}");
// Tampered catalog blob refuses boot with the remedy.
let blob_dir = dir.join("__cluster/resources/query/knowledge/find_person");

222
crates/omnigraph-server/tests/openapi.rs
View file

@ -8,10 +8,9 @@ use axum::body::{Body, to_bytes};
use axum::http::{Method, Request, StatusCode};
use omnigraph::db::Omnigraph;
use omnigraph::loader::{LoadMode, load_jsonl};
use omnigraph_server::{ApiDoc, AppState, build_app};
use omnigraph_server::{AppState, build_app, served_openapi};
use serde_json::Value;
use tower::ServiceExt;
use utoipa::OpenApi;
fn fixture(name: &str) -> PathBuf {
PathBuf::from(env!("CARGO_MANIFEST_DIR"))
@ -71,7 +70,10 @@ async fn json_response(app: &Router, request: Request<Body>) -> (StatusCode, Val
}
fn openapi_doc() -> utoipa::openapi::OpenApi {
ApiDoc::openapi()
// RFC-011 cluster-only: the canonical committed spec is the SERVED
// shape — protected routes nested under `/graphs/{graph_id}/…`,
// `/healthz` and `/graphs` flat. This matches what the server serves.
served_openapi()
}
fn openapi_json() -> Value {
@ -159,26 +161,28 @@ fn openapi_info_contains_version() {
// Path coverage tests
// ---------------------------------------------------------------------------
// The canonical served spec keeps `/healthz` and `/graphs` flat; every
// protected route nests under `/graphs/{graph_id}/…`.
const EXPECTED_PATHS: &[&str] = &[
"/healthz",
"/graphs",
"/snapshot",
"/read",
"/query",
"/export",
"/change",
"/mutate",
"/queries",
"/queries/{name}",
"/schema",
"/schema/apply",
"/load",
"/ingest",
"/branches",
"/branches/{branch}",
"/branches/merge",
"/commits",
"/commits/{commit_id}",
"/graphs/{graph_id}/snapshot",
"/graphs/{graph_id}/read",
"/graphs/{graph_id}/query",
"/graphs/{graph_id}/export",
"/graphs/{graph_id}/change",
"/graphs/{graph_id}/mutate",
"/graphs/{graph_id}/queries",
"/graphs/{graph_id}/queries/{name}",
"/graphs/{graph_id}/schema",
"/graphs/{graph_id}/schema/apply",
"/graphs/{graph_id}/load",
"/graphs/{graph_id}/ingest",
"/graphs/{graph_id}/branches",
"/graphs/{graph_id}/branches/{branch}",
"/graphs/{graph_id}/branches/merge",
"/graphs/{graph_id}/commits",
"/graphs/{graph_id}/commits/{commit_id}",
];
#[test]
@ -222,25 +226,25 @@ fn openapi_healthz_is_get() {
#[test]
fn openapi_read_is_post() {
let doc = openapi_json();
assert!(doc["paths"]["/read"]["post"].is_object());
assert!(doc["paths"]["/graphs/{graph_id}/read"]["post"].is_object());
}
#[test]
fn openapi_export_is_post() {
let doc = openapi_json();
assert!(doc["paths"]["/export"]["post"].is_object());
assert!(doc["paths"]["/graphs/{graph_id}/export"]["post"].is_object());
}
#[test]
fn openapi_change_is_post() {
let doc = openapi_json();
assert!(doc["paths"]["/change"]["post"].is_object());
assert!(doc["paths"]["/graphs/{graph_id}/change"]["post"].is_object());
}
#[test]
fn openapi_mutate_is_post() {
let doc = openapi_json();
assert!(doc["paths"]["/mutate"]["post"].is_object());
assert!(doc["paths"]["/graphs/{graph_id}/mutate"]["post"].is_object());
}
// Deprecation flagging — `/read` and `/change` are kept indefinitely for
@ -253,7 +257,7 @@ fn openapi_mutate_is_post() {
fn openapi_read_is_deprecated() {
let doc = openapi_json();
assert_eq!(
doc["paths"]["/read"]["post"]["deprecated"],
doc["paths"]["/graphs/{graph_id}/read"]["post"]["deprecated"],
serde_json::Value::Bool(true),
"/read must be flagged deprecated in OpenAPI; use /query instead"
);
@ -263,7 +267,7 @@ fn openapi_read_is_deprecated() {
fn openapi_change_is_deprecated() {
let doc = openapi_json();
assert_eq!(
doc["paths"]["/change"]["post"]["deprecated"],
doc["paths"]["/graphs/{graph_id}/change"]["post"]["deprecated"],
serde_json::Value::Bool(true),
"/change must be flagged deprecated in OpenAPI; use /mutate instead"
);
@ -272,7 +276,7 @@ fn openapi_change_is_deprecated() {
#[test]
fn openapi_query_is_not_deprecated() {
let doc = openapi_json();
let deprecated = doc["paths"]["/query"]["post"]
let deprecated = doc["paths"]["/graphs/{graph_id}/query"]["post"]
.get("deprecated")
.and_then(serde_json::Value::as_bool)
.unwrap_or(false);
@ -285,7 +289,7 @@ fn openapi_query_is_not_deprecated() {
#[test]
fn openapi_mutate_is_not_deprecated() {
let doc = openapi_json();
let deprecated = doc["paths"]["/mutate"]["post"]
let deprecated = doc["paths"]["/graphs/{graph_id}/mutate"]["post"]
.get("deprecated")
.and_then(serde_json::Value::as_bool)
.unwrap_or(false);
@ -298,15 +302,15 @@ fn openapi_mutate_is_not_deprecated() {
#[test]
fn openapi_ingest_is_post() {
let doc = openapi_json();
assert!(doc["paths"]["/ingest"]["post"].is_object());
assert!(doc["paths"]["/graphs/{graph_id}/ingest"]["post"].is_object());
}
#[test]
fn openapi_load_is_not_deprecated() {
// RFC-009 Phase 5: /load is the canonical bulk-load endpoint.
let doc = openapi_json();
assert!(doc["paths"]["/load"]["post"].is_object());
let deprecated = doc["paths"]["/load"]["post"]
assert!(doc["paths"]["/graphs/{graph_id}/load"]["post"].is_object());
let deprecated = doc["paths"]["/graphs/{graph_id}/load"]["post"]
.get("deprecated")
.and_then(serde_json::Value::as_bool)
.unwrap_or(false);
@ -321,7 +325,7 @@ fn openapi_ingest_is_deprecated() {
// RFC-009 Phase 5: /ingest is now the deprecated alias of /load.
let doc = openapi_json();
assert_eq!(
doc["paths"]["/ingest"]["post"]["deprecated"],
doc["paths"]["/graphs/{graph_id}/ingest"]["post"]["deprecated"],
serde_json::Value::Bool(true),
"/ingest must be flagged deprecated now that /load is canonical"
);
@ -330,32 +334,32 @@ fn openapi_ingest_is_deprecated() {
#[test]
fn openapi_branches_supports_get_and_post() {
let doc = openapi_json();
assert!(doc["paths"]["/branches"]["get"].is_object());
assert!(doc["paths"]["/branches"]["post"].is_object());
assert!(doc["paths"]["/graphs/{graph_id}/branches"]["get"].is_object());
assert!(doc["paths"]["/graphs/{graph_id}/branches"]["post"].is_object());
}
#[test]
fn openapi_branch_delete_is_delete() {
let doc = openapi_json();
assert!(doc["paths"]["/branches/{branch}"]["delete"].is_object());
assert!(doc["paths"]["/graphs/{graph_id}/branches/{branch}"]["delete"].is_object());
}
#[test]
fn openapi_branch_merge_is_post() {
let doc = openapi_json();
assert!(doc["paths"]["/branches/merge"]["post"].is_object());
assert!(doc["paths"]["/graphs/{graph_id}/branches/merge"]["post"].is_object());
}
#[test]
fn openapi_commits_is_get() {
let doc = openapi_json();
assert!(doc["paths"]["/commits"]["get"].is_object());
assert!(doc["paths"]["/graphs/{graph_id}/commits"]["get"].is_object());
}
#[test]
fn openapi_commit_show_is_get() {
let doc = openapi_json();
assert!(doc["paths"]["/commits/{commit_id}"]["get"].is_object());
assert!(doc["paths"]["/graphs/{graph_id}/commits/{commit_id}"]["get"].is_object());
}
// ---------------------------------------------------------------------------
@ -510,13 +514,13 @@ fn query_request_query_is_required() {
#[test]
fn openapi_query_is_post() {
let doc = openapi_json();
assert!(doc["paths"]["/query"]["post"].is_object());
assert!(doc["paths"]["/graphs/{graph_id}/query"]["post"].is_object());
}
#[test]
fn query_endpoint_documents_mutation_400() {
let doc = openapi_json();
let four_hundred = &doc["paths"]["/query"]["post"]["responses"]["400"];
let four_hundred = &doc["paths"]["/graphs/{graph_id}/query"]["post"]["responses"]["400"];
let description = four_hundred["description"].as_str().unwrap_or_default();
assert!(
description.contains("mutations") || description.contains("POST /mutate"),
@ -727,21 +731,21 @@ fn openapi_defines_bearer_token_security_scheme() {
fn protected_endpoints_reference_bearer_token_security() {
let doc = openapi_json();
let protected_paths = [
("/read", "post"),
("/change", "post"),
("/schema/apply", "post"),
("/queries", "get"),
("/queries/{name}", "post"),
("/load", "post"),
("/ingest", "post"),
("/export", "post"),
("/snapshot", "get"),
("/branches", "get"),
("/branches", "post"),
("/branches/{branch}", "delete"),
("/branches/merge", "post"),
("/commits", "get"),
("/commits/{commit_id}", "get"),
("/graphs/{graph_id}/read", "post"),
("/graphs/{graph_id}/change", "post"),
("/graphs/{graph_id}/schema/apply", "post"),
("/graphs/{graph_id}/queries", "get"),
("/graphs/{graph_id}/queries/{name}", "post"),
("/graphs/{graph_id}/load", "post"),
("/graphs/{graph_id}/ingest", "post"),
("/graphs/{graph_id}/export", "post"),
("/graphs/{graph_id}/snapshot", "get"),
("/graphs/{graph_id}/branches", "get"),
("/graphs/{graph_id}/branches", "post"),
("/graphs/{graph_id}/branches/{branch}", "delete"),
("/graphs/{graph_id}/branches/merge", "post"),
("/graphs/{graph_id}/commits", "get"),
("/graphs/{graph_id}/commits/{commit_id}", "get"),
];
for (path, method) in protected_paths {
@ -773,7 +777,7 @@ fn healthz_does_not_require_security() {
#[test]
fn branch_delete_has_branch_path_parameter() {
let doc = openapi_json();
let params = doc["paths"]["/branches/{branch}"]["delete"]["parameters"]
let params = doc["paths"]["/graphs/{graph_id}/branches/{branch}"]["delete"]["parameters"]
.as_array()
.unwrap();
let has_branch = params
@ -788,7 +792,7 @@ fn branch_delete_has_branch_path_parameter() {
#[test]
fn commit_show_has_commit_id_path_parameter() {
let doc = openapi_json();
let params = doc["paths"]["/commits/{commit_id}"]["get"]["parameters"]
let params = doc["paths"]["/graphs/{graph_id}/commits/{commit_id}"]["get"]["parameters"]
.as_array()
.unwrap();
let has_commit_id = params
@ -803,7 +807,7 @@ fn commit_show_has_commit_id_path_parameter() {
#[test]
fn snapshot_has_branch_query_parameter() {
let doc = openapi_json();
let params = doc["paths"]["/snapshot"]["get"]["parameters"]
let params = doc["paths"]["/graphs/{graph_id}/snapshot"]["get"]["parameters"]
.as_array()
.unwrap();
let has_branch = params
@ -818,7 +822,7 @@ fn snapshot_has_branch_query_parameter() {
#[test]
fn commits_has_branch_query_parameter() {
let doc = openapi_json();
let params = doc["paths"]["/commits"]["get"]["parameters"]
let params = doc["paths"]["/graphs/{graph_id}/commits"]["get"]["parameters"]
.as_array()
.unwrap();
let has_branch = params
@ -858,7 +862,7 @@ fn openapi_operations_have_tags() {
#[test]
fn read_endpoint_200_references_read_output_schema() {
let doc = openapi_json();
let content = &doc["paths"]["/read"]["post"]["responses"]["200"]["content"];
let content = &doc["paths"]["/graphs/{graph_id}/read"]["post"]["responses"]["200"]["content"];
let schema = &content["application/json"]["schema"];
let ref_path = schema["$ref"].as_str().unwrap();
assert!(
@ -870,7 +874,7 @@ fn read_endpoint_200_references_read_output_schema() {
#[test]
fn change_endpoint_200_references_change_output_schema() {
let doc = openapi_json();
let content = &doc["paths"]["/change"]["post"]["responses"]["200"]["content"];
let content = &doc["paths"]["/graphs/{graph_id}/change"]["post"]["responses"]["200"]["content"];
let schema = &content["application/json"]["schema"];
let ref_path = schema["$ref"].as_str().unwrap();
assert!(
@ -895,11 +899,11 @@ fn healthz_200_references_health_output_schema() {
fn error_responses_reference_error_output_schema() {
let doc = openapi_json();
let paths_with_errors = [
("/read", "post", "400"),
("/read", "post", "401"),
("/change", "post", "400"),
("/change", "post", "409"),
("/branches", "post", "409"),
("/graphs/{graph_id}/read", "post", "400"),
("/graphs/{graph_id}/read", "post", "401"),
("/graphs/{graph_id}/change", "post", "400"),
("/graphs/{graph_id}/change", "post", "409"),
("/graphs/{graph_id}/branches", "post", "409"),
];
for (path, method, status) in paths_with_errors {
@ -921,13 +925,13 @@ fn error_responses_reference_error_output_schema() {
fn post_endpoints_have_request_body() {
let doc = openapi_json();
let post_paths = [
("/read", "ReadRequest"),
("/change", "ChangeRequest"),
("/schema/apply", "SchemaApplyRequest"),
("/ingest", "IngestRequest"),
("/export", "ExportRequest"),
("/branches", "BranchCreateRequest"),
("/branches/merge", "BranchMergeRequest"),
("/graphs/{graph_id}/read", "ReadRequest"),
("/graphs/{graph_id}/change", "ChangeRequest"),
("/graphs/{graph_id}/schema/apply", "SchemaApplyRequest"),
("/graphs/{graph_id}/ingest", "IngestRequest"),
("/graphs/{graph_id}/export", "ExportRequest"),
("/graphs/{graph_id}/branches", "BranchCreateRequest"),
("/graphs/{graph_id}/branches/merge", "BranchMergeRequest"),
];
for (path, expected_schema) in post_paths {
@ -948,7 +952,7 @@ fn post_endpoints_have_request_body() {
#[test]
fn invoke_stored_query_request_body_is_optional() {
let doc = openapi_json();
let request_body = &doc["paths"]["/queries/{name}"]["post"]["requestBody"];
let request_body = &doc["paths"]["/graphs/{graph_id}/queries/{name}"]["post"]["requestBody"];
assert!(
request_body.is_object(),
"POST /queries/{{name}} should document its optional request body"
@ -1051,12 +1055,14 @@ async fn auth_mode_spec_has_security_on_protected_operations() {
.body(Body::empty())
.unwrap();
let (_, json) = json_response(&app, request).await;
// RFC-011 cluster-only: the served spec always nests protected
// routes under `/graphs/{graph_id}/...`.
let protected_paths = [
("/read", "post"),
("/change", "post"),
("/snapshot", "get"),
("/branches", "get"),
("/commits", "get"),
("/graphs/{graph_id}/read", "post"),
("/graphs/{graph_id}/change", "post"),
("/graphs/{graph_id}/snapshot", "get"),
("/graphs/{graph_id}/branches", "get"),
("/graphs/{graph_id}/commits", "get"),
];
for (path, method) in protected_paths {
let security = &json["paths"][path][method]["security"];
@ -1073,22 +1079,6 @@ async fn auth_mode_spec_has_security_on_protected_operations() {
}
}
#[tokio::test]
async fn auth_mode_spec_matches_static_generation() {
let (_temp, app) = app_for_loaded_graph_with_auth("secret").await;
let request = Request::builder()
.method(Method::GET)
.uri("/openapi.json")
.body(Body::empty())
.unwrap();
let (_, served) = json_response(&app, request).await;
let static_doc = openapi_json();
assert_eq!(
served, static_doc,
"auth-mode served spec must match static generation"
);
}
#[tokio::test]
async fn auth_mode_healthz_still_has_no_security() {
let (_temp, app) = app_for_loaded_graph_with_auth("secret").await;
@ -1394,8 +1384,9 @@ async fn multi_mode_operation_ids_are_unique() {
}
#[tokio::test]
async fn single_mode_openapi_unchanged_by_cluster_filter() {
// Regression: single mode still emits the legacy flat surface.
async fn served_spec_always_nests_under_cluster_prefix() {
// RFC-011 cluster-only: even a one-graph convenience app serves the
// nested cluster surface and never the flat protected routes.
let (_temp, app) = app_for_loaded_graph().await;
let request = Request::builder()
.method(Method::GET)
@ -1405,16 +1396,37 @@ async fn single_mode_openapi_unchanged_by_cluster_filter() {
let (_, json) = json_response(&app, request).await;
let paths = json["paths"].as_object().unwrap();
let path_keys: HashSet<&str> = paths.keys().map(|k| k.as_str()).collect();
for expected in EXPECTED_PATHS {
assert!(
path_keys.contains(expected),
"single mode must still emit flat path: {expected}"
);
}
for cluster in EXPECTED_CLUSTER_PATHS {
assert!(
!path_keys.contains(cluster),
"single mode must NOT emit cluster path: {cluster}"
path_keys.contains(cluster),
"served spec must emit cluster path: {cluster}. Found: {path_keys:?}"
);
}
// The flat protected routes must NOT appear — only the nested
// cluster surface plus the always-flat `/healthz` and `/graphs`.
let flat_protected = [
"/snapshot",
"/read",
"/query",
"/export",
"/change",
"/mutate",
"/queries",
"/queries/{name}",
"/schema",
"/schema/apply",
"/load",
"/ingest",
"/branches",
"/branches/{branch}",
"/branches/merge",
"/commits",
"/commits/{commit_id}",
];
for flat in flat_protected {
assert!(
!path_keys.contains(flat),
"served spec must NOT emit flat protected path: {flat}"
);
}
}

7
crates/omnigraph-server/tests/s3.rs
View file

@ -43,7 +43,7 @@ async fn server_opens_s3_graph_directly_and_serves_snapshot_and_read() {
let (snapshot_status, snapshot_body) = json_response(
&app,
Request::builder()
.uri("/snapshot")
.uri(g("/snapshot"))
.method(Method::GET)
.header("authorization", "Bearer s3-token")
.body(Body::empty())
@ -63,7 +63,7 @@ async fn server_opens_s3_graph_directly_and_serves_snapshot_and_read() {
let (read_status, read_body) = json_response(
&app,
Request::builder()
.uri("/read")
.uri(g("/read"))
.method(Method::POST)
.header("authorization", "Bearer s3-token")
.header("content-type", "application/json")
@ -134,11 +134,8 @@ async fn server_boots_cluster_from_bare_storage_uri_and_serves_query() {
}
let settings = omnigraph_server::load_server_settings(
None,
Some(&std::path::PathBuf::from(&root)),
None,
None,
None,
true,
)
.await

44
crates/omnigraph-server/tests/schema_routes.rs
View file

@ -30,7 +30,7 @@ async fn schema_apply_route_updates_graph_for_authorized_admin() {
let request = Request::builder()
.method(Method::POST)
.uri("/schema/apply")
.uri(g("/schema/apply"))
.header("content-type", "application/json")
.header("authorization", "Bearer admin-token")
.body(Body::from(
@ -65,7 +65,7 @@ async fn schema_apply_route_rejects_stored_query_breakage_before_publish() {
let request = Request::builder()
.method(Method::POST)
.uri("/schema/apply")
.uri(g("/schema/apply"))
.header("content-type", "application/json")
.header("authorization", "Bearer admin-token")
.body(Body::from(
@ -115,7 +115,7 @@ async fn schema_apply_route_noop_keeps_valid_stored_query_registry() {
let request = Request::builder()
.method(Method::POST)
.uri("/schema/apply")
.uri(g("/schema/apply"))
.header("content-type", "application/json")
.header("authorization", "Bearer admin-token")
.body(Body::from(
@ -142,7 +142,7 @@ async fn schema_apply_route_requires_schema_apply_policy_permission() {
let request = Request::builder()
.method(Method::POST)
.uri("/schema/apply")
.uri(g("/schema/apply"))
.header("content-type", "application/json")
.header("authorization", "Bearer admin-token")
.body(Body::from(
@ -173,7 +173,7 @@ async fn schema_apply_route_requires_bearer_token_when_policy_enabled() {
let request = Request::builder()
.method(Method::POST)
.uri("/schema/apply")
.uri(g("/schema/apply"))
.header("content-type", "application/json")
.body(Body::from(
serde_json::to_vec(&SchemaApplyRequest {
@ -203,7 +203,7 @@ async fn schema_apply_route_can_rename_type() {
let request = Request::builder()
.method(Method::POST)
.uri("/schema/apply")
.uri(g("/schema/apply"))
.header("content-type", "application/json")
.header("authorization", "Bearer admin-token")
.body(Body::from(
@ -239,7 +239,7 @@ async fn schema_apply_route_can_rename_property() {
let request = Request::builder()
.method(Method::POST)
.uri("/schema/apply")
.uri(g("/schema/apply"))
.header("content-type", "application/json")
.header("authorization", "Bearer admin-token")
.body(Body::from(
@ -279,7 +279,7 @@ async fn schema_apply_route_can_add_index() {
let request = Request::builder()
.method(Method::POST)
.uri("/schema/apply")
.uri(g("/schema/apply"))
.header("content-type", "application/json")
.header("authorization", "Bearer admin-token")
.body(Body::from(
@ -323,7 +323,7 @@ async fn schema_apply_route_rejects_unsupported_plan() {
let request = Request::builder()
.method(Method::POST)
.uri("/schema/apply")
.uri(g("/schema/apply"))
.header("content-type", "application/json")
.header("authorization", "Bearer admin-token")
.body(Body::from(
@ -364,7 +364,7 @@ async fn schema_apply_route_rejects_when_non_main_branch_exists() {
let request = Request::builder()
.method(Method::POST)
.uri("/schema/apply")
.uri(g("/schema/apply"))
.header("content-type", "application/json")
.header("authorization", "Bearer admin-token")
.body(Body::from(
@ -393,7 +393,7 @@ async fn schema_drift_returns_conflict_for_snapshot_read_and_change() {
let (snapshot_status, snapshot_body) = json_response(
&app,
Request::builder()
.uri("/snapshot?branch=main")
.uri(g("/snapshot?branch=main"))
.method(Method::GET)
.body(Body::empty())
.unwrap(),
@ -421,7 +421,7 @@ async fn schema_drift_returns_conflict_for_snapshot_read_and_change() {
let (read_status, read_body) = json_response(
&app,
Request::builder()
.uri("/read")
.uri(g("/read"))
.method(Method::POST)
.header("content-type", "application/json")
.body(Body::from(serde_json::to_vec(&read).unwrap()))
@ -449,7 +449,7 @@ async fn schema_drift_returns_conflict_for_snapshot_read_and_change() {
let (change_status, change_body) = json_response(
&app,
Request::builder()
.uri("/change")
.uri(g("/change"))
.method(Method::POST)
.header("content-type", "application/json")
.body(Body::from(serde_json::to_vec(&change).unwrap()))
@ -475,7 +475,7 @@ async fn schema_route_returns_current_source() {
let (status, body) = json_response(
&app,
Request::builder()
.uri("/schema")
.uri(g("/schema"))
.method(Method::GET)
.body(Body::empty())
.unwrap(),
@ -494,7 +494,7 @@ async fn schema_route_requires_bearer_token_when_auth_configured() {
let (missing_status, missing_body) = json_response(
&app,
Request::builder()
.uri("/schema")
.uri(g("/schema"))
.method(Method::GET)
.body(Body::empty())
.unwrap(),
@ -510,7 +510,7 @@ async fn schema_route_requires_bearer_token_when_auth_configured() {
let (ok_status, ok_body) = json_response(
&app,
Request::builder()
.uri("/schema")
.uri(g("/schema"))
.method(Method::GET)
.header("authorization", "Bearer demo-token")
.body(Body::empty())
@ -541,7 +541,7 @@ async fn schema_route_denied_when_actor_lacks_read_permission() {
let (status, body) = json_response(
&app,
Request::builder()
.uri("/schema")
.uri(g("/schema"))
.method(Method::GET)
.header("authorization", "Bearer team-token")
.body(Body::empty())
@ -582,7 +582,7 @@ async fn schema_apply_route_soft_drops_property_via_http() {
&app,
Request::builder()
.method(Method::POST)
.uri("/schema/apply")
.uri(g("/schema/apply"))
.header("content-type", "application/json")
.header("authorization", "Bearer admin-token")
.body(Body::from(
@ -639,7 +639,7 @@ async fn schema_apply_route_soft_drops_node_type_via_http() {
&app,
Request::builder()
.method(Method::POST)
.uri("/schema/apply")
.uri(g("/schema/apply"))
.header("content-type", "application/json")
.header("authorization", "Bearer admin-token")
.body(Body::from(
@ -691,7 +691,7 @@ async fn schema_apply_route_hard_drops_property_with_allow_data_loss() {
&app,
Request::builder()
.method(Method::POST)
.uri("/schema/apply")
.uri(g("/schema/apply"))
.header("content-type", "application/json")
.header("authorization", "Bearer admin-token")
.body(Body::from(
@ -746,7 +746,7 @@ async fn schema_apply_route_keeps_drops_soft_without_flag() {
&app,
Request::builder()
.method(Method::POST)
.uri("/schema/apply")
.uri(g("/schema/apply"))
.header("content-type", "application/json")
.header("authorization", "Bearer admin-token")
.body(Body::from(
@ -808,7 +808,7 @@ async fn schema_apply_route_additive_property_preserves_existing_rows() {
&app,
Request::builder()
.method(Method::POST)
.uri("/schema/apply")
.uri(g("/schema/apply"))
.header("content-type", "application/json")
.header("authorization", "Bearer admin-token")
.body(Body::from(

6
crates/omnigraph-server/tests/stored_queries.rs
View file

@ -324,7 +324,7 @@ async fn list_queries_returns_only_exposed_with_typed_params() {
INVOKE_POLICY_YAML,
)
.await;
let (status, body) = json_response(&app, get_request("/queries", "t-invoke")).await;
let (status, body) = json_response(&app, get_request(&g("/queries"), "t-invoke")).await;
assert_eq!(status, StatusCode::OK, "body: {body}");
let entries = body["queries"].as_array().unwrap();
@ -355,7 +355,7 @@ async fn list_queries_is_read_gated_so_a_non_invoker_can_list() {
INVOKE_POLICY_YAML,
)
.await;
let (status, body) = json_response(&app, get_request("/queries", "t-noinvoke")).await;
let (status, body) = json_response(&app, get_request(&g("/queries"), "t-noinvoke")).await;
assert_eq!(status, StatusCode::OK, "read-gated catalog; body: {body}");
let names: Vec<&str> = body["queries"]
.as_array()
@ -372,7 +372,7 @@ async fn list_queries_is_read_gated_so_a_non_invoker_can_list() {
#[tokio::test(flavor = "multi_thread")]
async fn list_queries_is_empty_when_no_registry() {
let (_temp, app) = app_for_loaded_graph_with_auth("demo-token").await;
let (status, body) = json_response(&app, get_request("/queries", "demo-token")).await;
let (status, body) = json_response(&app, get_request(&g("/queries"), "demo-token")).await;
assert_eq!(status, StatusCode::OK, "body: {body}");
assert!(
body["queries"].as_array().unwrap().is_empty(),

36
crates/omnigraph-server/tests/support/mod.rs
View file

@ -248,9 +248,17 @@ rules:
pub const FIND_PERSON_GQ: &str =
"query find_person($name: String) { match { $p: Person { name: $name } } return { $p.age } }";
/// RFC-011 cluster-only: the single-graph convenience apps built by the
/// `app_for_loaded_graph*` helpers serve the graph under the reserved id
/// `default`. This prefixes a flat per-graph path (e.g. `/snapshot`) with
/// the cluster route prefix so tests address `/graphs/default/snapshot`.
pub fn g(path: &str) -> String {
format!("/graphs/default{path}")
}
pub fn invoke_request(name: &str, token: &str, body: Value) -> Request<Body> {
Request::builder()
.uri(format!("/queries/{name}"))
.uri(g(&format!("/queries/{name}")))
.method(Method::POST)
.header("content-type", "application/json")
.header("authorization", format!("Bearer {token}"))
@ -265,7 +273,7 @@ pub fn invoke_request_bytes(
content_type: Option<&str>,
) -> Request<Body> {
let mut builder = Request::builder()
.uri(format!("/queries/{name}"))
.uri(g(&format!("/queries/{name}")))
.method(Method::POST)
.header("authorization", format!("Bearer {token}"));
if let Some(content_type) = content_type {
@ -656,7 +664,7 @@ pub mod matrix {
.clone()
.oneshot(
Request::builder()
.uri("/branches")
.uri(g("/branches"))
.method(Method::POST)
.header("content-type", "application/json")
.body(Body::from(body))
@ -686,7 +694,7 @@ pub mod matrix {
.clone()
.oneshot(
Request::builder()
.uri("/change")
.uri(g("/change"))
.method(Method::POST)
.header("content-type", "application/json")
.body(Body::from(body))
@ -728,7 +736,7 @@ pub mod matrix {
.clone()
.oneshot(
Request::builder()
.uri(format!("/snapshot?branch={}", branch))
.uri(g(&format!("/snapshot?branch={}", branch)))
.method(Method::GET)
.body(Body::empty())
.unwrap(),
@ -766,7 +774,7 @@ pub mod matrix {
.clone()
.oneshot(
Request::builder()
.uri("/read")
.uri(g("/read"))
.method(Method::POST)
.header("content-type", "application/json")
.body(Body::from(body))
@ -833,7 +841,7 @@ pub mod matrix {
.clone()
.oneshot(
Request::builder()
.uri("/change")
.uri(g("/change"))
.method(Method::POST)
.header("content-type", "application/json")
.body(Body::from(body))
@ -874,7 +882,7 @@ pub mod matrix {
let response = app
.oneshot(
Request::builder()
.uri("/branches/merge")
.uri(g("/branches/merge"))
.method(Method::POST)
.header("content-type", "application/json")
.body(Body::from(body))
@ -910,7 +918,7 @@ pub mod matrix {
let response = app
.oneshot(
Request::builder()
.uri("/change")
.uri(g("/change"))
.method(Method::POST)
.header("content-type", "application/json")
.body(Body::from(body))
@ -943,7 +951,7 @@ pub mod matrix {
let response = app
.oneshot(
Request::builder()
.uri("/branches")
.uri(g("/branches"))
.method(Method::POST)
.header("content-type", "application/json")
.body(Body::from(body))
@ -970,7 +978,7 @@ pub mod matrix {
let response = app
.oneshot(
Request::builder()
.uri(format!("/branches/{}", name))
.uri(g(&format!("/branches/{}", name)))
.method(Method::DELETE)
.body(Body::empty())
.unwrap(),
@ -1091,7 +1099,7 @@ pub async fn http_change_decision(
let (status, _body) = json_response(
&app,
Request::builder()
.uri("/change")
.uri(g("/change"))
.method(Method::POST)
.header(AUTHORIZATION, format!("Bearer {token}"))
.header("content-type", "application/json")
@ -1141,7 +1149,7 @@ pub async fn http_merge_decision(
let (status, _body) = json_response(
&app,
Request::builder()
.uri("/branches/merge")
.uri(g("/branches/merge"))
.method(Method::POST)
.header(AUTHORIZATION, format!("Bearer {token}"))
.header("content-type", "application/json")
@ -1191,5 +1199,5 @@ graphs:
}
pub async fn cluster_settings(dir: &Path) -> color_eyre::eyre::Result<omnigraph_server::ServerConfig> {
omnigraph_server::load_server_settings(None, Some(&dir.to_path_buf()), None, None, None, true).await
omnigraph_server::load_server_settings(Some(&dir.to_path_buf()), None, true).await
}