Migrate OTLP/OTAP/Arrow receivers to axum

[strawgate]

Summary

• migrate otlp_receiver, otap_receiver, and arrow_ipc_receiver from tiny_http request loops to axum servers running on the existing runtime-thread pattern
• extend BackgroundHttpTask with an axum shutdown handle so drop still performs graceful shutdown and thread join
• preserve receiver semantics for path/method handling, body limits, decompression/decoding, backpressure/disconnect responses, and health transitions
• update OTLP health-thread test helper to use the new axum background-task constructor

Validation

• cargo fmt --check
• cargo check -p logfwd-io
• cargo test -p logfwd-io --lib
• targeted suites:
  • cargo test -p logfwd-io arrow_ipc_receiver::tests:: -- --nocapture
  • cargo test -p logfwd-io otap_receiver::tests:: -- --nocapture
  • cargo test -p logfwd-io otlp_receiver::tests:: -- --nocapture

Note

Migrate OTLP, OTAP, and Arrow IPC receivers from tiny_http to axum

Replaces the synchronous tiny_http server in three receivers with async axum routers running on a tokio current-thread runtime inside a dedicated worker thread.

• Each receiver (otlp_receiver, otap_receiver, arrow_ipc_receiver) now binds a std::net::TcpListener, converts it to a tokio::net::TcpListener, and serves an axum Router with graceful shutdown via a oneshot channel.
• BackgroundHttpTask gains a new_axum constructor and a ShutdownHandle enum to support both tiny_http (unblock) and axum (oneshot send) shutdown paths on Drop.
• Request handling is moved into async axum handlers (handle_otlp_request, handle_otap_request, handle_arrow_ipc_request), each enforcing Content-Length limits, body size caps, and backpressure via the existing bounded channel.
• Behavioral Change: 404/405 responses are now generated by axum routing instead of manual tiny_http logic; the 429 response body in the Arrow IPC receiver is shorter; OTAP success responses now explicitly set application/x-protobuf as the Content-Type.

^{Macroscope summarized 205f954.}

[chatgpt-codex-connector]

You have reached your Codex usage limits for code reviews. You can see your limits in the Codex usage dashboard.

[coderabbitai]

Caution

Review failed

Pull request was closed or merged during review

Walkthrough

Replaces synchronous tiny_http accept loops with async axum/Tokio servers for arrow_ipc, otap, and otlp receivers. Each receiver now binds a non-blocking TcpListener, runs axum route handlers (e.g., POST /v1/arrow, /v1/arrow_logs, /v1/logs) on a dedicated current-thread runtime, and uses per-server state structs carrying a bounded mpsc::SyncSender<RecordBatch>, shutdown signaling, and health storage. Request validation, framed body reading, optional decompression, decoding, and health updates were moved into handlers; channel backpressure/disconnect map to HTTP 429/503. BackgroundHttpTask now supports oneshot-based axum shutdown.

Possibly related PRs

• PR 1568: Touches arrow_ipc_receiver request handling and backpressure/send semantics, modifying the same queueing and HTTP response code paths.
• PR 1587: Changes the same receiver files to enforce/report backpressure and adjusts OTLP wire-bytes accounting/delivery semantics.
• PR 1538: Modifies otlp_receiver request handling, decompression/content-encoding logic, and introduces axum-based server/shutdown wiring overlapping this work.

---

Caution

Pre-merge checks failed

Please resolve all errors before merging. Addressing warnings is optional.

• Ignore

❌ Failed checks (3 errors, 2 warnings)

Check name	Status	Explanation	Resolution
High-Quality Rust Practices	❌ Error	otlp_receiver.rs uses raw .store(Healthy) without monotonic reducer, allowing concurrent races to clobber health state; arrow_ipc_receiver can return 429 after partial batch enqueues; missing in-flight concurrency control.	Replace raw .store() with monotonic reducer (store_health_event); reserve channel capacity before send loops; add semaphore to gate in-flight request processing across all receivers.
Formal Verification Coverage	❌ Error	PR adds new public function BackgroundHttpTask::new_axum() in logfwd-io/src/background_http_task.rs without required Kani formal verification proof.	Add Kani proof (verify_new_axum_init) for BackgroundHttpTask::new_axum() in #[cfg(kani)] block and update VERIFICATION.md with proof counts.
Crate Boundary And Dependency Integrity	❌ Error	New dependencies (axum, http-body-util, socket2, bytes, flate2, zstd) declared directly in logfwd-io/Cargo.toml instead of centralized in workspace.dependencies.	Add all six dependencies to root Cargo.toml [workspace.dependencies] with explicit versions, then update logfwd-io/Cargo.toml to use workspace = true references.
Documentation Thoroughly Updated	⚠️ Warning	PR lacks doc comments for new public new_axum function and contains undeclared dependency violations.	Add /// doc comment to BackgroundHttpTask::new_axum; update CRATE_RULES.md to include axum and transitive dependencies.
Maintainer Fitness	⚠️ Warning	PR contains four documented review concerns about serious concurrency issues that remain unresolved in the code, violating the maintainer fitness rule requiring documentation of risk surface and known limitations.	Resolve flagged concurrency issues using atomic CAS loops, batch capacity reservation, and semaphores for in-flight limits, OR add 'Known Limitations' section documenting races, severity, and follow-up plans with concurrent test cases.

