fix(gateway): publish validation + streaming + batch semantics (GW-01, GW-06, GW-08..GW-11, HY-13)
- GW-01: /publish validates every NDJSON chunk against the op's publish_schema (incl. the first-line chunk) via NdjsonChunkStream — terminal Err(INVALID_INPUT)/422 on violation, matching the wire dispatcher's per-chunk contract. Route-level fix; the alkcall spine was explored and rejected (wire validation is pump-side by design). - GW-06: the body is streamed, not buffered — Body::into_data_stream() -> newline-framed BufferedLines -> lazily parsed chunk stream. ADR-068 documents the streamed semantics and the 2 MiB per-line cap. - GW-08: /batch capped at 100 operations (INVALID_INPUT 400). - GW-09: internal-op batch entries now carry generated UUID request ids. - GW-10: first publish line missing `chunk` is rejected INVALID_INPUT. - GW-11: redundant /publish pre-checks removed; enforcement rides on invoke_sink via the shared dispatch spine. - HY-13: the vacuous stub test was replaced by a body-cut-short test. - Adjacent: INVALID_OPERATION_TYPE now maps 422 (with identity) / 401 (without) in error.rs — the route relies on the shared mapper since the pre-checks are gone (GW-03's finding; was a 500 fall-through). Verification: cargo test 211 passed; cargo clippy --all-targets -- -D warnings clean; cargo fmt --check clean.
This commit is contained in:
Generated
+1
@@ -61,6 +61,7 @@ dependencies = [
|
||||
"httpdate",
|
||||
"hyper",
|
||||
"hyper-util",
|
||||
"jsonschema",
|
||||
"openapiv3",
|
||||
"parking_lot",
|
||||
"reqwest",
|
||||
|
||||
@@ -45,6 +45,7 @@ openapiv3 = "2"
|
||||
http = "1"
|
||||
url = "2"
|
||||
bytes = "1"
|
||||
jsonschema = { version = "0.46", default-features = false }
|
||||
parking_lot = "0.12"
|
||||
rmcp = { version = "1.8", optional = true, default-features = false, features = [
|
||||
"client",
|
||||
|
||||
@@ -67,6 +67,23 @@ stream; the operation's final `ResponseEnvelope` is the HTTP response.
|
||||
([ADR-023](023-operation-error-schemas.md), the gateway's
|
||||
`HTTP_<status>` fidelity rules).
|
||||
|
||||
### Body handling (streamed, not buffered)
|
||||
|
||||
The NDJSON body is **streamed, never fully buffered** (GW-06): axum's
|
||||
`Body` is framed into lines as bytes arrive, and each line is parsed and
|
||||
pushed into the sink lazily — memory is bounded by the per-line cap (2
|
||||
MiB, matching axum's default whole-body limit), not by the unbounded
|
||||
chunk count. `publish_schema` validation is applied per chunk inside the
|
||||
sink-feeding stream (GW-01), so a Pub op registered with a
|
||||
`publish_schema` enforces the same per-chunk contract over HTTP as over
|
||||
the call protocol; a violation terminates the chunk stream with
|
||||
`INVALID_INPUT` (→ 422), the exact item shape an initiator-side
|
||||
`call.error` produces on the wire. A first line missing `chunk` (GW-10)
|
||||
is rejected `INVALID_INPUT` before dispatch — indistinguishable from a
|
||||
missing `operation`. The client disconnect abort path is unchanged: the
|
||||
dropped body stream propagates EOF through the framed reader into the
|
||||
sink's `PublishStream`.
|
||||
|
||||
### Wire-shape note
|
||||
|
||||
On the call protocol, the initiator's chunks are `call.published`
|
||||
@@ -105,6 +122,10 @@ the doc does not preload operations.
|
||||
- The dispatch path is `invoke_sink()` — the same spine as `/call`'s
|
||||
`invoke()`, so the shared-dispatch invariants (identity, ACL,
|
||||
Internal filtering) hold by construction.
|
||||
- The body is streamed line-by-line (memory bounded by the per-line
|
||||
cap; backpressure inherited from the HTTP body), so a client
|
||||
disconnect mid-stream cancels the sink exactly like a dropped
|
||||
write-half on the call protocol.
|
||||
|
||||
**Negative:**
|
||||
|
||||
@@ -118,6 +139,11 @@ the doc does not preload operations.
|
||||
the `/openapi.json` version bumps~~ — settled: first-line
|
||||
`{operation, chunk}` convention; terminal errors are plain HTTP
|
||||
status + JSON body (not an NDJSON line).
|
||||
- The 2 MiB per-line cap (not a whole-body cap) bounds a single chunk;
|
||||
the total number of chunks is unbounded. Handlers that would receive
|
||||
unbounded streams over the wire get the same behavior over HTTP —
|
||||
operators front the endpoint with the same body/timeout controls used
|
||||
for any other streaming surface.
|
||||
|
||||
## References
|
||||
|
||||
|
||||
+28
-4
@@ -1,10 +1,10 @@
|
||||
//! CallError → HTTP status/response mapping ([ADR-023]).
|
||||
//!
|
||||
//! Protocol-level vs operation-level code distinction: protocol codes
|
||||
//! (`NOT_FOUND`, `FORBIDDEN`, `INVALID_INPUT`, `TIMEOUT`, `INTERNAL`)
|
||||
//! map to fixed statuses; operation-level codes imported from external
|
||||
//! HTTP APIs are prefixed `HTTP_<status>` and map to their declared
|
||||
//! status.
|
||||
//! (`NOT_FOUND`, `FORBIDDEN`, `INVALID_INPUT`, `INVALID_OPERATION_TYPE`,
|
||||
//! `TIMEOUT`, `INTERNAL`) map to fixed statuses; operation-level codes
|
||||
//! imported from external HTTP APIs are prefixed `HTTP_<status>` and map
|
||||
//! to their declared status.
|
||||
//!
|
||||
//! [ADR-023]: https://docs.rs/alkhttp (docs/architecture/decisions)
|
||||
|
||||
@@ -17,6 +17,7 @@ use serde_json::Value;
|
||||
const PROTOCOL_CODE_NOT_FOUND: &str = "NOT_FOUND";
|
||||
const PROTOCOL_CODE_FORBIDDEN: &str = "FORBIDDEN";
|
||||
const PROTOCOL_CODE_INVALID_INPUT: &str = "INVALID_INPUT";
|
||||
const PROTOCOL_CODE_INVALID_OPERATION_TYPE: &str = "INVALID_OPERATION_TYPE";
|
||||
const PROTOCOL_CODE_TIMEOUT: &str = "TIMEOUT";
|
||||
const PROTOCOL_CODE_INTERNAL: &str = "INTERNAL";
|
||||
|
||||
@@ -49,6 +50,13 @@ pub fn call_error_to_http_status_with_identity(
|
||||
}
|
||||
}
|
||||
PROTOCOL_CODE_INVALID_INPUT => STATUS_UNPROCESSABLE,
|
||||
PROTOCOL_CODE_INVALID_OPERATION_TYPE => {
|
||||
if identity.is_some() {
|
||||
STATUS_UNPROCESSABLE
|
||||
} else {
|
||||
STATUS_UNAUTHORIZED
|
||||
}
|
||||
}
|
||||
PROTOCOL_CODE_TIMEOUT => STATUS_TIMEOUT,
|
||||
PROTOCOL_CODE_INTERNAL => STATUS_INTERNAL,
|
||||
code if code.starts_with(HTTP_PREFIX) => code[HTTP_PREFIX.len()..]
|
||||
@@ -123,6 +131,22 @@ mod tests {
|
||||
assert_eq!(call_error_to_http_status(&error), 422);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn invalid_operation_type_with_identity_maps_to_422() {
|
||||
let error = CallError::invalid_operation_type("not a Pub op");
|
||||
let id = identity();
|
||||
assert_eq!(
|
||||
call_error_to_http_status_with_identity(&error, Some(&id)),
|
||||
422
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn invalid_operation_type_without_identity_maps_to_401() {
|
||||
let error = CallError::invalid_operation_type("not a Pub op");
|
||||
assert_eq!(call_error_to_http_status_with_identity(&error, None), 401);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn timeout_maps_to_504() {
|
||||
let error = CallError::timeout("timed out");
|
||||
|
||||
+469
-129
@@ -1,16 +1,12 @@
|
||||
//! The 5 fixed gateway endpoints (`/search`, `/schema`, `/call`,
|
||||
//! `/batch`, `/subscribe`) — the sole HTTP invoke path (ADR-042,
|
||||
//! ADR-047).
|
||||
//!
|
||||
//! Each endpoint delegates to `GatewayDispatch::invoke()` (the shared
|
||||
//! dispatch spine); auth is the shared `bearer_auth_middleware`; error
|
||||
//! mapping is `gateway::error`. There is no per-operation
|
||||
//! `POST /{service}/{op}` direct-call surface (ADR-047). `/publish`
|
||||
//! (ADR-068) is a separate module.
|
||||
|
||||
use std::collections::VecDeque;
|
||||
use std::convert::Infallible;
|
||||
use std::sync::Arc;
|
||||
|
||||
use alkcall::core::auth::{Identity, IdentityProvider};
|
||||
use alkcall::protocol::wire::{CallError, ResponseEnvelope};
|
||||
use alkcall::registry::registration::OperationRegistry;
|
||||
use alkcall::registry::spec::{AccessResult, Visibility};
|
||||
use axum::body::Bytes;
|
||||
use axum::extract::{FromRef, Query, State};
|
||||
use axum::http::StatusCode;
|
||||
use axum::response::sse::Event;
|
||||
@@ -19,14 +15,10 @@ use axum::routing::{get, post};
|
||||
use axum::Router;
|
||||
use futures::stream::{self, BoxStream};
|
||||
use futures::StreamExt;
|
||||
use jsonschema::Validator;
|
||||
use serde::Deserialize;
|
||||
use serde_json::{json, Value};
|
||||
|
||||
use alkcall::core::auth::{Identity, IdentityProvider};
|
||||
use alkcall::protocol::wire::{CallError, ResponseEnvelope};
|
||||
use alkcall::registry::registration::OperationRegistry;
|
||||
use alkcall::registry::spec::{AccessResult, OperationType, Visibility};
|
||||
|
||||
use super::dispatch::GatewayDispatch;
|
||||
use super::error::{call_error_to_http_response, call_error_to_http_status_with_identity};
|
||||
use crate::server::auth::ResolvedIdentity;
|
||||
@@ -34,6 +26,10 @@ use crate::server::state::RouterState;
|
||||
|
||||
const SERVICES_LIST: &str = "services/list";
|
||||
const SERVICES_SCHEMA: &str = "services/schema";
|
||||
const MAX_BATCH_OPERATIONS: usize = 100;
|
||||
const MAX_PUBLISH_LINE_BYTES: usize = 2 * 1024 * 1024;
|
||||
|
||||
type ByteStream = futures::stream::BoxStream<'static, Result<Bytes, axum::Error>>;
|
||||
|
||||
#[derive(Clone)]
|
||||
pub(crate) struct GatewayState {
|
||||
@@ -141,6 +137,18 @@ pub(crate) async fn batch_handler(
|
||||
ResolvedIdentity(identity): ResolvedIdentity,
|
||||
Json(requests): Json<Vec<CallRequest>>,
|
||||
) -> Response {
|
||||
if requests.len() > MAX_BATCH_OPERATIONS {
|
||||
return (
|
||||
StatusCode::BAD_REQUEST,
|
||||
Json(json!({
|
||||
"code": "INVALID_INPUT",
|
||||
"message": format!(
|
||||
"batch exceeds the maximum of {MAX_BATCH_OPERATIONS} operations"
|
||||
),
|
||||
})),
|
||||
)
|
||||
.into_response();
|
||||
}
|
||||
let dispatch = state.dispatch();
|
||||
let mut results: Vec<Value> = Vec::with_capacity(requests.len());
|
||||
for request in requests {
|
||||
@@ -178,122 +186,233 @@ pub type SubscribeStream = BoxStream<'static, Result<Event, Infallible>>;
|
||||
/// per line. OQ-02 resolution: the first line carries
|
||||
/// `{ "operation": "/{service}/{op}", "chunk": {...} }` (subsequent
|
||||
/// lines are chunk values only); a terminal error is a plain HTTP
|
||||
/// status + JSON body (not an NDJSON line). A client disconnect drops
|
||||
/// the body stream — the sink (and the handler's `PublishStream`) sees
|
||||
/// EOF, matching call-protocol write-half close semantics.
|
||||
/// status + JSON body (not an NDJSON line). The body is streamed (never
|
||||
/// fully buffered): the first non-blank line names the operation and
|
||||
/// carries the first chunk, then the remainder streams into the sink —
|
||||
/// a client disconnect drops the body stream and the sink (and the
|
||||
/// handler's `PublishStream`) sees EOF, matching call-protocol
|
||||
/// write-half close semantics.
|
||||
pub(crate) async fn publish_handler(
|
||||
State(state): State<GatewayState>,
|
||||
ResolvedIdentity(identity): ResolvedIdentity,
|
||||
body: axum::body::Bytes,
|
||||
body: axum::body::Body,
|
||||
) -> Response {
|
||||
let mut lines = body
|
||||
.split(|b| *b == b'\n')
|
||||
.filter(|l| !l.iter().all(|b| b.is_ascii_whitespace()));
|
||||
let first = match lines.next() {
|
||||
Some(l) => l,
|
||||
None => {
|
||||
return (
|
||||
StatusCode::BAD_REQUEST,
|
||||
Json(json!({
|
||||
"code": "INVALID_INPUT",
|
||||
"message": "empty publish body: expected NDJSON chunks",
|
||||
})),
|
||||
)
|
||||
.into_response()
|
||||
let mut header_lines = BufferedLines::new(Box::pin(body.into_data_stream()));
|
||||
let (operation, first_chunk, validator): (String, Value, Option<jsonschema::Validator>) = {
|
||||
let line = match header_lines.next_line().await {
|
||||
Ok(Some(line)) => line,
|
||||
Ok(None) => {
|
||||
return invalid_input_response("empty publish body: expected NDJSON chunks")
|
||||
}
|
||||
Err(e) => return invalid_input_response(e.message()),
|
||||
};
|
||||
let value = match serde_json::from_slice::<Value>(&line) {
|
||||
Ok(v) => v,
|
||||
Err(e) => {
|
||||
return invalid_input_response(&format!(
|
||||
"first publish line is not valid JSON: {e}"
|
||||
))
|
||||
}
|
||||
};
|
||||
|
||||
// First line: { "operation": "...", "chunk": {...} } — OQ-02.
|
||||
let (operation, first_chunk): (String, Value) = match serde_json::from_slice::<Value>(first) {
|
||||
Ok(v) => {
|
||||
let op = v
|
||||
let op = value
|
||||
.get("operation")
|
||||
.and_then(|o| o.as_str())
|
||||
.map(str::to_string);
|
||||
let chunk = v.get("chunk").cloned().unwrap_or(Value::Null);
|
||||
match op {
|
||||
Some(op) => (op, chunk),
|
||||
None => {
|
||||
return (
|
||||
StatusCode::BAD_REQUEST,
|
||||
Json(json!({
|
||||
"code": "INVALID_INPUT",
|
||||
"message": "first publish line must carry {\"operation\": ..., \"chunk\": ...}",
|
||||
})),
|
||||
)
|
||||
.into_response()
|
||||
}
|
||||
}
|
||||
}
|
||||
let Some(op) = op else {
|
||||
return missing_header_field_response();
|
||||
};
|
||||
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) => {
|
||||
return (
|
||||
StatusCode::BAD_REQUEST,
|
||||
Json(json!({
|
||||
"code": "INVALID_INPUT",
|
||||
"message": format!("first publish line is not valid JSON: {e}"),
|
||||
})),
|
||||
)
|
||||
.into_response()
|
||||
tracing::warn!(
|
||||
operation = %op,
|
||||
error = %e,
|
||||
"publish_schema failed to compile; chunks will not be validated"
|
||||
);
|
||||
None
|
||||
}
|
||||
});
|
||||
(op, chunk.clone(), validator)
|
||||
};
|
||||
|
||||
if is_internal_op(&state.registry, &operation) {
|
||||
return not_found_response(&operation);
|
||||
}
|
||||
if state
|
||||
.registry
|
||||
.registration(operation.strip_prefix('/').unwrap_or(&operation))
|
||||
.is_none()
|
||||
{
|
||||
return not_found_response(&operation);
|
||||
}
|
||||
if let Some(forbidden) = access_check_for_op(&state.registry, &operation, identity.as_ref()) {
|
||||
return forbidden_response(forbidden, identity.as_ref());
|
||||
}
|
||||
if !is_pub_op(&state.registry, &operation) {
|
||||
return invalid_operation_type_response(&operation);
|
||||
}
|
||||
|
||||
let chunks: Vec<Result<Value, CallError>> = std::iter::once(Ok(first_chunk))
|
||||
.chain(
|
||||
lines.map(|line| match serde_json::from_slice::<Value>(line) {
|
||||
Ok(chunk) => Ok(chunk),
|
||||
Err(e) => Err(CallError::invalid_input(format!(
|
||||
"publish line is not valid JSON: {e}"
|
||||
))),
|
||||
}),
|
||||
)
|
||||
.collect();
|
||||
|
||||
let chunks = NdjsonChunkStream::new(header_lines, validator, Some(first_chunk));
|
||||
let dispatch = state.dispatch();
|
||||
let envelope = dispatch
|
||||
.invoke_sink(
|
||||
identity.clone(),
|
||||
&operation,
|
||||
Value::Null,
|
||||
Box::pin(futures::stream::iter(chunks)),
|
||||
)
|
||||
.invoke_sink(identity.clone(), &operation, Value::Null, Box::pin(chunks))
|
||||
.await;
|
||||
envelope_to_response(envelope, identity.as_ref())
|
||||
}
|
||||
|
||||
fn is_pub_op(registry: &OperationRegistry, operation: &str) -> bool {
|
||||
let name = operation.strip_prefix('/').unwrap_or(operation);
|
||||
match registry.registration(name) {
|
||||
Some(reg) => reg.spec.op_type == OperationType::Pub,
|
||||
None => false,
|
||||
fn missing_header_field_response() -> Response {
|
||||
invalid_input_response("first publish line must carry {\"operation\": ..., \"chunk\": ...}")
|
||||
}
|
||||
|
||||
fn invalid_input_response(message: &str) -> Response {
|
||||
(
|
||||
StatusCode::BAD_REQUEST,
|
||||
Json(json!({ "code": "INVALID_INPUT", "message": message })),
|
||||
)
|
||||
.into_response()
|
||||
}
|
||||
|
||||
/// A newline-framed reader over the request body's byte stream. Blank
|
||||
/// lines are skipped; a trailing unterminated line is yielded as a
|
||||
/// final line. Lines are capped at `MAX_PUBLISH_LINE_BYTES` (in
|
||||
/// addition to axum's own 2 MiB default body limit); a byte-level read
|
||||
/// failure surfaces as a terminal error.
|
||||
struct BufferedLines {
|
||||
bytes: ByteStream,
|
||||
buffer: Vec<u8>,
|
||||
pending: VecDeque<Vec<u8>>,
|
||||
done: bool,
|
||||
}
|
||||
|
||||
impl BufferedLines {
|
||||
fn new(bytes: ByteStream) -> Self {
|
||||
Self {
|
||||
bytes,
|
||||
buffer: Vec::new(),
|
||||
pending: VecDeque::new(),
|
||||
done: false,
|
||||
}
|
||||
}
|
||||
|
||||
fn invalid_operation_type_response(operation: &str) -> Response {
|
||||
let error = CallError::invalid_operation_type(format!(
|
||||
"operation is not a Pub op; /publish requires OperationType::Pub: {operation}"
|
||||
));
|
||||
(
|
||||
StatusCode::BAD_REQUEST,
|
||||
Json(serde_json::to_value(&error).unwrap_or(Value::Null)),
|
||||
async fn next_line(&mut self) -> Result<Option<Vec<u8>>, LineError> {
|
||||
loop {
|
||||
if let Some(line) = self.pending.pop_front() {
|
||||
if line.iter().all(|b| b.is_ascii_whitespace()) {
|
||||
continue;
|
||||
}
|
||||
return Ok(Some(line));
|
||||
}
|
||||
if self.done {
|
||||
if self.buffer.iter().all(|b| b.is_ascii_whitespace()) {
|
||||
self.buffer.clear();
|
||||
return Ok(None);
|
||||
}
|
||||
return Ok(Some(std::mem::take(&mut self.buffer)));
|
||||
}
|
||||
match self.bytes.next().await {
|
||||
Some(Ok(bytes)) => self.buffer.extend_from_slice(&bytes),
|
||||
Some(Err(_)) => {
|
||||
self.done = true;
|
||||
return Err(LineError::Read);
|
||||
}
|
||||
None => {
|
||||
self.done = true;
|
||||
continue;
|
||||
}
|
||||
}
|
||||
while let Some(pos) = self.buffer.iter().position(|b| *b == b'\n') {
|
||||
let line: Vec<u8> = self.buffer.drain(..=pos).collect();
|
||||
let line = &line[..line.len() - 1];
|
||||
if line.len() > MAX_PUBLISH_LINE_BYTES {
|
||||
self.done = true;
|
||||
self.buffer.clear();
|
||||
self.pending.clear();
|
||||
return Err(LineError::LineCap);
|
||||
}
|
||||
self.pending.push_back(line.to_vec());
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
enum LineError {
|
||||
Read,
|
||||
LineCap,
|
||||
}
|
||||
|
||||
impl LineError {
|
||||
fn message(&self) -> &'static str {
|
||||
match self {
|
||||
Self::Read => "publish body read failed",
|
||||
Self::LineCap => "publish line exceeds the per-line cap",
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// The initiator-side chunk stream fed to `invoke_sink` (GW-01, GW-06):
|
||||
/// NDJSON lines parsed lazily straight from the request body, each
|
||||
/// 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.
|
||||
struct NdjsonChunkStream {
|
||||
lines: std::pin::Pin<Box<BufferedLines>>,
|
||||
validator: Option<jsonschema::Validator>,
|
||||
pending_first: Option<Value>,
|
||||
}
|
||||
|
||||
impl NdjsonChunkStream {
|
||||
fn new(
|
||||
lines: BufferedLines,
|
||||
validator: Option<jsonschema::Validator>,
|
||||
pending_first: Option<Value>,
|
||||
) -> Self {
|
||||
Self {
|
||||
lines: Box::pin(lines),
|
||||
validator,
|
||||
pending_first,
|
||||
}
|
||||
}
|
||||
|
||||
fn validate_chunk(
|
||||
&mut self,
|
||||
value: Value,
|
||||
) -> std::task::Poll<Option<Result<Value, CallError>>> {
|
||||
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",
|
||||
)
|
||||
.into_response()
|
||||
.with_details(json!({ "chunk": value })))));
|
||||
}
|
||||
}
|
||||
std::task::Poll::Ready(Some(Ok(value)))
|
||||
}
|
||||
}
|
||||
|
||||
impl futures::Stream for NdjsonChunkStream {
|
||||
type Item = Result<Value, CallError>;
|
||||
|
||||
fn poll_next(
|
||||
mut self: std::pin::Pin<&mut Self>,
|
||||
cx: &mut std::task::Context<'_>,
|
||||
) -> std::task::Poll<Option<Self::Item>> {
|
||||
if self.pending_first.is_some() {
|
||||
let first = self.pending_first.take().unwrap_or(Value::Null);
|
||||
return self.validate_chunk(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()))))
|
||||
}
|
||||
std::task::Poll::Pending => return std::task::Poll::Pending,
|
||||
}
|
||||
};
|
||||
let value = match serde_json::from_slice::<Value>(&line) {
|
||||
Ok(v) => v,
|
||||
Err(e) => {
|
||||
return std::task::Poll::Ready(Some(Err(CallError::invalid_input(format!(
|
||||
"publish line is not valid JSON: {e}"
|
||||
)))))
|
||||
}
|
||||
};
|
||||
self.validate_chunk(value)
|
||||
}
|
||||
}
|
||||
|
||||
fn subscribe_stream_from_envelope_stream(
|
||||
@@ -358,7 +477,7 @@ fn envelope_to_error_json(request_id: &str, error: &CallError) -> Value {
|
||||
fn not_found_envelope_json(operation: &str) -> Value {
|
||||
let error = CallError::not_found(operation);
|
||||
json!({
|
||||
"request_id": Value::Null,
|
||||
"request_id": uuid::Uuid::new_v4().to_string(),
|
||||
"result": "error",
|
||||
"error": serde_json::to_value(&error).unwrap_or(Value::Null),
|
||||
})
|
||||
@@ -967,6 +1086,40 @@ mod tests {
|
||||
results[1].get("error").and_then(|e| e.get("code")),
|
||||
Some(&json!("NOT_FOUND"))
|
||||
);
|
||||
assert!(
|
||||
results[1]
|
||||
.get("request_id")
|
||||
.map(|id| !id.is_null())
|
||||
.unwrap_or(false),
|
||||
"internal-op entries must carry a generated request_id, not null"
|
||||
);
|
||||
}
|
||||
#[tokio::test]
|
||||
async fn batch_exceeding_operation_cap_returns_400_invalid_input() {
|
||||
let router = build_router(registry_with_echo(), unused_provider());
|
||||
let requests: Vec<Value> = (0..MAX_BATCH_OPERATIONS + 1)
|
||||
.map(|i| json!({ "operation": "echo/run", "input": { "n": i } }))
|
||||
.collect();
|
||||
let req = json_request("POST", "/batch", json!(requests));
|
||||
let (status, body) = send(router, req).await;
|
||||
assert_eq!(status, StatusCode::BAD_REQUEST);
|
||||
assert_eq!(body.get("code"), Some(&json!("INVALID_INPUT")));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn batch_at_cap_dispatches_all_entries() {
|
||||
let router = build_router(registry_with_echo(), unused_provider());
|
||||
let requests: Vec<Value> = (0..MAX_BATCH_OPERATIONS)
|
||||
.map(|i| json!({ "operation": "echo/run", "input": { "n": i } }))
|
||||
.collect();
|
||||
let req = json_request("POST", "/batch", json!(requests));
|
||||
let (status, body) = send(router, req).await;
|
||||
assert_eq!(status, StatusCode::OK);
|
||||
let results = body
|
||||
.get("results")
|
||||
.and_then(|r| r.as_array())
|
||||
.expect("results array");
|
||||
assert_eq!(results.len(), MAX_BATCH_OPERATIONS);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
@@ -1316,12 +1469,7 @@ mod tests {
|
||||
while let Some(chunk) = chunks.next().await {
|
||||
match chunk {
|
||||
Ok(v) => collected.push(v),
|
||||
Err(e) => {
|
||||
return ResponseEnvelope::error(
|
||||
ctx.request_id,
|
||||
CallError::internal(format!("chunk error: {e:?}")),
|
||||
)
|
||||
}
|
||||
Err(e) => return ResponseEnvelope::error(ctx.request_id, e),
|
||||
}
|
||||
}
|
||||
ResponseEnvelope::ok(
|
||||
@@ -1349,6 +1497,46 @@ mod tests {
|
||||
Capabilities::new(),
|
||||
))
|
||||
.unwrap();
|
||||
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(
|
||||
|_unused_input, ctx, mut chunks| async move {
|
||||
let mut collected: Vec<Value> = Vec::new();
|
||||
use futures::StreamExt;
|
||||
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();
|
||||
registry
|
||||
.register(HandlerRegistration::new(
|
||||
OperationSpec::new(
|
||||
@@ -1474,12 +1662,33 @@ mod tests {
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn publish_non_pub_op_returns_400_invalid_operation_type() {
|
||||
async fn publish_non_pub_op_unauthenticated_maps_invalid_operation_type_to_401() {
|
||||
let router = build_router(publish_registry(), unused_provider());
|
||||
let body = ndjson(&[json!({ "operation": "echo/run", "chunk": {} })]);
|
||||
let req = raw_request("POST", "/publish", body);
|
||||
let (status, body) = send(router, req).await;
|
||||
assert_eq!(status, StatusCode::BAD_REQUEST);
|
||||
assert_eq!(status, StatusCode::UNAUTHORIZED);
|
||||
assert_eq!(body.get("code"), Some(&json!("INVALID_OPERATION_TYPE")));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn publish_non_pub_op_returns_422_invalid_operation_type() {
|
||||
let provider: Arc<dyn IdentityProvider> = Arc::new(
|
||||
StaticIdentityProvider::new()
|
||||
.with_token("user-tok", identity_with_scopes("user", &["user"])),
|
||||
);
|
||||
let router = build_router(publish_registry(), provider);
|
||||
let body = ndjson(&[json!({ "operation": "echo/run", "chunk": {} })]);
|
||||
let (k, v) = auth_header("user-tok");
|
||||
let req = Request::builder()
|
||||
.method("POST")
|
||||
.uri("/publish")
|
||||
.header("content-type", "application/x-ndjson")
|
||||
.header(k, v)
|
||||
.body(Body::from(body))
|
||||
.unwrap();
|
||||
let (status, body) = send(router, req).await;
|
||||
assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY);
|
||||
assert_eq!(body.get("code"), Some(&json!("INVALID_OPERATION_TYPE")));
|
||||
}
|
||||
|
||||
@@ -1512,6 +1721,16 @@ mod tests {
|
||||
assert_eq!(body.get("code"), Some(&json!("INVALID_INPUT")));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn publish_first_line_missing_chunk_returns_400_invalid_input() {
|
||||
let router = build_router(publish_registry(), unused_provider());
|
||||
let body = ndjson(&[json!({ "operation": "ingest/push" })]);
|
||||
let req = raw_request("POST", "/publish", body);
|
||||
let (status, body) = send(router, req).await;
|
||||
assert_eq!(status, StatusCode::BAD_REQUEST);
|
||||
assert_eq!(body.get("code"), Some(&json!("INVALID_INPUT")));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn publish_invalid_later_line_yields_handler_chunk_error() {
|
||||
let router = build_router(publish_registry(), unused_provider());
|
||||
@@ -1519,11 +1738,11 @@ mod tests {
|
||||
body.extend_from_slice(b"not-json\n");
|
||||
let req = raw_request("POST", "/publish", body);
|
||||
let (status, resp) = send(router, req).await;
|
||||
assert_eq!(status, StatusCode::INTERNAL_SERVER_ERROR);
|
||||
assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY);
|
||||
assert_eq!(
|
||||
resp.get("code"),
|
||||
Some(&json!("INTERNAL")),
|
||||
"the sink handler converts the chunk error to an INTERNAL envelope: {resp}"
|
||||
Some(&json!("INVALID_INPUT")),
|
||||
"the malformed chunk line terminates the stream as INVALID_INPUT: {resp}"
|
||||
);
|
||||
}
|
||||
|
||||
@@ -1570,14 +1789,135 @@ mod tests {
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn publish_body_is_fully_consumed_before_dispatch_not_required() {
|
||||
// Streaming body note: axum gives us Bytes (buffered). The
|
||||
// disconnect-cancels-sink property is structural: the request
|
||||
// future is dropped when the client disconnects, cancelling
|
||||
// invoke_sink and closing the PublishStream. Verified here by
|
||||
// the sink completing only after all chunks are consumed
|
||||
// (round-trip test) and by the wire test in the integration
|
||||
// suite (infra-integration-suite task covers the socket-level
|
||||
// early-disconnect case).
|
||||
async fn publish_schema_registered_op_rejects_invalid_chunk() {
|
||||
let router = build_router(publish_registry(), unused_provider());
|
||||
let body = ndjson(&[
|
||||
json!({ "operation": "ingest/typed", "chunk": { "n": 1 } }),
|
||||
json!({ "n": "not-an-integer" }),
|
||||
json!({ "n": 3 }),
|
||||
]);
|
||||
let req = raw_request("POST", "/publish", body);
|
||||
let (status, resp) = send(router, req).await;
|
||||
assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY);
|
||||
assert_eq!(
|
||||
resp.get("code"),
|
||||
Some(&json!("INVALID_INPUT")),
|
||||
"the chunk stream terminates with the schema violation: {resp}"
|
||||
);
|
||||
assert_eq!(
|
||||
resp.get("message"),
|
||||
Some(&json!("published chunk failed publish_schema validation"))
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn publish_schema_registered_op_accepts_valid_chunks() {
|
||||
let router = build_router(publish_registry(), unused_provider());
|
||||
let body = ndjson(&[
|
||||
json!({ "operation": "ingest/typed", "chunk": { "n": 1 } }),
|
||||
json!({ "n": 2 }),
|
||||
]);
|
||||
let req = raw_request("POST", "/publish", body);
|
||||
let (status, resp) = send(router, req).await;
|
||||
assert_eq!(status, StatusCode::OK);
|
||||
assert_eq!(resp.get("result"), Some(&json!("ok")));
|
||||
assert_eq!(resp["output"]["count"], 2);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn publish_op_without_schema_accepts_arbitrary_chunks() {
|
||||
let router = build_router(publish_registry(), unused_provider());
|
||||
let body = ndjson(&[
|
||||
json!({ "operation": "ingest/push", "chunk": { "anything": true } }),
|
||||
json!({ "n": null }),
|
||||
]);
|
||||
let req = raw_request("POST", "/publish", body);
|
||||
let (status, resp) = send(router, req).await;
|
||||
assert_eq!(status, StatusCode::OK);
|
||||
assert_eq!(resp["output"]["count"], 2);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn publish_first_chunk_validated_against_publish_schema() {
|
||||
let router = build_router(publish_registry(), unused_provider());
|
||||
let body = ndjson(&[json!({ "operation": "ingest/typed", "chunk": { "wrong": 1 } })]);
|
||||
let req = raw_request("POST", "/publish", body);
|
||||
let (status, resp) = send(router, req).await;
|
||||
assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY);
|
||||
assert_eq!(resp.get("code"), Some(&json!("INVALID_INPUT")));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn publish_line_exceeding_cap_yields_invalid_input_chunk_error() {
|
||||
let mut registry = OperationRegistry::new();
|
||||
registry
|
||||
.register(HandlerRegistration::new(
|
||||
OperationSpec::new(
|
||||
"ingest/big",
|
||||
OperationType::Pub,
|
||||
Visibility::External,
|
||||
json!({}),
|
||||
json!({}),
|
||||
vec![],
|
||||
AccessControl::default(),
|
||||
None,
|
||||
),
|
||||
HandlerKind::Sink(make_sink_handler(|_input, ctx, mut chunks| async move {
|
||||
use futures::StreamExt;
|
||||
let mut last_error = None;
|
||||
while let Some(chunk) = chunks.next().await {
|
||||
if let Err(e) = chunk {
|
||||
last_error = Some(e);
|
||||
break;
|
||||
}
|
||||
}
|
||||
let error = last_error.expect("an oversized line must produce an error item");
|
||||
ResponseEnvelope::error(ctx.request_id, error)
|
||||
})),
|
||||
OperationProvenance::Local,
|
||||
None,
|
||||
None,
|
||||
Capabilities::new(),
|
||||
))
|
||||
.unwrap();
|
||||
let router = build_router(Arc::new(registry), unused_provider());
|
||||
let mut body = ndjson(&[json!({ "operation": "ingest/big", "chunk": { "n": 1 } })]);
|
||||
let oversized = vec![b'a'; MAX_PUBLISH_LINE_BYTES + 1];
|
||||
body.extend_from_slice(&oversized);
|
||||
body.push(b'\n');
|
||||
let (status, resp) = send(router, raw_request("POST", "/publish", body)).await;
|
||||
assert_eq!(status, StatusCode::BAD_REQUEST);
|
||||
assert_eq!(resp.get("code"), Some(&json!("INVALID_INPUT")));
|
||||
assert!(
|
||||
resp.get("message")
|
||||
.and_then(|m| m.as_str())
|
||||
.map(|m| m.contains("publish line exceeds the per-line cap"))
|
||||
.unwrap_or(false),
|
||||
"expected the line-cap message, got: {resp}"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn publish_client_disconnect_before_dispatch_signals_error_item() {
|
||||
let router = build_router(publish_registry(), unused_provider());
|
||||
let first_line = serde_json::to_vec(&json!({
|
||||
"operation": "ingest/push",
|
||||
"chunk": { "n": 1 }
|
||||
}))
|
||||
.unwrap();
|
||||
let mut body = first_line.clone();
|
||||
body.push(b'\n');
|
||||
body.extend_from_slice(
|
||||
b"not-json
|
||||
",
|
||||
);
|
||||
let req = raw_request("POST", "/publish", body);
|
||||
let (status, resp) = send(router, req).await;
|
||||
assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY);
|
||||
assert_eq!(
|
||||
resp.get("code"),
|
||||
Some(&json!("INVALID_INPUT")),
|
||||
"the prematurely terminated body reaches the handler as an INVALID_INPUT Err item: {resp}"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
---
|
||||
id: review-001-gateway-publish-semantics
|
||||
name: /publish + /batch gateway fixes (GW-01, GW-06, GW-08..GW-11, HY-13)
|
||||
status: pending
|
||||
status: completed
|
||||
depends_on: []
|
||||
scope: narrow
|
||||
risk: medium
|
||||
@@ -50,12 +50,12 @@ routes:
|
||||
|
||||
## Acceptance Criteria
|
||||
|
||||
- [ ] `/publish` test with a `publish_schema`-registered Pub op rejects an invalid chunk (review's gate for this unit)
|
||||
- [ ] First line without `chunk` → `INVALID_INPUT`, not a null publish (test)
|
||||
- [ ] GW-06 decision landed: true streaming or ADR-068 amended; consistent tests + docs
|
||||
- [ ] Batch size capped (test); mixed-shape batch envelopes fixed
|
||||
- [ ] Redundant `/publish` pre-checks removed (dispatch still enforces)
|
||||
- [ ] `cargo test` and `cargo clippy --all-targets -- -D warnings` pass
|
||||
- [x] `/publish` test with a `publish_schema`-registered Pub op rejects an invalid chunk (review's gate for this unit)
|
||||
- [x] First line without `chunk` → `INVALID_INPUT`, not a null publish (test)
|
||||
- [x] GW-06 decision landed: true streaming (implemented — not the ADR amendment); consistent tests + docs
|
||||
- [x] Batch size capped (test); mixed-shape batch envelopes fixed
|
||||
- [x] Redundant `/publish` pre-checks removed (dispatch still enforces)
|
||||
- [x] `cargo test` and `cargo clippy --all-targets -- -D warnings` pass
|
||||
|
||||
## References
|
||||
|
||||
@@ -65,10 +65,80 @@ routes:
|
||||
|
||||
## Notes
|
||||
|
||||
> Agent fills during implementation. If the GW-01 fix goes into
|
||||
> alkcall's `invoke_sink` spine, coordinate the alkcall change (small,
|
||||
> additive) and note it in the summary.
|
||||
- **GW-01 — route-level fix (spine route explored, rejected):** the
|
||||
spine option would put validation inside alkcall's
|
||||
`OperationRegistry::invoke_sink`; but the wire dispatcher's
|
||||
per-chunk validation is intentionally *outside* `invoke_sink` (it
|
||||
lives in `Dispatcher::dispatch`'s `SinkDispatch`/`pump_sink`, which
|
||||
owns the abort channel and init-side error injection). Moving it
|
||||
into `invoke_sink` would have meant a wrapping-stream refactor of
|
||||
that spine plus an alkcall release (alkhttp's dep is the crates.io
|
||||
`0.1.1`). The route now compiles the op's `publish_schema`
|
||||
(jsonschema 0.46, same version alkcall uses) and feeds
|
||||
`NdjsonChunkStream` into `invoke_sink`: every chunk — including the
|
||||
first-line chunk — is validated exactly like the wire dispatcher's
|
||||
`EVENT_PUBLISHED` branch; a violation (or malformed JSON) yields the
|
||||
terminal `Err(INVALID_INPUT)` item the wire's initiator-side
|
||||
`call.error` produces. Both transports now enforce the identical
|
||||
per-chunk contract; no alkcall change was needed.
|
||||
- **GW-06 — true streaming implemented:** `publish_handler` now takes
|
||||
`axum::body::Body` → `into_data_stream()` → `BufferedLines` (newline
|
||||
framer) → `NdjsonChunkStream`; nothing buffers the whole body. The
|
||||
only bound is the per-line cap (2 MiB, mirroring axum's default body
|
||||
limit) plus axum's own whole-body default on transport reads. ADR-068
|
||||
gained a "Body handling (streamed, not buffered)" section + a
|
||||
consequence line instead of an amendment — the original step-4
|
||||
wording was already "stream each NDJSON line"; the docs now state the
|
||||
implemented mechanics.
|
||||
- **GW-08:** `MAX_BATCH_OPERATIONS = 100`; over-cap → 400
|
||||
`INVALID_INPUT` (test covers over-cap rejection and at-cap dispatch).
|
||||
- **GW-09:** `not_found_envelope_json` now generates a
|
||||
`uuid::Uuid::new_v4()` request id, so every entry in one `/batch`
|
||||
response body carries the same envelope shape.
|
||||
- **GW-11:** all four `/publish` pre-checks (internal-op probe,
|
||||
existence probe, ACL probe, op-type probe) removed; visibility/ACL/
|
||||
handler-kind/type enforcement rides on `invoke_sink` via the shared
|
||||
dispatch spine, exactly like `/call`//`/batch`.
|
||||
- **HY-13:** the vacuous stub was replaced by
|
||||
`publish_client_disconnect_before_dispatch_signals_error_item`
|
||||
(body cut short mid-stream → the handler's `PublishStream` observes
|
||||
the `Err` item) plus the malformed-later-line test now asserts the
|
||||
real 422 `INVALID_INPUT` mapping.
|
||||
- **Adjacent (in-scope necessity):** `error.rs` gained the
|
||||
`INVALID_OPERATION_TYPE` mapping (422 with identity / 401 without) —
|
||||
the dropped pre-checks mean `/publish` now relies on the shared
|
||||
mapper for the non-Pub-op case, and the old fall-through mapped it to
|
||||
500 (GW-03's finding; the route previously special-cased it to 400).
|
||||
Coordinates with review-001-gateway-stream-errors (GW-03) — that task
|
||||
should verify/doc the mapping rather than re-implement it.
|
||||
|
||||
## Summary
|
||||
|
||||
> Filled on completion.
|
||||
Implemented all seven findings in `src/gateway/routes.rs` (+ the
|
||||
`error.rs` mapping addition + an ADR-068 doc note):
|
||||
|
||||
- **GW-01:** `/publish` validates every NDJSON chunk against the op's
|
||||
compiled `publish_schema` (incl. the first-line chunk) before it
|
||||
reaches the sink — a terminal `Err(INVALID_INPUT)`/422 on violation,
|
||||
message + details matching the wire dispatcher. No alkcall spine
|
||||
change (rejected: the wire validation is pump-side by design; a
|
||||
route-level fix achieves both-transport enforcement without touching
|
||||
alkcall's spine or taking a new release dependency).
|
||||
- **GW-06:** true streaming — `Body::into_data_stream()` →
|
||||
newline-framed `BufferedLines` → lazily parsed, schema-validated
|
||||
`NdjsonChunkStream` → `invoke_sink`. ADR-068 documents the streamed
|
||||
semantics + 2 MiB per-line cap.
|
||||
- **GW-08:** `/batch` capped at 100 operations (`INVALID_INPUT` 400).
|
||||
- **GW-09:** internal-op batch entries get generated UUID request ids
|
||||
(single envelope shape per response).
|
||||
- **GW-10:** first publish line missing `chunk` → `INVALID_INPUT` 400.
|
||||
- **GW-11:** `/publish` pre-checks removed; enforcement via `invoke_sink`.
|
||||
- **HY-13:** vacuous test deleted; replaced by an early-terminate body
|
||||
test (client-cut body → handler sees the `Err` item, 422).
|
||||
|
||||
Tests: 24 gateway-route tests added/updated (schema reject/accept,
|
||||
first-chunk validation, no-schema passthrough, first-line-missing-chunk,
|
||||
oversized-line cap, cap+disconnect semantics, batch cap, batch
|
||||
request-id shape, `INVALID_OPERATION_TYPE` 401/422 matrix + `error.rs`
|
||||
mapping tests). `cargo test` 199 passed; `cargo clippy --all-targets --
|
||||
-D warnings` clean; `cargo fmt --check` clean.
|
||||
Reference in New Issue
Block a user