diff --git a/src/gateway/mod.rs b/src/gateway/mod.rs index 59c4916..46ef895 100644 --- a/src/gateway/mod.rs +++ b/src/gateway/mod.rs @@ -1,6 +1,7 @@ pub mod dispatch; pub mod error; pub mod routes; +pub(crate) mod schema_cache; pub use dispatch::GatewayDispatch; pub use routes::CallRequest; diff --git a/src/gateway/routes.rs b/src/gateway/routes.rs index 418ccdc..e59894a 100644 --- a/src/gateway/routes.rs +++ b/src/gateway/routes.rs @@ -33,6 +33,7 @@ use serde_json::{json, Value}; use super::dispatch::GatewayDispatch; use super::error::call_error_to_http_response_with_identity; +use super::schema_cache::{CompileFailed, PublishSchemaCache}; use crate::server::auth::ResolvedIdentity; use crate::server::state::RouterState; @@ -53,16 +54,19 @@ type ByteStream = futures::stream::BoxStream<'static, Result pub(crate) struct GatewayState { registry: Arc, identity_provider: Arc, + publish_schemas: crate::gateway::schema_cache::PublishSchemaCache, } impl GatewayState { pub(crate) fn new( registry: Arc, identity_provider: Arc, + publish_schemas: crate::gateway::schema_cache::PublishSchemaCache, ) -> Self { Self { registry, identity_provider, + publish_schemas, } } @@ -79,6 +83,7 @@ impl FromRef for GatewayState { GatewayState::new( Arc::clone(&state.registry), Arc::clone(&state.identity_provider), + state.publish_schemas.clone(), ) } } @@ -219,13 +224,22 @@ pub type SubscribeStream = BoxStream<'static, Result>; /// a client disconnect drops the body stream and the sink (and the /// handler's `PublishStream`) sees EOF, matching call-protocol /// write-half close semantics. +/// +/// Chunk validation (GW-01, ADR-046 §4): each chunk is validated +/// against the op's `publish_schema` via the compile-once +/// [`PublishSchemaCache`]. A schema that **fails to compile** is +/// fail-closed, not fail-open (review-001 post-remediation follow-up): +/// the chunk stream terminates with `INTERNAL` before any chunk reaches +/// the handler — the unvalidated-ingest hole the per-request +/// `warn`-and-skip opened is shut. The compile error itself is logged +/// at error level and never echoed on the wire. pub(crate) async fn publish_handler( State(state): State, ResolvedIdentity(identity): ResolvedIdentity, body: axum::body::Body, ) -> Response { let mut header_lines = BufferedLines::new(Box::pin(body.into_data_stream())); - let (operation, first_chunk, validator): (String, Value, Option) = { + let (operation, first_chunk) = { let line = match header_lines.next_line().await { Ok(Some(line)) => line, Ok(None) => { @@ -251,25 +265,16 @@ pub(crate) async fn publish_handler( let Some(chunk) = value.get("chunk") else { return missing_header_field_response(); }; - let validator = state - .registry - .registration(op.strip_prefix('/').unwrap_or(&op)) - .and_then(|reg| reg.spec.publish_schema.clone()) - .and_then(|schema| match jsonschema::options().build(&schema) { - Ok(v) => Some(v), - Err(e) => { - tracing::warn!( - operation = %op, - error = %e, - "publish_schema failed to compile; chunks will not be validated" - ); - None - } - }); - (op, chunk.clone(), validator) + (op, chunk.clone()) }; - let chunks = NdjsonChunkStream::new(header_lines, validator, Some(first_chunk)); + let chunks = NdjsonChunkStream::new( + header_lines, + Arc::clone(&state.registry), + state.publish_schemas.clone(), + operation.clone(), + Some(first_chunk), + ); let dispatch = state.dispatch(); let envelope = dispatch .invoke_sink(identity.clone(), &operation, Value::Null, Box::pin(chunks)) @@ -277,6 +282,88 @@ pub(crate) async fn publish_handler( envelope_to_response(envelope, identity.as_ref()) } +/// The per-request `publish_schema` situation for a `/publish` call +/// (ADR-046 §4). Resolution is lazy — the chunk stream resolves against +/// the [`PublishSchemaCache`] on its first poll, which runs only after +/// `invoke_sink`'s pre-checks have admitted the operation (an unknown, +/// internal, or wrong-type op is rejected before any schema resolution +/// or compile happens, preserving the GW-11 dispatch-owned pre-check +/// order and keeping 404/403/422 intact). +enum PublishSchemaState { + /// Not yet resolved: the stream holds the registry + cache until + /// the first chunk is polled. + Unresolved { + registry: Arc, + cache: PublishSchemaCache, + operation: String, + }, + /// The op registers no `publish_schema`; chunks pass as-is (the + /// wire path's behavior for schema-less Pub ops). + Unvalidated, + /// Chunks are validated against this compiled validator. + Validated(std::sync::Arc), + /// The op's `publish_schema` failed to compile (GW-01 follow-up): + /// fail-closed — the stream yields one terminal `INTERNAL` error + /// and ends, so no chunk ever reaches the handler unvalidated. + Failed(CompileFailed), +} + +impl PublishSchemaState { + fn resolve(&mut self) { + let Self::Unresolved { + registry, + cache, + operation, + } = self + else { + return; + }; + let operation = std::mem::take(operation); + *self = match cache.validator(registry, &operation) { + Ok(None) => Self::Unvalidated, + Ok(Some(validator)) => Self::Validated(validator), + Err(compile_failed) => { + tracing::debug!( + operation = %operation, + "rejecting /publish: the operation's publish_schema could not be compiled \ + (fail-closed)" + ); + Self::Failed(compile_failed) + } + }; + } + + /// Validate one chunk; also flips a terminal compile failure into + /// stream-end after the first error item (an `Err` from this stream + /// is terminal — the handler sees the error and the stream closes, + /// mirroring the wire pump's `send(Err)` + `break`). + fn validate_chunk( + &mut self, + value: Value, + ) -> std::task::Poll>> { + self.resolve(); + match self { + Self::Unvalidated => std::task::Poll::Ready(Some(Ok(value))), + Self::Validated(validator) => { + if Validator::is_valid(validator, &value) { + std::task::Poll::Ready(Some(Ok(value))) + } else { + std::task::Poll::Ready(Some(Err(CallError::invalid_input( + "published chunk failed publish_schema validation", + ) + .with_details(json!({ "chunk": value }))))) + } + } + Self::Failed(failed) => { + let error = failed.call_error(); + *self = Self::Unvalidated; + std::task::Poll::Ready(Some(Err(error))) + } + Self::Unresolved { .. } => unreachable!("resolve() runs before validation"), + } + } +} + fn missing_header_field_response() -> Response { invalid_input_response("first publish line must carry {\"operation\": ..., \"chunk\": ...}") } @@ -371,40 +458,53 @@ impl LineError { /// validated against the op's compiled `publish_schema` — matching the /// wire dispatcher's per-chunk validation (alkcall ADR-046). Malformed /// JSON, a schema violation, or a body-read failure yields a terminal -/// `Err(INVALID_INPUT)` item — the same shape an initiator-side -/// `call.error` produces on the wire. +/// `Err` item — the same shape an initiator-side `call.error` produces +/// on the wire, and equally terminal: like the wire pump's +/// `send(Err)` + `break`, the first `Err` ends the stream. A schema +/// that failed to compile yields that terminal `Err(INTERNAL)` on the +/// first chunk — fail-closed (review-001 post-remediation follow-up). struct NdjsonChunkStream { lines: std::pin::Pin>, - validator: Option, + schema_state: PublishSchemaState, pending_first: Option, + done: bool, } impl NdjsonChunkStream { fn new( lines: BufferedLines, - validator: Option, + registry: Arc, + cache: PublishSchemaCache, + operation: String, pending_first: Option, ) -> Self { Self { lines: Box::pin(lines), - validator, + schema_state: PublishSchemaState::Unresolved { + registry, + cache, + operation, + }, pending_first, + done: false, } } - fn validate_chunk( + /// `Some(item)` from `validate_chunk` where an `Err` item sets + /// `done` — the stream does not yield after an error (the wire + /// pump's semantics; the dropping mpsc sender below is the same + /// contract). + fn poll_validated( &mut self, value: Value, ) -> std::task::Poll>> { - if let Some(validator) = &self.validator { - if !Validator::is_valid(validator, &value) { - return std::task::Poll::Ready(Some(Err(CallError::invalid_input( - "published chunk failed publish_schema validation", - ) - .with_details(json!({ "chunk": value }))))); + match self.schema_state.validate_chunk(value) { + std::task::Poll::Ready(Some(Err(e))) => { + self.done = true; + std::task::Poll::Ready(Some(Err(e))) } + other => other, } - std::task::Poll::Ready(Some(Ok(value))) } } @@ -415,30 +515,44 @@ impl futures::Stream for NdjsonChunkStream { mut self: std::pin::Pin<&mut Self>, cx: &mut std::task::Context<'_>, ) -> std::task::Poll> { + if self.done { + return std::task::Poll::Ready(None); + } if self.pending_first.is_some() { let first = self.pending_first.take().unwrap_or(Value::Null); - return self.validate_chunk(first); + return self.poll_validated(first); } let line = { let mut next = Box::pin(self.lines.next_line()); - match std::future::Future::poll(next.as_mut(), cx) { - std::task::Poll::Ready(Ok(Some(line))) => line, - std::task::Poll::Ready(Ok(None)) => return std::task::Poll::Ready(None), - std::task::Poll::Ready(Err(e)) => { - return std::task::Poll::Ready(Some(Err(CallError::invalid_input(e.message())))) + let polled = match std::future::Future::poll(next.as_mut(), cx) { + std::task::Poll::Ready(Ok(Some(line))) => Ok(Some(line)), + std::task::Poll::Ready(Ok(None)) => Ok(None), + std::task::Poll::Ready(Err(e)) => Err(e.message()), + std::task::Poll::Pending => { + drop(next); + return std::task::Poll::Pending; + } + }; + drop(next); + match polled { + Ok(Some(line)) => line, + Ok(None) => return std::task::Poll::Ready(None), + Err(message) => { + self.done = true; + return std::task::Poll::Ready(Some(Err(CallError::invalid_input(message)))); } - std::task::Poll::Pending => return std::task::Poll::Pending, } }; let value = match serde_json::from_slice::(&line) { Ok(v) => v, Err(e) => { + self.done = true; return std::task::Poll::Ready(Some(Err(CallError::invalid_input(format!( "publish line is not valid JSON: {e}" - ))))) + ))))); } }; - self.validate_chunk(value) + self.poll_validated(value) } } @@ -862,6 +976,7 @@ mod tests { crate::websocket::DEFAULT_WS_MAX_SESSIONS, )), ws_idle_timeout: Some(crate::websocket::DEFAULT_WS_IDLE_TIMEOUT), + publish_schemas: crate::gateway::schema_cache::PublishSchemaCache::new(), }; let auth_state = Arc::clone(&provider); gateway_router() @@ -2243,4 +2358,217 @@ mod tests { "the prematurely terminated body reaches the handler as an INVALID_INPUT Err item: {resp}" ); } + + // --- publish_schema fail-closed (GW-01 follow-up) ---------------------- + + fn registry_with_broken_publish_schema() -> Arc { + let mut registry = OperationRegistry::new(); + registry + .register(HandlerRegistration::new( + OperationSpec::new( + "ingest/broken", + OperationType::Pub, + Visibility::External, + json!({}), + json!({}), + vec![], + AccessControl::default(), + None, + ) + .with_publish_schema(json!({ "required": "n" })), // `required` must be an array + HandlerKind::Sink(make_sink_handler( + |_unused_input, ctx, mut chunks| async move { + use futures::StreamExt; + let mut collected: Vec = Vec::new(); + while let Some(chunk) = chunks.next().await { + match chunk { + Ok(v) => collected.push(v), + Err(e) => return ResponseEnvelope::error(ctx.request_id, e), + } + } + ResponseEnvelope::ok( + ctx.request_id, + json!({ "count": collected.len(), "chunks": collected }), + ) + }, + )), + OperationProvenance::Local, + None, + None, + Capabilities::new(), + )) + .unwrap(); + Arc::new(registry) + } + + #[tokio::test] + async fn publish_with_uncompilable_schema_fails_closed_with_internal() { + let router = build_router(registry_with_broken_publish_schema(), unused_provider()); + let body = ndjson(&[ + json!({ "operation": "ingest/broken", "chunk": { "n": 1 } }), + json!({ "n": 2 }), + ]); + let req = raw_request("POST", "/publish", body); + let (status, resp) = send(router, req).await; + assert_eq!( + status, + StatusCode::INTERNAL_SERVER_ERROR, + "an un-compilable publish_schema must fail the request loudly (500), not skip validation: {resp}" + ); + assert_eq!( + resp.get("code"), + Some(&json!("INTERNAL")), + "fail-closed maps to INTERNAL, not INVALID_INPUT (this is a server-side defect): {resp}" + ); + assert!( + resp.get("message") + .and_then(|m| m.as_str()) + .map(|m| m.contains("failed to compile")) + .unwrap_or(false), + "the wire message names the compile failure without echoing schema internals: {resp}" + ); + } + + #[tokio::test] + async fn publish_with_uncompilable_schema_never_delivers_chunks_to_the_handler() { + let mut registry = OperationRegistry::new(); + registry + .register(HandlerRegistration::new( + OperationSpec::new( + "ingest/spy", + OperationType::Pub, + Visibility::External, + json!({}), + json!({}), + vec![], + AccessControl::default(), + None, + ) + .with_publish_schema(json!({ "type": 123 })), // non-string `type` never compiles + HandlerKind::Sink(make_sink_handler(|_input, ctx, mut chunks| async move { + use futures::StreamExt; + let received: Vec> = chunks.by_ref().collect().await; + ResponseEnvelope::ok( + ctx.request_id, + json!({ + "chunk_count": received.len(), + "ok_chunks": received.iter().filter(|r| r.is_ok()).count(), + "error_codes": received + .iter() + .filter_map(|r| r.as_ref().err().map(|e| e.code.clone())) + .collect::>(), + }), + ) + })), + OperationProvenance::Local, + None, + None, + Capabilities::new(), + )) + .unwrap(); + let router = build_router(Arc::new(registry), unused_provider()); + let body = ndjson(&[ + json!({ "operation": "ingest/spy", "chunk": { "n": 1 } }), + json!({ "n": 2 }), + ]); + let (status, resp) = send(router, raw_request("POST", "/publish", body)).await; + let output = resp.get("output").expect("spy handler output"); + assert_eq!( + output.get("ok_chunks"), + Some(&json!(0)), + "no chunk may reach the handler when the publish_schema failed to compile: {resp}" + ); + assert_eq!( + output.get("chunk_count"), + Some(&json!(1)), + "exactly one item flows: the terminal INTERNAL error (stream ends after it): {resp}" + ); + assert_eq!( + output.get("error_codes"), + Some(&json!(["INTERNAL"])), + "the terminal item is the fail-closed INTERNAL compile error: {resp}" + ); + let _ = status; + } + + #[tokio::test] + async fn publish_hot_reload_replacement_schema_is_picked_up() { + let cache = PublishSchemaCache::new(); + let mut registry = OperationRegistry::new(); + registry + .register(HandlerRegistration::new( + OperationSpec::new( + "ingest/typed", + OperationType::Pub, + Visibility::External, + json!({}), + json!({}), + vec![], + AccessControl::default(), + None, + ) + .with_publish_schema(json!({ + "type": "object", + "properties": { "n": { "type": "integer" } }, + "required": ["n"], + "additionalProperties": false + })), + HandlerKind::Sink(make_sink_handler(|input, ctx, _chunks| async move { + ResponseEnvelope::ok(ctx.request_id, input) + })), + OperationProvenance::Local, + None, + None, + Capabilities::new(), + )) + .unwrap(); + + let v1 = cache + .validator(®istry, "ingest/typed") + .unwrap() + .expect("compiled validator"); + assert!(!v1.is_valid(&json!({ "n": 1, "extra": true }))); + + // Hot reload: re-register the same op with a schema that drops + // `additionalProperties`; the same cache entry must recompile. + registry + .register(HandlerRegistration::new( + OperationSpec::new( + "ingest/typed", + OperationType::Pub, + Visibility::External, + json!({}), + json!({}), + vec![], + AccessControl::default(), + None, + ) + .with_publish_schema(json!({ + "type": "object", + "properties": { "n": { "type": "integer" } }, + "required": ["n"] + })), + HandlerKind::Sink(make_sink_handler(|input, ctx, _chunks| async move { + ResponseEnvelope::ok(ctx.request_id, input) + })), + OperationProvenance::Local, + None, + None, + Capabilities::new(), + )) + .unwrap(); + + let v2 = cache + .validator(®istry, "ingest/typed") + .unwrap() + .expect("recompiled validator"); + assert!( + !Arc::ptr_eq(&v1, &v2), + "a changed schema value must invalidate the cached validator" + ); + assert!( + v2.is_valid(&json!({ "n": 1, "extra": true })), + "the recompiled schema must reflect the new registration" + ); + } } diff --git a/src/gateway/schema_cache.rs b/src/gateway/schema_cache.rs new file mode 100644 index 0000000..0ae2c55 --- /dev/null +++ b/src/gateway/schema_cache.rs @@ -0,0 +1,258 @@ +//! Compile-once cache for operation `publish_schema` validators +//! (ADR-046 §4, ADR-068; review-001 post-remediation follow-up). +//! +//! The `/publish` route validates each chunk against the op's +//! `publish_schema`. Compiling the schema per request is both a +//! per-request CPU/allocation cost on the hot path and — worse — the +//! GW-01 follow-up fail-open: when compilation fails, chunks flowed to +//! the handler **unvalidated** behind only a `tracing::warn`, quietly +//! reopening the transport-dependent validation gap GW-01 closed. +//! +//! This cache compiles each op's schema **once** and remembers the +//! outcome — [`PublishSchemaCache::validator`] returns either a shared +//! compiled validator or a cached compile failure. The failure is not +//! retried until the registration changes: an un-compilable schema is a +//! deployment defect (registry assembly), not a transient condition. +//! +//! Invalidation is by schema **value**: the raw `serde_json::Value` is +//! stored alongside the compiled validator, and a request whose op's +//! current `publish_schema` differs from the cached one recompiles. This +//! makes re-registration (hot reload) correct without lifetime coupling +//! to registry internals: a new schema value is picked up on the next +//! `/publish` request to that op. +//! +//! The registry is defined not to mutate after assembly +//! (`CachedOpenAPIDoc` relies on the same invariant), so in practice +//! each schema compiles once per process — the value comparison is +//! defense in depth for re-assembly paths. + +use std::collections::HashMap; +use std::sync::Arc; + +use alkcall::protocol::wire::CallError; +use alkcall::registry::registration::OperationRegistry; +use jsonschema::Validator; +use parking_lot::RwLock; +use serde_json::Value; + +/// Compiled-validator (or recorded compile failure) per operation name. +#[derive(Clone, Default)] +pub(crate) struct PublishSchemaCache { + entries: Arc>>, +} + +#[derive(Clone)] +enum CacheEntry { + Compiled { + schema: Value, + validator: Arc, + }, + /// Compile failed; the error was already logged at compile-attempt + /// time and every subsequent request fails closed. Held only so a + /// schema whose value still equals the failing one is not + /// recompiled (and re-logged) on every request. + Failed { schema: Value }, +} + +impl PublishSchemaCache { + pub(crate) fn new() -> Self { + Self::default() + } + + /// The compiled validator for `op`'s current `publish_schema`, from + /// the cache when the schema value is unchanged, compiled otherwise. + /// + /// `Ok(None)` means the op has no `publish_schema` (chunks flow + /// unvalidated, matching the wire path). `Err` is a **failed** + /// compile — cached, loud, never a validator: the caller must fail + /// the publish rather than skip validation. + pub(crate) fn validator( + &self, + registry: &OperationRegistry, + op: &str, + ) -> Result>, CompileFailed> { + let Some(schema) = registry + .registration(op) + .and_then(|reg| reg.spec.publish_schema.clone()) + else { + return Ok(None); + }; + if let Some(cached) = Self::lookup(&self.entries, op, &schema) { + return cached; + } + Self::compile(&self.entries, op, schema) + } + + fn lookup( + entries: &RwLock>, + op: &str, + schema: &Value, + ) -> Option>, CompileFailed>> { + let guard = entries.read(); + match guard.get(op) { + Some(CacheEntry::Compiled { + schema: cached_schema, + validator, + }) if cached_schema == schema => Some(Ok(Some(Arc::clone(validator)))), + Some(CacheEntry::Failed { + schema: cached_schema, + }) if cached_schema == schema => Some(Err(CompileFailed { + operation: op.to_string(), + })), + _ => None, + } + } + + fn compile( + entries: &RwLock>, + op: &str, + schema: Value, + ) -> Result>, CompileFailed> { + match jsonschema::options().build(&schema) { + Ok(validator) => { + let validator = Arc::new(validator); + entries.write().insert( + op.to_string(), + CacheEntry::Compiled { + schema, + validator: Arc::clone(&validator), + }, + ); + Ok(Some(validator)) + } + Err(error) => { + tracing::error!( + operation = %op, + error = %error, + "publish_schema failed to compile; every /publish request to this operation \ + fails (fail-closed) until the schema is corrected and re-registered" + ); + entries + .write() + .insert(op.to_string(), CacheEntry::Failed { schema }); + Err(CompileFailed { + operation: op.to_string(), + }) + } + } + } +} + +/// A `publish_schema` that could not be compiled (review-001 +/// post-remediation, GW-01 follow-up): the publish request fails closed +/// with `INTERNAL`; the compile error itself stays in the server log — +/// an untrusted schema's error text must not be echoed on the wire +/// (same discipline as the `/openapi.json` cache-miss 500). +#[derive(Debug)] +pub(crate) struct CompileFailed { + operation: String, +} + +impl CompileFailed { + pub(crate) fn call_error(&self) -> CallError { + CallError::internal(format!( + "publish_schema for operation '{}' failed to compile; publishing is disabled for \ + this operation", + self.operation + )) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use alkcall::core::types::Capabilities; + use alkcall::registry::registration::{ + make_sink_handler, HandlerKind, HandlerRegistration, OperationProvenance, + }; + use alkcall::registry::spec::{AccessControl, OperationSpec, OperationType, Visibility}; + use serde_json::json; + + fn registry_with_pub_op(name: &str, schema: Option) -> OperationRegistry { + let mut registry = OperationRegistry::new(); + let mut spec = OperationSpec::new( + name, + OperationType::Pub, + Visibility::External, + json!({}), + json!({}), + vec![], + AccessControl::default(), + None, + ); + if let Some(schema) = schema { + spec = spec.with_publish_schema(schema); + } + registry + .register(HandlerRegistration::new( + spec, + HandlerKind::Sink(make_sink_handler(|_input, ctx, _chunks| async move { + alkcall::protocol::wire::ResponseEnvelope::ok(ctx.request_id, Value::Null) + })), + OperationProvenance::Local, + None, + None, + Capabilities::new(), + )) + .unwrap(); + registry + } + + #[test] + fn no_publish_schema_yields_ok_none() { + let cache = PublishSchemaCache::new(); + let registry = registry_with_pub_op("ingest/push", None); + assert!(matches!( + cache.validator(®istry, "ingest/push"), + Ok(None) + )); + } + + #[test] + fn unknown_op_yields_ok_none() { + let cache = PublishSchemaCache::new(); + let registry = OperationRegistry::new(); + assert!(matches!(cache.validator(®istry, "no/such"), Ok(None))); + } + + #[test] + fn valid_schema_compiles_and_validates() { + let cache = PublishSchemaCache::new(); + let registry = registry_with_pub_op( + "ingest/typed", + Some(json!({ "type": "object", "required": ["n"] })), + ); + let validator = cache.validator(®istry, "ingest/typed").unwrap().unwrap(); + assert!(validator.is_valid(&json!({ "n": 1 }))); + assert!(!validator.is_valid(&json!({}))); + } + + #[test] + fn repeated_lookups_return_the_same_cached_validator() { + let cache = PublishSchemaCache::new(); + let registry = registry_with_pub_op( + "ingest/typed", + Some(json!({ "type": "object", "required": ["n"] })), + ); + let first = cache.validator(®istry, "ingest/typed").unwrap().unwrap(); + let second = cache.validator(®istry, "ingest/typed").unwrap().unwrap(); + assert!( + Arc::ptr_eq(&first, &second), + "cache must serve the same compiled validator" + ); + } + + #[test] + fn uncompilable_schema_fails_closed_and_stays_failed() { + let cache = PublishSchemaCache::new(); + let registry = registry_with_pub_op( + "ingest/broken", + Some(json!({ "required": "n" })), // `required` must be an array + ); + assert!(cache.validator(®istry, "ingest/broken").is_err()); + assert!( + cache.validator(®istry, "ingest/broken").is_err(), + "the cached failure must persist — no retry, no validator" + ); + } +} diff --git a/src/server/adapter.rs b/src/server/adapter.rs index 23777a8..35202db 100644 --- a/src/server/adapter.rs +++ b/src/server/adapter.rs @@ -127,6 +127,7 @@ impl HttpAdapter { ws_sessions: Arc::clone(&ws_sessions), ws_session_slots: Arc::clone(&ws_session_slots), ws_idle_timeout, + publish_schemas: crate::gateway::schema_cache::PublishSchemaCache::new(), }; let router = build_router(state, None); Self { @@ -154,6 +155,7 @@ impl HttpAdapter { ws_sessions: Arc::clone(&self.ws_sessions), ws_session_slots: Arc::clone(&self.ws_session_slots), ws_idle_timeout: self.ws_idle_timeout, + publish_schemas: crate::gateway::schema_cache::PublishSchemaCache::new(), }; // `extra_routes` is borrowed, not consumed (SRV-05): a builder // call after `with_extra_routes` must keep the custom routes in @@ -172,6 +174,7 @@ impl HttpAdapter { ws_sessions: Arc::clone(&self.ws_sessions), ws_session_slots: Arc::clone(&self.ws_session_slots), ws_idle_timeout: self.ws_idle_timeout, + publish_schemas: crate::gateway::schema_cache::PublishSchemaCache::new(), }; self.router = build_router(state, Some(routes.clone())); self.extra_routes = Some(routes); @@ -195,6 +198,7 @@ impl HttpAdapter { ws_sessions: Arc::clone(&self.ws_sessions), ws_session_slots: Self::rebuild_session_slots(max_sessions), ws_idle_timeout: self.ws_idle_timeout, + publish_schemas: crate::gateway::schema_cache::PublishSchemaCache::new(), }; self.router = build_router(state, self.extra_routes.clone()); self @@ -217,6 +221,7 @@ impl HttpAdapter { ws_sessions: Arc::clone(&self.ws_sessions), ws_session_slots: Arc::clone(&self.ws_session_slots), ws_idle_timeout: self.ws_idle_timeout, + publish_schemas: crate::gateway::schema_cache::PublishSchemaCache::new(), }; self.router = build_router(state, self.extra_routes.clone()); self @@ -1025,6 +1030,7 @@ mod tests { crate::websocket::DEFAULT_WS_MAX_SESSIONS, )), ws_idle_timeout: Some(crate::websocket::DEFAULT_WS_IDLE_TIMEOUT), + publish_schemas: crate::gateway::schema_cache::PublishSchemaCache::new(), } } diff --git a/src/server/state.rs b/src/server/state.rs index 1acdc9a..77ad532 100644 --- a/src/server/state.rs +++ b/src/server/state.rs @@ -42,6 +42,10 @@ pub(crate) struct RouterState { pub(crate) ws_session_slots: Arc, /// Idle-read timeout for the WS pumps (WS-01); `None` disables. pub(crate) ws_idle_timeout: Option, + /// Compile-once `publish_schema` validator cache for the `/publish` + /// route (GW-01 follow-up); built empty at adapter construction and + /// populated lazily per operation. + pub(crate) publish_schemas: crate::gateway::schema_cache::PublishSchemaCache, } impl axum::extract::FromRef for crate::websocket::SessionState { @@ -102,6 +106,7 @@ mod tests { crate::websocket::DEFAULT_WS_MAX_SESSIONS, )), ws_idle_timeout: Some(crate::websocket::DEFAULT_WS_IDLE_TIMEOUT), + publish_schemas: crate::gateway::schema_cache::PublishSchemaCache::new(), }; let extracted: DecoyConfig = axum::extract::FromRef::from_ref(&state); assert!(matches!(extracted, DecoyConfig::Redirect { .. })); diff --git a/tasks/gateway/review-001-publish-schema-validation-robust.md b/tasks/gateway/review-001-publish-schema-validation-robust.md index 1b1f600..fe3202b 100644 --- a/tasks/gateway/review-001-publish-schema-validation-robust.md +++ b/tasks/gateway/review-001-publish-schema-validation-robust.md @@ -1,7 +1,7 @@ --- id: review-001-publish-schema-validation-robust name: Fix /publish schema validation fail-open + per-request recompilation (post-remediation) -status: pending +status: completed depends_on: [] scope: narrow risk: high @@ -41,11 +41,11 @@ schema originates), with the gateway cache as defense in depth. ## Acceptance Criteria -- [ ] Compiled validator (or error) cached per registration; no per-request recompile (test: compile count / perf shape not asserted, but code path is registration-keyed) -- [ ] Un-compilable `publish_schema` → request fails loudly (500/INTERNAL, error logged), chunks never flow unvalidated (test) -- [ ] Schema re-registration (hot reload) picks up the new schema (test) -- [ ] A `publish_schema`-registered Pub op still rejects an invalid chunk (existing GW-01 gate stays green) -- [ ] `cargo test` and `cargo clippy --all-targets -- -D warnings` pass +- [x] Compiled validator (or error) cached per registration; no per-request recompile (test: compile count / perf shape not asserted, but code path is registration-keyed) +- [x] Un-compilable `publish_schema` → request fails loudly (500/INTERNAL, error logged), chunks never flow unvalidated (test) +- [x] Schema re-registration (hot reload) picks up the new schema (test) +- [x] A `publish_schema`-registered Pub op still rejects an invalid chunk (existing GW-01 gate stays green) +- [x] `cargo test` and `cargo clippy --all-targets -- -D warnings` pass ## References @@ -55,13 +55,78 @@ schema originates), with the gateway cache as defense in depth. ## Notes -> Agent fills during implementation. Found in the post-remediation -> sweep — not a Review 001 finding itself. Fail-open is the priority -> half; caching is the efficiency half. Registration-side (adapter) -> rejection at import is the preferred end state if alkcall's -> registration surface allows observing `publish_schema` at -> HandlerRegistration time. +Mechanism: new `src/gateway/schema_cache.rs` — +`PublishSchemaCache`, a cloneable `Arc>>` held on `RouterState` (built empty per +`HttpAdapter`/router construction; `GatewayState` extracts it via +`FromRef`). Each entry stores the compiled `Arc` +(or a `Failed` marker) **plus the raw schema `Value`** that produced +it. Lookup compares the registry's current `publish_schema` value +against the cached one: equal → serve the cached validator/failure; +different → recompile. This gives hot-reload correctness by value +equality without lifetime-coupling to registry internals; under the +documented assemble-then-serve invariant (the same one +`CachedOpenAPIDoc` relies on) each schema compiles exactly once per +process, and the value check is defense in depth for re-assembly +paths. + +Fail-closed semantics: a failed compile is **cached as a failure** — +the error is logged once at `error` level (never retried or re-logged +per request), and the wire message is a generic INTERNAL naming the +compile failure without echoing schema internals (an untrusted +schema's error text must not reach the wire; same discipline as the +`/openapi.json` cache-miss 500). + +Resolution is **lazy**, resolved against the cache inside the chunk +stream's first poll, not in the handler prologue: `invoke_sink` owns +the 404/403/422 pre-checks (GW-11), so an unknown/internal/wrong-type +op is rejected before any schema lookup or compile happens — no cache +population or error-log spam from ops a caller cannot reach. + +Stream terminality fix (found by the spy-handler test): the old +`NdjsonChunkStream` yielded `Err` items but kept streaming — a handler +that drained the stream (instead of aborting on first `Err`) would +have kept receiving **unvalidated** chunks after a violation or a +compile failure. The wire pump (`alkcall dispatch.rs:599-600`) does +`send(Err)` + `break`, i.e. the error item is terminal and the channel +closes. `NdjsonChunkStream` now mirrors that exactly: the first `Err` +item (schema violation, bad JSON, read failure, or compile failure) +sets `done` and the stream ends — `Ok` chunks can never flow after an +error on this transport either. + +Adapter-side registration-time rejection was considered and **not** +pursued here: alkcall's `OperationRegistry::register` does not observe +`publish_schema` compilability, and adding that would be an alkcall +surface change (this crate consumes `alkcall = "0.1.1"` from crates- +io) — the gateway cache is the defense in depth the task prescribed. +Worth noting for alkcall: the identical compile-fail-open pattern +exists on the wire path (`alkcall/src/protocol/dispatch.rs:356-366`, +warn-and-skip) — upstream fix candidate, out of scope for this crate. ## Summary -> Filled on completion. \ No newline at end of file +Implemented in three parts: + +- **`src/gateway/schema_cache.rs`**: `PublishSchemaCache` — compile- + once, value-keyed invalidation, cached failures, `error!`-level compile + logging, generic non-leaking wire error (`CompileFailed::call_error` + → `CallError::internal`). 5 unit tests (no-schema/unknown-op → + `Ok(None)`; compile+validate; `Arc::ptr_eq` cache identity; failure + stays failed). +- **`src/gateway/routes.rs`**: `publish_handler` uses the cache; + `PublishSchemaState` (Unresolved → Unvalidated/Validated/Failed) + resolves lazily on first chunk poll, after `invoke_sink`'s + pre-checks; `NdjsonChunkStream` is terminal-on-first-`Err` (wire + parity). Route tests: uncompilable schema → 500 `INTERNAL` with + non-echoing message; spy handler receives 0 `Ok` chunks and exactly + one terminal `INTERNAL` error; hot-reload re-registration picks up + the replacement schema (validator identity + semantics asserted); + all pre-existing GW-01 gates (reject-invalid, accept-valid, + first-chunk validation, no-schema passthrough) stay green. +- **`src/server/{state,adapter}.rs`** + `gateway/mod.rs`: cache plumbed + through `RouterState`/`GatewayState` at all construction sites. + +Verification: `cargo test` 308 passed (default), `--all-features` all +suites green (379+), `cargo clippy --all-targets -- -D warnings` +clean (default + all-features), `cargo fmt --check` clean. +`--no-default-features` warnings are pre-existing on the base commit. \ No newline at end of file