feat(ingester): add the WAL-before-ack ingest pipeline (RFC0003.1/.12) - #134
Conversation
Fifth green slice of the OTLP receiver (RFC 0003 §6.5): the business-logic layer the live transports will wrap. Flips RFC0003.1/.12 live. `receiver::pipeline::IngestPipeline` owns the single-writer Wal, the per-process MinerCluster, and the TenantRule. `ingest` runs the §6.5 sequence: encode the export to a protobuf payload, fan out per tenant (an unresolvable Resource rejects the whole batch before any WAL write), append it as one FrameKind::OtlpBatch frame, fsync, then hand the records to the miner, then ack. The fsync completes before `ingest` returns Ok, so no batch is acked before it is durable (§3.4 / RFC0003.1). An empty batch takes the fast path: Ok with no WAL frame and no miner work (RFC0003.12). Hand-rolled `ReceiveError` (no thiserror): TenantResolution (whole-batch reject) + WalAppend/WalSync (not acked). Adds ourios-wal + ourios-miner as ingester dependencies. Tests: rfc0003_1 (ingest Ok ⇒ the OtlpBatch frame is durable on a fresh replay + recovers the export; the record reached the miner), rfc0003_12 (all three empty shapes → Ok(0), no frame, miner untouched), over a real Wal in a tempdir. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
|
Warning Review limit reached
More reviews will be available in 30 minutes and 37 seconds. Learn how PR review limits work. Your organization has run out of usage credits. Purchase more in the billing tab. ⌛ How to resolve this issue?After more reviews become available, a review can be triggered using the We recommend that you space out your commits to avoid hitting the rate limit. 🚦 How do rate limits work?CodeRabbit enforces hourly rate limits for each developer per organization. Our paid plans include higher PR review limits than trial, open-source, and free plans. In all cases, reviews become available again over time. During sustained high-volume PR review activity, CodeRabbit may temporarily slow when the next review becomes available. Please see our Fair Usage Limits Policy for further information. ℹ️ Review info⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (7)
📝 WalkthroughWalkthroughThis PR implements the WAL-before-ack ingest pipeline for ourios-ingester, enabling durable log ingestion with ordering guarantees. The implementation adds the ChangesIngest Pipeline with WAL Durability
Sequence DiagramsequenceDiagram
participant Caller
participant IngestPipeline
participant TenantRule
participant Wal
participant MinerCluster
Caller->>IngestPipeline: ingest(ExportLogsServiceRequest)
IngestPipeline->>IngestPipeline: encode request payload
IngestPipeline->>TenantRule: fan_out (derive tenant per ResourceLogs)
alt empty request
IngestPipeline->>Caller: Ok(0)
else non-empty request
IngestPipeline->>Wal: append(FrameKind::OtlpBatch, encoded)
IngestPipeline->>Wal: sync() for durability
Wal->>IngestPipeline: ack
IngestPipeline->>MinerCluster: ingest records per tenant
MinerCluster->>IngestPipeline: template counts updated
IngestPipeline->>Caller: Ok(record_count)
end
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~25 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
|
@coderabbitai review |
✅ Action performedReview finished.
|
There was a problem hiding this comment.
Pull request overview
Adds the ingester-side “business logic” ingest pipeline for RFC 0003 §6.5 (WAL-before-ack), wiring OTLP batch durability (append + fsync) ahead of miner ingestion/ack, and flips the previously-ignored acceptance tests for RFC0003.1 and RFC0003.12 to live integration tests.
Changes:
- Introduces
receiver::pipeline::IngestPipeline(WAL append + fsync before miner ingest / ack; empty-batch fast path). - Adds shared integration-test helpers (
tests/ingest_support) and enables RFC0003.1 / RFC0003.12 tests against a real WAL in a tempdir. - Wires new deps (
ourios-wal,ourios-miner) intoourios-ingester.
Reviewed changes
Copilot reviewed 7 out of 8 changed files in this pull request and generated 5 comments.
Show a summary per file
| File | Description |
|---|---|
| crates/ourios-ingester/src/receiver/pipeline.rs | New WAL-before-ack ingest pipeline + ReceiveError. |
| crates/ourios-ingester/src/receiver.rs | Exposes the new pipeline module and re-exports IngestPipeline/ReceiveError. |
| crates/ourios-ingester/src/lib.rs | Updates crate docs to include the new pipeline slice as landed. |
| crates/ourios-ingester/tests/ingest_support/mod.rs | Shared helpers for opening pipeline/WAL and building OTLP requests for integration tests. |
| crates/ourios-ingester/tests/rfc0003_1_wal_before_ack.rs | Enables and implements RFC0003.1 integration test over real WAL. |
| crates/ourios-ingester/tests/rfc0003_12_empty_request_success.rs | Enables and implements RFC0003.12 empty-shapes integration test. |
| crates/ourios-ingester/Cargo.toml | Adds ourios-wal and ourios-miner dependencies for the pipeline. |
| Cargo.lock | Locks new dependency edges for ourios-ingester. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| //! (`CLAUDE.md` §3.4 / RFC0003.1), and only then are the records handed | ||
| //! to the miner (RFC0003.5 step ordering). An empty batch takes the | ||
| //! fast path: success with no WAL write (RFC0003.12). |
There was a problem hiding this comment.
Fixed in 2f1edaf — the module doc now says "§6.5 step ordering" (RFC0003.5 is the decode-equivalence scenario; the durability/step-ordering contract is §6.5 / RFC0003.1).
| match self { | ||
| Self::TenantResolution(e) => write!(f, "tenant resolution failed: {e}"), | ||
| Self::WalAppend(e) => write!(f, "WAL append failed: {e:?}"), | ||
| Self::WalSync(e) => write!(f, "WAL sync failed: {e:?}"), |
There was a problem hiding this comment.
Fixed in 2f1edaf — ReceiveErrors tenant arm now delegates (write!(f, "{e}")); TenantResolutionErrors own Display already leads with "tenant resolution failed: …", so the double prefix is gone.
| assert_eq!( | ||
| recovered.resource_logs.len(), | ||
| 1, | ||
| "the durable frame recovers the acked export", | ||
| ); |
There was a problem hiding this comment.
Fixed in 2f1edaf — the test now decodes the recovered frame and asserts the record body is StringValue("user 1 logged in") verbatim, so a payload that lost/changed content fails.
| assert_eq!(ingested, 1); | ||
| assert!( | ||
| pipeline.miner().template_count(&TenantId::new("checkout")) >= 1, | ||
| "the record reached the miner", | ||
| ); |
There was a problem hiding this comment.
Fixed in 2f1edaf — tightened to template_count == 1 (one distinct line → exactly one template), so unexpected miner behavior is caught.
| pub enum ReceiveError { | ||
| /// A `ResourceLogs` group's Resource did not resolve to a tenant. | ||
| TenantResolution(TenantResolutionError), | ||
| /// Appending the `OtlpBatch` frame to the WAL failed. |
There was a problem hiding this comment.
Good catch — this is the RFC0003.4 §5 requirement ("names the failing ResourceLogs index and the missing attribute key"), which the error did not meet. Fixed in 2f1edaf: TenantResolutionError gains resource_index (None at the single-Resource derive, Some(i) once fan_out attaches the batch position), Display names ResourceLogs[i], and rfc0003_4 now asserts resource_index() == Some(1).
Review fixes for the pipeline PR: - RFC0003.4 §5 requires the tenant-resolution error to name the failing ResourceLogs *index* and the missing attribute; the error named only the attribute. Add `TenantResolutionError::resource_index` (None at the single-Resource `derive`, Some(i) once `fan_out` attaches the batch position), surface it in Display, and strengthen rfc0003_4 to assert `resource_index() == Some(1)`. - `ReceiveError`'s Display for the tenant arm double-prefixed "tenant resolution failed:" (the inner error already leads with it) — delegate. - pipeline module doc cited "RFC0003.5 step ordering"; the step ordering is §6.5 (RFC0003.5 is decode-equivalence). - Strengthen rfc0003_1: assert the recovered frame's body content verbatim (not just resource_logs.len), and template_count == 1 (one distinct line → one template) rather than >= 1. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
| //! (§6.5 WAL-before-ack ingest path, RFC0003.1/.12). The live | ||
| //! gRPC/HTTP transports wrap the pipeline next. | ||
| //! - **WAL-before-ack** (RFC 0008 / `CLAUDE.md` §3.4) — durability | ||
| //! before acknowledgement, via the shipped `ourios-wal`. Wired into | ||
| //! the ingest path once the receiver lands; not exercised here. |
There was a problem hiding this comment.
Fixed in the latest commit — the crate doc now says WAL-before-ack is wired into the ingest path by receiver::pipeline (append + fsync before ack, RFC0003.1), not "not exercised here".
…peline The WAL-before-ack bullet still said "not exercised here"; this slice wires ourios-wal into receiver::pipeline (append + fsync before ack). Surfaced in review. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
| impl std::fmt::Display for ReceiveError { | ||
| fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { | ||
| match self { | ||
| // `TenantResolutionError`'s own Display already leads with | ||
| // "tenant resolution failed: …"; delegate, don't re-prefix. | ||
| Self::TenantResolution(e) => write!(f, "{e}"), | ||
| Self::WalAppend(e) => write!(f, "WAL append failed: {e:?}"), | ||
| Self::WalSync(e) => write!(f, "WAL sync failed: {e:?}"), | ||
| } | ||
| } | ||
| } | ||
|
|
||
| impl std::error::Error for ReceiveError { | ||
| fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { | ||
| match self { | ||
| Self::TenantResolution(e) => Some(e), | ||
| // `ourios_wal::{AppendError, SyncError}` don't implement | ||
| // `std::error::Error` (RFC 0008 hand-rolled enums), so they | ||
| // can't be returned as a `source`; their detail is in the | ||
| // `Display` above. | ||
| Self::WalAppend(_) | Self::WalSync(_) => None, | ||
| } | ||
| } | ||
| } |
There was a problem hiding this comment.
Fixed in c8ca599 — gave ourios_wal::{AppendError, SyncError} proper Display (deterministic, rendering op + the io error) and std::error::Error (source → the inner io::Error). ReceiveError now delegates {e} and chains via source(), no more {:?}.
| // The ack returned only after the frame was fsync'd: reopening the | ||
| // WAL and replaying finds exactly one durable OtlpBatch frame whose | ||
| // payload recovers the export content (not just its shape). | ||
| drop(pipeline); | ||
| let frames = replay_frames(tmp.path()); | ||
| assert_eq!(frames.len(), 1, "exactly one OtlpBatch frame is durable"); | ||
| assert_eq!(frames[0].0, FrameKind::OtlpBatch); |
There was a problem hiding this comment.
Good point — added an explicit spy. rfc0003_1 now has a second test using a Journal spy that records call order and asserts ingest does [Append, Sync] before returning Ok, so the ack is gated on sync (a removed fsync would drop Sync and fail). The real-WAL replay test stays as the durability/recoverability check.
| // Assert: success with zero records, miner untouched, no frame. | ||
| assert_eq!(ingested, 0, "no records ingested"); | ||
| assert_eq!( | ||
| pipeline.miner().template_count(&TenantId::new("svc")), | ||
| 0, | ||
| "no record reached the miner", | ||
| ); | ||
| drop(pipeline); | ||
| assert!( | ||
| replay_frames(tmp.path()).is_empty(), | ||
| "no OtlpBatch frame was appended for an empty request", | ||
| ); |
There was a problem hiding this comment.
Fixed — rfc0003_12 now uses the Journal spy and asserts the call log is empty for every empty shape, i.e. an empty request neither appends nor syncs. That catches a regression that calls sync without appending (which the prior no-frame check would have missed).
Addresses review on the pipeline:
- Introduce a `Journal` trait (append_batch + sync) the pipeline appends/
fsyncs through, with the production `Wal` as the only real impl. The
pipeline now holds a `Box<dyn Journal>`, so tests can drive it with a
spy that records the call order. rfc0003_1 gains an assertion that
ingest appends then fsyncs *before* acking (the ack is gated on sync,
not merely on the frame being visible — §8 probe). rfc0003_12 now
asserts an empty request neither appends nor syncs (stronger than
"no frame": catches a regression that syncs without appending).
- Give ourios-wal `AppendError`/`SyncError` proper `Display` +
`std::error::Error` (source → the inner io::Error) so `ReceiveError`
renders them deterministically and chains, instead of `{:?}`.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
What
Fifth green slice of the OTLP receiver (RFC 0003 §6.5): the WAL-before-ack ingest pipeline — the business-logic layer the live transports will wrap — flipping RFC0003.1/.12 live.
Design
receiver::pipeline::IngestPipelineowns the single-writerWal, the per-processMinerCluster, and theTenantRule.ingestruns the §6.5 sequence:Resourcerejects the whole batch before any WAL write (RFC0003.4);FrameKind::OtlpBatchframe;fsync— completes beforeingestreturnsOk, so no batch is acked before it's durable ([§3.4]/ RFC0003.1);An empty batch takes the fast path:
Ok(0)with no WAL frame and no miner work (RFC0003.12).Hand-rolled
ReceiveError(nothiserror):TenantResolution(whole-batch reject) +WalAppend/WalSync(not acked). Addsourios-wal+ourios-mineras ingester deps.Scenarios (now live)
ingestreturnsOk⇒ a fresh WAL replay finds exactly one durableOtlpBatchframe whose payload decodes back to the export; the record reached the miner.resource_logs; a Resource with noscope_logs; aScopeLogswith nolog_records) →Ok(0), no frame appended, miner untouched.Both over a real
Walin a tempdir.Verification
cargo test -p ourios-ingester✓ — RFC0003.1/.3–.10/.12 live; 5 remain ignored (.2/.11/.13/.14/.15 — the live-transport + crash group).cargo fmt --all --check✓ ·cargo clippy --all-targets --all-features -- -D warnings✓ (workspace)Next
The live
axumHTTP +tonicgRPC listeners wrap this pipeline (.11/.13/.14 + .15), and the SIGKILL crash-before-ack harness (.2) exercises it — the remaining slices.🤖 Generated with Claude Code
Summary by CodeRabbit
Release Notes
New Features
Tests
Chores