feat(ingester): add the OTLP/HTTP listener (RFC0003.13/.14 + .11 HTTP arms)

crates/ourios-ingester/Cargo.toml

@@ -53,9 +53,20 @@ prost = { version = "0.14", default-features = false, features = ["std"] }
 # `Deserialize` impls — the `application/json` transport body is an
 # `ExportLogsServiceRequest` in the proto3-JSON mapping.
 serde_json = { version = "1", default-features = false, features = ["std"] }
+# HTTP transport (RFC 0003 §6.2): the OTLP/HTTP `/v1/logs` listener.
+# `tokio` + `http1` so the router can be served; the handler + extractors
+# (`State`, `Bytes`, `HeaderMap`) are core. On hyper 1.x, already in the
+# tree.
+axum = { version = "0.8", default-features = false, features = ["tokio", "http1"] }
+# gzip decode for OTLP/HTTP `Content-Encoding: gzip` (an OTLP MUST,
+# RFC0003.13). Pure-Rust backend — no zlib C dependency (CLAUDE.md).
+flate2 = { version = "1", default-features = false, features = ["rust_backend"] }
 
 [dev-dependencies]
 tempfile = "3"
+# Drive the axum `Router` in-process (`ServiceExt::oneshot`) for the
+# HTTP-transport acceptance tests — no real socket.
+tower = { version = "0.5", default-features = false, features = ["util"] }
 # Proptest strategy over the OTLP proto value space for the
 # wire-decode equivalence scenarios (RFC0003.5/.6 per §8).
 proptest = "1"

crates/ourios-ingester/src/lib.rs

@@ -11,9 +11,10 @@
 //!   (§6.2 wire decode — protobuf + OTLP/JSON, RFC0003.5/.6),
 //!   [`receiver::materialize`] (§6.1 `LogRecord` → `OtlpLogRecord`,
 //!   RFC0003.7–.10), [`receiver::tenant`] (per-`ResourceLogs` tenant
-//!   derivation + fan-out, RFC0003.3/.4), and [`receiver::pipeline`]
-//!   (§6.5 WAL-before-ack ingest path, RFC0003.1/.12). The live
-//!   gRPC/HTTP transports wrap the pipeline next.
+//!   derivation + fan-out, RFC0003.3/.4), [`receiver::pipeline`]
+//!   (§6.5 WAL-before-ack ingest path, RFC0003.1/.12), and
+//!   [`receiver::http`] (the OTLP/HTTP listener, RFC0003.11-HTTP/.13/.14).
+//!   The gRPC listener follows.
 //! - **WAL-before-ack** (RFC 0008 / `CLAUDE.md` §3.4) — durability
 //!   before acknowledgement, via the shipped `ourios-wal`. Wired into
 //!   the ingest path by [`receiver::pipeline`]: every non-empty batch is

crates/ourios-ingester/src/receiver.rs

@@ -19,10 +19,14 @@
 //!   (RFC0003.3/.4).
 //! - [`pipeline`] — the §6.5 WAL-before-ack ingest path
 //!   ([`pipeline::IngestPipeline`]): fan out → append one `OtlpBatch`
-//!   frame → fsync → miner → ack (RFC0003.1/.12). The live gRPC/HTTP
-//!   transports wrap this layer next.
+//!   frame → fsync → miner → ack (RFC0003.1/.12).
+//! - [`http`] — the OTLP/HTTP listener ([`http::router`]) wrapping the
+//!   pipeline: `Content-Type`/`Content-Encoding` dispatch, controlled
+//!   transport errors, configurable path (RFC0003.11 HTTP arms / .13 /
+//!   .14). The gRPC listener (`tonic`) follows.
 
 pub mod decode;
+pub mod http;
 pub mod materialize;
 pub mod pipeline;
 pub mod tenant;

crates/ourios-ingester/src/receiver/http.rs

