fix(server): persist miner template audit events from the receiver (#302) - #312
Merged
Conversation
|
Warning Review limit reached
Next review available in: 12 minutes Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available. How can I continue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews. How do review limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window. Please refer docs for additional details. Review details⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (8)
📝 WalkthroughWalkthroughAdds buffered audit-event persistence with flush metrics, coordinates audit-before-record publication, wires the audit sink into receiver snapshot and shutdown flows, and updates tests to verify faithful body reconstruction from persisted template audit data. ChangesAudit Sink Persistence for Receiver Miner
Sequence Diagram(s)sequenceDiagram
participant MinerCluster
participant SharedParquetAuditSink
participant PublishCoordinator
participant SharedParquetSink
participant Receiver
participant Store
MinerCluster->>SharedParquetAuditSink: emit(template AuditEvent)
Receiver->>SharedParquetAuditSink: flush / drain on snapshot cadence
Receiver->>PublishCoordinator: drain_aged / drain_all
PublishCoordinator->>SharedParquetAuditSink: write audit batch first
SharedParquetAuditSink->>Store: put audit Parquet
PublishCoordinator->>SharedParquetSink: publish record batch after audit succeeds
SharedParquetSink->>Store: put record Parquet
Receiver->>Store: write snapshot only when both sinks drain
Estimated code review effort🎯 5 (Critical) | ⏱️ ~120 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ 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 |
Contributor
There was a problem hiding this comment.
Pull request overview
This PR fixes issue #302 where the receiver failed to persist the miner’s template audit events, leaving the querier’s read-time template registry empty and causing clean rows to render with an empty body. It introduces a buffering/shared audit sink wired into the receiver (and flushed off the async runtime) so template events reach the RFC 0005 audit Parquet stream, enabling faithful body reconstruction.
Changes:
- Add a buffering, shared audit sink (
BufferingAuditSink+SharedParquetAuditSink) and wire it into the receiver before recovery, with flush+snapshot gating aligned to the record sink. - Ensure flush ordering and snapshot gating cover both record and audit sinks to maintain the “no-loss / WAL-before-ack” invariant for template events.
- Extend/adjust tests to assert faithful body reconstruction (local + S3) and scope the WAL-local test’s Parquet readback to
data/now thataudit/coexists.
Reviewed changes
Copilot reviewed 7 out of 7 changed files in this pull request and generated no comments.
Show a summary per file
| File | Description |
|---|---|
| crates/ourios-ingester/src/audit_sink.rs | New buffering/shared audit sink that batches audit events per partition and flushes via AuditWriter off the async runtime. |
| crates/ourios-ingester/src/lib.rs | Exposes the new audit_sink module. |
| crates/ourios-server/src/receiver.rs | Wires the shared audit sink into MinerCluster pre-recovery; flushes audit before records; gates snapshot on both sinks draining; extends receiver tests with a regression for clean-row reconstruction. |
| crates/ourios-parquet/src/audit_writer.rs | Makes derive_audit_partition public and documents its canonicalization (hour fixed to 0) to support batching/grouping. |
| crates/ourios-parquet/src/lib.rs | Re-exports derive_audit_partition for cross-crate use. |
| crates/ourios-server/tests/rfc0019_storage_backend.rs | Adds assertions that S3-backed queries reconstruct clean-row body text faithfully (using template registry derived from persisted audit events). |
| crates/ourios-server/tests/rfc0013_6_wal_stays_local.rs | Scopes Parquet data readback to data/ so audit/ Parquet files aren’t read with the data reader. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
|
Caution Failed to replace (edit) comment. This is likely due to insufficient permissions or the comment being deleted. Error details |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@crates/ourios-ingester/src/audit_sink.rs`:
- Around line 48-55: The new BufferingAuditSink currently keeps flush and error
counters private, so its backlog and failure behavior are not visible in
production. Add exported Prometheus metrics for this sink, mirroring the record
sink’s metrics surface, and wire them through the BufferingAuditSink state and
its flush/derive paths so updates are emitted on the hot path. Use the existing
ingester metrics pattern and ensure the new counters are exposed rather than
only stored in the struct.
- Around line 108-125: The retry path in audit_sink::write_partition re-buffers
every failed batch, which causes permanent retry loops for non-transient
AuditWriterError::Batch failures from AuditWriter::append_events. Update the
flush logic so only transient/open/IO-style errors are requeued, while
invalid-content batch errors are dropped or counted as permanently failed
instead of extending self.buffer. Use write_partition and the surrounding
group-flush loop to distinguish error kinds before calling
self.buffer.extend(batch).
In `@crates/ourios-server/src/receiver.rs`:
- Around line 90-93: The flush path in `receiver.rs` still calls
`sink.flush_aged()` even when `audit_sink.flush()` has not drained retained
events. Update the `spawn_blocking` flush sequence so `sink.flush_aged()` only
runs after a successful `audit_sink.flush()` (and skip it on the failure path)
in both affected flush sites, using the existing `audit_sink`/`sink` logic to
keep clean rows hidden until template events are durable.
🪄 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: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: 50a6c88e-72a0-44a7-9593-e0ecc2217b4c
📒 Files selected for processing (7)
crates/ourios-ingester/src/audit_sink.rscrates/ourios-ingester/src/lib.rscrates/ourios-parquet/src/audit_writer.rscrates/ourios-parquet/src/lib.rscrates/ourios-server/src/receiver.rscrates/ourios-server/tests/rfc0013_6_wal_stays_local.rscrates/ourios-server/tests/rfc0019_storage_backend.rs
jensholdgaard
added a commit
that referenced
this pull request
Jun 29, 2026
…#302) Address PR #312 review feedback on the receiver audit sink. OTel metrics (§6.3): add `ourios.audit_sink.*` instruments mirroring the record sink's `SinkMetrics` — `buffer.usage` (observable gauge of buffered events), `flushes`, `flush.events`, `flush.errors` (split transient vs permanent via the new `ourios.audit_sink.flush.outcome` attribute), and `derive.errors`. Names go through the weaver registry (`semconv/registry/{metrics,attributes}.yaml`); the generated `ourios-semconv` constants are regenerated, not hand-written. Instruments resolve through the global meter (no-op without a provider). A dedicated test binary asserts the stream exports (separate process — `init_in_memory` installs the global provider). Data integrity: classify a failed partition flush. A store-`Io` error is transient → retain + retry (the WAL is the durability of record); a `Batch` / `Parquet` / `PartitionMismatch` / `Poisoned` error is permanent → drop + count, so one malformed event can't requeue forever and wedge every newer good event for that tenant/day behind it. §3.3 flush gating: in both the age-sweep and `flush_then_snapshot`, flush the audit sink first and skip the record flush this cycle if it didn't fully drain — a non-empty buffer means a transient store error (permanents drop), so the record flush to the same store would fail anyway, and flushing it would expose a clean row before its template event is durable. Bounded buffer: `emit` stays non-blocking but enforces a soft event ceiling (default 100k); reaching it signals a `tokio::sync::Notify` the age-sweep selects on, so adversarial template churn flushes promptly off the runtime rather than growing the buffer until OOM. Signal-to-flush, never drop. Tests: poison-pill (permanent drops + counts, does not requeue; transient retains), flush-gating (record flush skipped while audit retains), bounding (emit past the ceiling fires the notify), plus the metrics-export test. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
There was a problem hiding this comment.
Actionable comments posted: 3
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@crates/ourios-ingester/src/audit_sink.rs`:
- Around line 148-155: The buffer in buffer_event only triggers overflow_notify
after reaching ceiling_events, but it still keeps growing unbounded, so enforce
an actual cap in AuditBuffer rather than just coalescing wakeups. Update the
buffer_event flow (and any related flush/ingest path in AuditSink or
AuditBuffer) so once the ceiling is reached it either applies hard backpressure,
drops/spills events in a bounded way, or otherwise prevents self.buffer from
exceeding the configured limit; keep metrics.set_buffered in sync with the
enforced bound.
- Around line 33-36: The inline record flush path in the record sink can publish
query-visible rows before the audit stream is durable, leaving bodies
temporarily empty until the next cadence. Update the `emit`-driven size-trigger
path in `audit_sink.rs` to go through the same audit-first drain gate as the
normal audit flush, or otherwise coordinate the audit and record flush so
records are only published after audit durability is established.
- Around line 309-311: `SharedParquetAuditSink::emit` is currently blocked
behind the same mutex held during `flush`, which can stall request-path event
buffering while `AuditWriter` does I/O. Update the buffering flow so pending
events are drained or swapped out under the lock, then release the lock before
performing store writes, and finally re-lock only to update counters and requeue
any transient failures ahead of newly buffered events. Keep the fix localized
around `emit`, `flush`, and the shared sink lock/buffer handling.
🪄 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: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: d774bb3a-15c1-4fac-9751-131f019fa001
📒 Files selected for processing (9)
crates/ourios-ingester/Cargo.tomlcrates/ourios-ingester/src/audit_sink.rscrates/ourios-ingester/src/metrics.rscrates/ourios-ingester/tests/audit_sink_metrics.rscrates/ourios-parquet/src/audit_writer.rscrates/ourios-semconv/src/lib.rscrates/ourios-server/src/receiver.rssemconv/registry/attributes.yamlsemconv/registry/metrics.yaml
✅ Files skipped from review due to trivial changes (1)
- crates/ourios-semconv/src/lib.rs
🚧 Files skipped from review as they are similar to previous changes (2)
- crates/ourios-parquet/src/audit_writer.rs
- crates/ourios-server/src/receiver.rs
The OTLP receiver wired the template miner with a record sink but no audit sink, so the miner's `template_created` / `template_widened` / `template_type_expanded` events never reached the RFC 0005 audit Parquet stream. The querier's read-time registry (RFC 0017 `derive_template_registry`) was therefore empty and `render_log_body` fell back to the row's retained `body` — empty for clean, high-confidence rows — so queries over freshly-ingested clean logs returned empty body text, breaking `CLAUDE.md` §3.3. Mirror the RFC 0014 record sink rather than wiring `ParquetAuditSink` directly (which does a blocking per-event store write — request-path stall + one tiny file per event, hazard #4): - `BufferingAuditSink` / `SharedParquetAuditSink` (ourios-ingester): `emit` buffers cheaply on the request path; `flush` drains the buffer, groups events by audit partition, and writes each partition's batch with one `AuditWriter` (open_in → append_events → close) — few files, not one-per-event. A failed partition write retains its events (the WAL is the durability of record); an empty-buffer flush is a no-op. - The receiver constructs the sink on the same `Store`, wires it via `MinerCluster::with_audit_sink(...).with_record_sink(...)` before recovery (so replay re-emits template events), and flushes it off the async runtime at the same cadence + rotation + shutdown points as the record sink — audit *before* records (durable no later than the rows it describes), with the snapshot gated on both sinks draining. - Expose `derive_audit_partition` from ourios-parquet for the grouping. Tests: unit tests for the buffering sink (per-partition batching round trip + empty-buffer no-op); an in-process receiver test that ingests clean logs, drains, derives the registry, and asserts every clean row reconstructs `Faithful` from its template rather than empty. The RFC0019 `.3`/`.5` localstack scenarios now also assert the returned body text. `rfc0013_6` scopes its data round-trip to `data/` so the new `audit/` files aren't read with the data schema. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
derive_audit_partition is now pub; its doc referenced the private audit_partition_matches via an intra-doc link, which trips rustdoc::private-intra-doc-links under cargo doc -D warnings. Use plain backticks (the recurring private-item-doc convention). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
…#302) Address PR #312 review feedback on the receiver audit sink. OTel metrics (§6.3): add `ourios.audit_sink.*` instruments mirroring the record sink's `SinkMetrics` — `buffer.usage` (observable gauge of buffered events), `flushes`, `flush.events`, `flush.errors` (split transient vs permanent via the new `ourios.audit_sink.flush.outcome` attribute), and `derive.errors`. Names go through the weaver registry (`semconv/registry/{metrics,attributes}.yaml`); the generated `ourios-semconv` constants are regenerated, not hand-written. Instruments resolve through the global meter (no-op without a provider). A dedicated test binary asserts the stream exports (separate process — `init_in_memory` installs the global provider). Data integrity: classify a failed partition flush. A store-`Io` error is transient → retain + retry (the WAL is the durability of record); a `Batch` / `Parquet` / `PartitionMismatch` / `Poisoned` error is permanent → drop + count, so one malformed event can't requeue forever and wedge every newer good event for that tenant/day behind it. §3.3 flush gating: in both the age-sweep and `flush_then_snapshot`, flush the audit sink first and skip the record flush this cycle if it didn't fully drain — a non-empty buffer means a transient store error (permanents drop), so the record flush to the same store would fail anyway, and flushing it would expose a clean row before its template event is durable. Bounded buffer: `emit` stays non-blocking but enforces a soft event ceiling (default 100k); reaching it signals a `tokio::sync::Notify` the age-sweep selects on, so adversarial template churn flushes promptly off the runtime rather than growing the buffer until OOM. Signal-to-flush, never drop. Tests: poison-pill (permanent drops + counts, does not requeue; transient retains), flush-gating (record flush skipped while audit retains), bounding (emit past the ceiling fires the notify), plus the metrics-export test. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
…302) Round 3 on #312 — close the empty-body window fully: a mined record must never become query-visible before its template's audit event is durable, under concurrency and the inline size trigger (strengthens `CLAUDE.md` §3.3 to hold on every publication path). #4 — `emit` no longer blocks behind flush I/O. The audit sink's flush now drains the buffer under the lock, releases it, does the `AuditWriter` store I/O **unlocked**, then re-locks only to settle counters and requeue a transient failure's events ahead of anything `emit` buffered meanwhile (mirrors the record sink's documented "drain under the lock, I/O unlocked, re-lock to settle"). A slow flush can't stall the request path. #3 — the buffer is hard-bounded. The soft ceiling still signals an eager off-runtime flush; a new hard cap (`AUDIT_SINK_MAX_EVENTS`, well above the ceiling) is the OOM backstop: at the cap `emit` drops (counted via the new `ourios.audit_sink.dropped` metric, logged once) rather than grow without bound under sustained store-unavailability. Dropped template events degrade those templates to retained/empty bodies until the WAL re-mines them on restart — bounded memory is the deliberate trade. #1/#2 — publication is audit-ordered and race-free via snapshot-then- ordered-write. A new `PublishCoordinator` (ourios-ingester) drains both sink buffers into owned batches under the pipeline's miner lock (a microsecond memory move, no I/O — atomic w.r.t. `ingest`, closing the cadence TOCTOU race), then writes off-lock: the audit batch to durability first, the record partitions only after. A transient audit failure holds the records (requeued, retried next cadence); a permanent audit failure drops the audit batch and still publishes the records (the documented degraded case). The receiver's age-sweep now publishes through it. The inline size/ceiling trigger routes through a new record-sink audit barrier (`ParquetRecordSink::with_audit_barrier`) that flushes the audit sink to durability before the partition is put — race-free because that publish runs under the miner lock. Rotation/shutdown already drain audit-before- record under the miner lock (`flush_then_snapshot`), unchanged. The record sink gains a drain/publish/requeue split (`drain_aged` / `drain_all` / `requeue` / `publish_owned`) so the coordinator can move the encode+put off the lock; its existing RFC 0014 emit/flush behavior and tests are intact. Tests: the coordinator holds records when the audit write fails transiently (no data partition published though the data store is healthy); the size trigger flushes audit-before-publish and is skipped when audit can't drain; the hard cap drops + bounds; transient retains vs permanent drops; the metrics export; plus all round-1/2 tests and the #302 regression stay green. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
jensholdgaard
force-pushed
the
fix-302-receiver-audit-sink
branch
from
June 29, 2026 08:21
6f2f21f to
306cee6
Compare
Owner
Author
|
Round-3 review (maintainer-approved heavy lift) — all four addressed. The invariant is now: a mined record never becomes query-visible before its template's audit event is durable, on every publication path and race-free.
Rotation/shutdown were already audit-first + atomic under the miner lock ( |
PublishCoordinator.requeue re-buffers a transient-failed batch ahead of records emit added during the off-lock publish, but left PartitionBuffer.oldest at the newer records' timestamp — so the already-aged requeued records could miss the next age-sweep and retry late. Pin oldest to the age threshold (min with the existing oldest) on requeue so the next sweep re-drains promptly. Test: requeue_keeps_the_partition_aged_for_prompt_retry. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@crates/ourios-ingester/src/audit_sink.rs`:
- Around line 93-101: The audit sink error paths are using unstructured
eprintln!, which should be replaced with the project’s structured Ourios
logging. Update the audit-sink error handling in audit_sink.rs, including the
code paths around the existing audit sink drop/permanent write error messages,
to log through the structured logger with fields for tenant, partition date,
cap, and error. Use the existing audit sink logging setup/symbols in
audit_sink.rs so the messages remain machine-readable and consistent with the
rest of the subsystem.
- Around line 27-32: The hard-cap drop path in audit_sink.rs allows discarded
audit events to look fully drained, which can let WAL snapshot advancement skip
records that were never safely persisted. Update the buffering/emit flow in the
audit sink so hard-cap drops are tracked as an undrained or backpressured state,
or ensure the affected records retain their body text before snapshot
advancement can occur. Use the existing bounded-buffer logic in emit and the
drop accounting around the max_events cap to prevent dropped template events
from being treated as eligible for WAL recovery.
🪄 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: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: 0e77df23-6a94-4b33-84ee-4c9e921d91e5
📒 Files selected for processing (15)
crates/ourios-ingester/Cargo.tomlcrates/ourios-ingester/src/audit_sink.rscrates/ourios-ingester/src/lib.rscrates/ourios-ingester/src/metrics.rscrates/ourios-ingester/src/publish.rscrates/ourios-ingester/src/record_sink.rscrates/ourios-ingester/tests/audit_sink_metrics.rscrates/ourios-parquet/src/audit_writer.rscrates/ourios-parquet/src/lib.rscrates/ourios-semconv/src/lib.rscrates/ourios-server/src/receiver.rscrates/ourios-server/tests/rfc0013_6_wal_stays_local.rscrates/ourios-server/tests/rfc0019_storage_backend.rssemconv/registry/attributes.yamlsemconv/registry/metrics.yaml
✅ Files skipped from review due to trivial changes (2)
- crates/ourios-parquet/src/lib.rs
- crates/ourios-semconv/src/lib.rs
🚧 Files skipped from review as they are similar to previous changes (8)
- semconv/registry/attributes.yaml
- crates/ourios-ingester/Cargo.toml
- crates/ourios-ingester/src/lib.rs
- crates/ourios-ingester/tests/audit_sink_metrics.rs
- crates/ourios-server/tests/rfc0013_6_wal_stays_local.rs
- crates/ourios-server/tests/rfc0019_storage_backend.rs
- crates/ourios-parquet/src/audit_writer.rs
- crates/ourios-server/src/receiver.rs
…p) (#302) Round 5 on #312. Item 1: reword the audit_sink module-doc typo "The Drain miner emits…" → "The template miner emits…". Item 2 (CodeRabbit Major — the round-2 hard-cap drop could lose data): the hard cap dropped audit events at `AUDIT_SINK_MAX_EVENTS`, which is unsafe. A dropped event isn't counted by `buffered_events()`, so the no-loss snapshot gate (`flush_then_snapshot`) doesn't see it, the miner snapshot advances past that line's WAL position, and on restart the template event is never re-mined → those clean rows become permanently unreconstructable (a §3.3 violation), not merely degraded-until-restart. Adopt the record sink's posture (follow the reference, §5.4): under sustained store-unavailability the audit buffer is RETAINED and may transiently exceed the ceiling — never dropped. The WAL is the durability of record; the snapshot gate prevents loss because it won't advance while the buffer is non-empty. - Delete `AUDIT_SINK_MAX_EVENTS` and the drop branch in `buffer_event`; `emit` always buffers. The soft ceiling still fires the `Notify` for an eager off-runtime flush — the bound for the realistic (healthy-store) case. `requeue_ahead` no longer caps/drops. - Remove the `ourios.audit_sink.dropped` metric: reverted its `semconv/registry/metrics.yaml` entry, regenerated `ourios-semconv` (the const is gone; weaver no-diff verified), and dropped its use. - Replace the `hard_cap_drops_and_bounds_the_buffer` test with `persistent_store_failure_retains_every_event_never_drops`: under a persistently failing store, repeated emit + flush retains every event (buffer grows past the ceiling) and drops nothing. - Module docs state the posture explicitly (healthy store → ceiling + Notify bound it; sustained outage → retained, like the record sink; OOM under a total outage is the same accepted failure mode the record sink carries). The transient-vs-permanent flush classification (Io retain / Batch + Parquet + PartitionMismatch + Poisoned drop+count) is unchanged — that's about un-writable content, unrelated to the memory bound. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Owner
Author
|
Follow-up on the hard-cap (the round-4 CodeRabbit "don't let hard-cap drops advance past WAL recovery" point): reversed in |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Fixes #302.
Problem
The receiver wired the miner with
with_record_sinkonly — no audit sink — so the miner'stemplate_created/template_widened/template_type_expandedevents went to the defaultNoOpAuditSinkand never reached the RFC 0005 audit Parquet stream. The querier's read-time template registry (RFC 0017derive_template_registry) was therefore empty, andrender_log_bodyfell back to the row's retained body — which is empty for clean, high-confidence rows. Net: queries over freshly-ingested clean logs returned rows with empty body text, breakingCLAUDE.md§3.3 ("show me what was actually logged"). Cross-backend (local + S3 identical); predates RFC 0019.Fix — a buffering, shared audit sink (mirrors the RFC 0014 record sink)
ParquetAuditSink.emitdoes a blocking store write per event, and the miner emits synchronously on the async ingest request path — so wiring it directly would stall handlers on store latency and spray one tiny object per template event. Instead, following the record sink (§5.4):BufferingAuditSink+SharedParquetAuditSink(ourios-ingester):emitonly buffers (cheap, request-path-safe);flushdrains the buffer, groups events by audit partition, and writes each partition's batch with oneAuditWriter— few files, not one-per-event.spawn_blockingage-sweep /block_in_placerotation + shutdown) at the same cadence/rotation/shutdown points as the record sink, and wires it into the miner before recovery so replay re-emits template events.Invariants / hazards addressed
Faithfulfrom their template, bit-for-bit. Covered by a true regression test (fails before the fix) + property-style assertions in the e2e.flush_then_snapshotnow gates the miner snapshot on both sinks draining: a flush failure retains events and skips the snapshot so recovery re-mines + re-emits them. No acknowledged data is lost.Store(local or S3); the WAL stays local.Known limitation (documented)
The record sink's inline 256 MiB size trigger is not paired with an audit flush; a clean row flushed by that trigger may render empty in the narrow window before the next audit cadence flush (≤ the sweep tick), self-healing thereafter, and a crash in that window re-mines from the WAL — so durability is unaffected.
Tests
audit_sinkunits: per-partition batching round-trip (2 partitions → 2 files) + empty-buffer no-op.receiver_persists_template_audit_so_clean_rows_reconstruct(in-processserve()+ OTLP/HTTP, local): ingest clean logs → shutdown drain → derive registry → every row rendersFaithful+ bit-identical. Genuine regression (fails before the fix).rfc0019_3/rfc0019_5now also assert body text reconstructs faithfully out of S3 (was rows + stats + isolation only).rfc0013_6data round-trip scoped todata/(audit Parquet now coexists underaudit/).Verification (local)
cargo fmt --all --check✅ ·cargo clippy -p ourios-ingester -p ourios-parquet -p ourios-server --all-targets --all-features✅rfc0013_6all pass. localstackrfc0019_3/.5are#[ignore]d (run in thes3-integrationCI job); they compile + are clippy-clean.🤖 Generated with Claude Code
Summary by CodeRabbit
body.linetext can be faithfully restored from persisted templates.