---

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[macroscopeapp]

Approvability

Verdict: Needs human review

Major infrastructure refactor replacing the HTTP server framework (tiny_http → axum) across three receiver components. While the request handling logic appears preserved, changing the underlying HTTP serving implementation is a significant runtime behavior change that warrants human review.

^{You can customize Macroscope's approvability policy. Learn more.}

[Copilot]

Pull request overview

This PR migrates the OTLP, OTAP, and Arrow IPC HTTP receivers in logfwd-io from tiny_http request loops to axum servers, while preserving the existing background runtime-thread model and graceful shutdown behavior.

Changes:

• Replaced per-receiver tiny_http accept/read loops with axum routers + axum::serve(...) on a single-thread Tokio runtime.
• Extended BackgroundHttpTask to support graceful shutdown for both tiny_http and axum via a unified shutdown handle.
• Updated OTLP receiver tests/helpers to construct the new axum-backed background task.

Reviewed changes

Copilot reviewed 4 out of 4 changed files in this pull request and generated 5 comments.

File	Description
crates/logfwd-io/src/otlp_receiver.rs	Migrates OTLP HTTP receiver to axum, adds request handler helpers, updates a test helper to use new_axum.
crates/logfwd-io/src/otap_receiver.rs	Migrates OTAP receiver to axum with an async handler and bounded-body reading.
crates/logfwd-io/src/arrow_ipc_receiver.rs	Migrates Arrow IPC receiver to axum, keeping decompression + backpressure semantics in an async handler.
crates/logfwd-io/src/background_http_task.rs	Generalizes shutdown/join to support both tiny_http and axum shutdown signaling.

[coderabbitai]

Actionable comments posted: 2

🤖 Prompt for all review comments with AI agents

Verify each finding against the current code and only fix it if needed.

Inline comments:
In `@crates/logfwd-io/src/otap_receiver.rs`:
- Around line 317-338: Extract the duplicated helpers declared_content_length
and read_limited_body into a new shared module (e.g., receiver_http) and update
otap_receiver, otlp_receiver, and arrow_ipc_receiver to import them;
specifically, move the functions currently in otap_receiver.rs into
receiver_http::{declared_content_length, read_limited_body}, make them pub (keep
signatures the same and preserve error types like StatusCode and MAX_BODY_SIZE),
and replace local definitions in each receiver file with use
receiver_http::{declared_content_length, read_limited_body} so all three callers
reference the shared implementations.

In `@crates/logfwd-io/src/otlp_receiver.rs`:
- Around line 414-428: read_limited_body currently always starts with an empty
Vec causing repeated reallocations; change its signature (or add a wrapper) to
accept an optional content_length (u64/usize) and, if content_length is present,
validate it is <= MAX_BODY_SIZE (return StatusCode::PAYLOAD_TOO_LARGE if not)
and pre-allocate the output buffer with Vec::with_capacity(content_length as
usize); otherwise keep the current incremental behavior. Update callers to pass
the request's Content-Length when available and trusted; keep existing frame
iteration (Body.frame(), frame.into_data(), saturation checks) unchanged to
preserve safety.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Repository YAML (base), Organization UI (inherited)

Review profile: ASSERTIVE

Plan: Pro

Run ID: 8972db59-7834-47aa-90a4-2ab9b78e6f5e

📥 Commits

Reviewing files that changed from the base of the PR and between d538bd1 and d156838.

📒 Files selected for processing (4)

• crates/logfwd-io/src/arrow_ipc_receiver.rs
• crates/logfwd-io/src/background_http_task.rs
• crates/logfwd-io/src/otap_receiver.rs
• crates/logfwd-io/src/otlp_receiver.rs