@@ -0,0 +1,179 @@
+//! OTLP/HTTP listener (RFC 0003 §6.2).
+//!
+//! An `axum` `POST` handler at a configurable path (default `/v1/logs`)
+//! that decodes the body per `Content-Type` (`application/x-protobuf` or
+//! `application/json`) and `Content-Encoding` (`identity` or `gzip`),
+//! hands the decoded `ExportLogsServiceRequest` to the [`IngestPipeline`]
+//! (WAL-before-ack), and returns an `ExportLogsServiceResponse`.
+//!
+//! Transport errors are controlled (RFC0003.11): unsupported media type
+//! / encoding → 415, malformed body → 400, oversize → 413, an
+//! unconfigured path → 404, tenant-resolution failure → 400. No panics.
+//!
+//! The pipeline is shared behind a `Mutex` — the WAL is a single writer
+//! (RFC 0008 §3.1), so concurrent requests serialize on it. The lock is
+//! never held across an `.await`, so a plain `std::sync::Mutex` suffices.
+
+use std::sync::{Arc, Mutex};
+
+use axum::Router;
+use axum::body::Bytes;
+use axum::extract::{DefaultBodyLimit, State};
+use axum::http::{HeaderMap, StatusCode, header};
+use axum::response::{IntoResponse, Response};
+use axum::routing::post;
+use opentelemetry_proto::tonic::collector::logs::v1::ExportLogsServiceResponse;
+use prost::Message;
+
+use crate::receiver::decode::{decode_json, decode_protobuf};
+use crate::receiver::pipeline::{IngestPipeline, ReceiveError};
+
+/// The ingest pipeline shared across requests. The single-writer WAL
+/// forces serialization; concurrent requests queue on the mutex.
+pub type SharedPipeline = Arc<Mutex<IngestPipeline>>;
+
+/// OTLP/HTTP listener configuration.
+#[derive(Debug, Clone)]
+pub struct HttpConfig {
+    /// The accepted `POST` path (default `/v1/logs`; configurable per
+    /// §6.2 / RFC0003.14).
+    pub path: String,
+    /// Maximum request body size in bytes; a larger body is rejected with
+    /// 413 (RFC0003.11).
+    pub max_body_bytes: usize,
+}
+
+impl Default for HttpConfig {
+    fn default() -> Self {
+        Self {
+            path: "/v1/logs".to_owned(),
+            max_body_bytes: 4 * 1024 * 1024,
+        }
+    }
+}
+
+/// Build the OTLP/HTTP router over `pipeline`.
+pub fn router(pipeline: SharedPipeline, config: &HttpConfig) -> Router {
+    Router::new()
+        .route(&config.path, post(handle_logs))
+        .layer(DefaultBodyLimit::max(config.max_body_bytes))
+        .with_state(pipeline)
+}
+
+/// The OTLP wire format selected by `Content-Type`.
+#[derive(Clone, Copy)]
+enum WireFormat {
+    Protobuf,
+    Json,
+}
+
+/// The supported request `Content-Encoding`s.
+enum Encoding {
+    Identity,
+    Gzip,
+}
+
+async fn handle_logs(
+    State(pipeline): State<SharedPipeline>,
+    headers: HeaderMap,
+    body: Bytes,
+) -> Response {
+    let Some(format) = content_type(&headers) else {
+        return StatusCode::UNSUPPORTED_MEDIA_TYPE.into_response();
+    };
+    let raw = match content_encoding(&headers) {
+        Some(Encoding::Identity) => body.to_vec(),
+        Some(Encoding::Gzip) => match gunzip(&body) {
+            Ok(bytes) => bytes,
+            // A corrupt gzip stream is a malformed request, not an
+            // unsupported one.
+            Err(_) => return StatusCode::BAD_REQUEST.into_response(),
+        },
+        None => return StatusCode::UNSUPPORTED_MEDIA_TYPE.into_response(),
+    };
+    let decoded = match format {
+        WireFormat::Protobuf => decode_protobuf(&raw),
+        WireFormat::Json => decode_json(&raw),
+    };
+    let Ok(request) = decoded else {
+        return StatusCode::BAD_REQUEST.into_response();
+    };
+
+    // WAL-before-ack ingest. The lock spans only this synchronous call.
+    // Recover the guard even if a prior holder panicked: a poisoned lock
+    // must not turn into a panic here (the handler promises not to), so
+    // take the inner guard regardless and let this request proceed.
+    let outcome = pipeline
+        .lock()
+        .unwrap_or_else(std::sync::PoisonError::into_inner)
+        .ingest(request);
+
+    match outcome {
+        Ok(_) => success_response(format),
+        // A Resource that doesn't resolve to a tenant is a client error
+        // (the whole batch is rejected, RFC0003.4).
+        Err(ReceiveError::TenantResolution(_)) => StatusCode::BAD_REQUEST.into_response(),
+        // A WAL failure is server-side; the batch was not acked (§3.4).
+        Err(_) => StatusCode::INTERNAL_SERVER_ERROR.into_response(),
+    }
+}
+
+/// Map `Content-Type` to a wire format, ignoring any `; charset=…`
+/// parameters. `None` = missing or unsupported (→ 415).
+fn content_type(headers: &HeaderMap) -> Option<WireFormat> {
+    let value = headers.get(header::CONTENT_TYPE)?.to_str().ok()?;
+    let media_type = value.split(';').next().unwrap_or_default().trim();
+    match media_type {
+        "application/x-protobuf" => Some(WireFormat::Protobuf),
+        "application/json" => Some(WireFormat::Json),
+        _ => None,
+    }
+}
+
+/// Map `Content-Encoding` to a supported encoding. Absent or `identity`
+/// is identity; `gzip` is supported; anything else (`zstd`, `br`, …) is
+/// `None` (→ 415; zstd is deferred per §9).
+fn content_encoding(headers: &HeaderMap) -> Option<Encoding> {
+    match headers.get(header::CONTENT_ENCODING) {
+        None => Some(Encoding::Identity),
+        Some(value) => match value.to_str().ok()?.trim() {
+            "" | "identity" => Some(Encoding::Identity),
+            "gzip" => Some(Encoding::Gzip),
+            _ => None,
+        },
+    }
+}
+
+fn gunzip(bytes: &[u8]) -> std::io::Result<Vec<u8>> {
+    use std::io::Read;
+    let mut decoder = flate2::read::GzDecoder::new(bytes);
+    let mut out = Vec::new();
+    decoder.read_to_end(&mut out)?;
+    Ok(out)
+}
+
+/// A 200 carrying an empty `ExportLogsServiceResponse` (`partial_success`
+/// unset), encoded in the request's wire format.
+fn success_response(format: WireFormat) -> Response {
+    let response = ExportLogsServiceResponse::default();
+    match format {
+        WireFormat::Protobuf => (
+            StatusCode::OK,
+            [(header::CONTENT_TYPE, "application/x-protobuf")],
+            response.encode_to_vec(),
+        )
+            .into_response(),
+        WireFormat::Json => match serde_json::to_vec(&response) {
+            Ok(body) => (
+                StatusCode::OK,
+                [(header::CONTENT_TYPE, "application/json")],
+                body,
+            )
+                .into_response(),
+            // Encoding the (trivial) success response shouldn't fail; if
+            // it ever did, a 500 is honest — never a 200 with an empty
+            // body.
+            Err(_) => StatusCode::INTERNAL_SERVER_ERROR.into_response(),
+        },
+    }
+}

crates/ourios-ingester/src/receiver/pipeline.rs

@@ -24,7 +24,10 @@ use crate::receiver::tenant::{TenantResolutionError, TenantRule, fan_out};
 /// ordering can be exercised with a spy that records the `append`/`sync`
 /// calls (RFC0003.1/.12 per §8). The only production implementation is
 /// [`Wal`].
-pub trait Journal {
+///
+/// `Send` so the pipeline can live behind a shared `Arc<Mutex<_>>` as
+/// state in the async HTTP/gRPC listeners.
+pub trait Journal: Send {
     /// Append one `OtlpBatch` frame carrying `payload` (not yet durable).
     ///
     /// # Errors