fix(ingester): rfc 0035 review fixes — capture-slot unwind safety, honest barrier docs - #579
Conversation
…batch panic salvage A miner panic inside ingest_mined now settles the capture slot before the unwind continues: the slot always resets (a stale Captured can no longer swallow the next batch item) and a record captured before the panic is forwarded to the real sink and counted (mined_capture_salvages_total) instead of being dropped until restart replay. The pipeline's ordered phase catch-unwinds the per-batch mining loop and submits the records mined before the panic to the encode pool before re-panicking — pre-split those records had already been emitted inline, and the deferral must not widen that survived failure mode into whole-batch loss. Covered by unit tests on both unwind paths and an integration test that panics the miner mid-batch (injected audit-sink panic) and asserts the pre-panic records reach Parquet and the next batch acks. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01WQY9wfrfRggqSpMLH8Xj3Y
…rability claim The barrier comment overclaimed: a record the mark claims durable can also be in an age-sweep's in-flight off-lock write_ordered — neither buffered (invisible to flush_all) nor yet durable at stamp time. State the pool-coverage claim accurately and reference issue #578 for the sweep window (mechanism deliberately not implemented here). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01WQY9wfrfRggqSpMLH8Xj3Y
…then_snapshot path Two arms against the production rotation_snapshot_hook with the never-flush-sized production config: a buffered-but-unflushed record at or below the mark is flushed by the hook before the snapshot is stamped (and the stamp carries the rotation-point mark), and when the store cannot accept the flush the stamp is skipped — the records are retained and no snapshot advances the horizon past them. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01WQY9wfrfRggqSpMLH8Xj3Y
|
Warning Review limit reached
Next review available in: 40 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 (3)
📝 WalkthroughWalkthroughThe PR separates soak encode-pool sizing from runtime worker sizing, makes rotation draining runtime-aware, adds miner panic capture salvage, and adds RFC0035 tests for panic recovery and snapshot durability. ChangesPipeline reliability and configuration
Estimated code review effort: 4 (Complex) | ~60 minutes Possibly related PRs
Sequence Diagram(s)sequenceDiagram
participant IngestTask
participant IngestPipeline
participant MinerCluster
participant AuditSink
participant ParquetSink
IngestTask->>IngestPipeline: ingest batch
IngestPipeline->>MinerCluster: ingest_mined(record)
MinerCluster->>AuditSink: process mined record
AuditSink-->>MinerCluster: panic after capture
MinerCluster->>ParquetSink: salvage captured record
MinerCluster-->>IngestPipeline: resume panic
IngestTask->>IngestPipeline: follow-up batch
IngestPipeline->>ParquetSink: persist recovered batch
🚥 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 |
There was a problem hiding this comment.
Pull request overview
This PR applies the compensating review fixes for RFC 0035’s ordered/concurrent ingest split, focusing on correctness under panic/unwind and on making the rotation barrier’s durability claims both test-backed and accurately documented.
Changes:
- Make pooled ingest unwind-safe: ensure already-mined records are submitted before re-panicking, and ensure the miner’s capture slot is always settled across unwinds (with salvage observability).
- Strengthen/align rotation barrier coverage by pinning the “flush half” through the production
rotation_snapshot_hook→flush_then_snapshotpath (including a sabotage arm that proves “no stamp on failed drain”). - Add a soak harness knob to independently control encode-pool sizing (
--encode-workers) to mirror the server’sreceiver.encode_workers.
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-server/src/receiver.rs | Adds rotation-hook-backed tests that prove buffered records flush before stamping, and stamping is skipped when flush cannot drain. |
| crates/ourios-miner/src/cluster.rs | Makes ingest_mined unwind-safe via catch_unwind, salvages captured records on panic, and adds unit tests for slot cleanliness + salvage behavior. |
| crates/ourios-ingester/tests/it/rfc0035_f2_miner_panic_salvage.rs | New integration test driving a real pooled pipeline to assert pre-panic records reach Parquet and the next batch still acks/lands. |
| crates/ourios-ingester/tests/it/main.rs | Registers the new RFC0035 F2 integration test module. |
| crates/ourios-ingester/src/receiver/pipeline.rs | Submits already-mined records before re-panicking on mid-batch miner panics; updates barrier commentary; flavor-guards block_in_place for rotation drain+hook. |
| crates/ourios-bench/src/soak.rs | Adds encode_workers to soak config and validation; uses it for the encode pool size. |
| crates/ourios-bench/src/main.rs | Adds --encode-workers CLI flag and threads it into SoakConfig. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (3)
crates/ourios-ingester/src/receiver/pipeline.rs (1)
424-434: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winConsider extracting the flavor-guarded drain into a helper.
The flavor-check +
block_in_place/inline dispatch adds another branch to an already largeingest_bound. Pulling it into a small method (mirroring the existingfire_rotation_hookextraction) would keepingest_bound's critical section easier to scan.♻️ Proposed extraction
- let drain_and_hook = || { - self.quiesce_encodes(); - self.fire_rotation_hook(&miner, prev); - }; - if tokio::runtime::Handle::current().runtime_flavor() - == tokio::runtime::RuntimeFlavor::MultiThread - { - tokio::task::block_in_place(drain_and_hook); - } else { - drain_and_hook(); - } + self.quiesce_and_fire_rotation_hook(&miner, prev);/// Drain the encode pool and fire the rotation hook, blocking a /// multi-thread worker without starving the runtime; runs inline on /// a current-thread runtime, where `block_in_place` would panic. fn quiesce_and_fire_rotation_hook(&self, miner: &MinerCluster, mark: WalOffset) { let drain_and_hook = || { self.quiesce_encodes(); self.fire_rotation_hook(miner, mark); }; if tokio::runtime::Handle::current().runtime_flavor() == tokio::runtime::RuntimeFlavor::MultiThread { tokio::task::block_in_place(drain_and_hook); } else { drain_and_hook(); } }🤖 Prompt for 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. In `@crates/ourios-ingester/src/receiver/pipeline.rs` around lines 424 - 434, Extract the flavor-guarded drain-and-hook closure into a helper method named quiesce_and_fire_rotation_hook, preserving the existing MultiThread block_in_place behavior and inline dispatch for current-thread runtimes. Replace the inline branch in ingest_bound with a call to this helper, passing miner and prev.crates/ourios-miner/src/cluster.rs (2)
192-192: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winConsider wiring
mined_capture_salvagesinto the OTel/Prometheus surface.The comment states a non-zero value "means a panic fired between capture and return" — i.e. this counter is meant to signal a real production anomaly. Today it's only reachable via the
mined_capture_salvages_total()accessor with no caller in the reviewed files that polls or exports it, so an operator has no way to alert on it firing in production without new plumbing. SinceMinerMetricsalready exposes several counters this way (merges_total,parse_failures_total, etc.), wiring this one in similarly would give it real observability.Also applies to: 1266-1271
🤖 Prompt for 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. In `@crates/ourios-miner/src/cluster.rs` at line 192, The mined_capture_salvages counter is not exposed through the existing metrics surface, so production alerts cannot observe it. Wire mined_capture_salvages_total() into the same OTel/Prometheus registration or export path used by MinerMetrics counters such as merges_total and parse_failures_total, preserving its counter semantics and existing accessor.
4730-4755: 📐 Maintainability & Code Quality | 🔵 TrivialTest correctly exercises
salvage_mined_capturein isolation, given the real path isn't externally injectable.The comment at line 4741 honestly documents the limitation: the tail of
ingestpast the emit can't be panicked from outside the crate, so this stages the post-capture state by hand and calls the private helper directly rather than driving it through a realcatch_unwind/resume_unwindround-trip. That's a reasonable compromise given the constraint, but it means no test in this diff exercises the combined real-panic-after-capture →resume_unwind→ salvage flow end-to-end (only "panic before capture" is driven through a real panic, inrfc0035_f2_capture_slot_is_clean_after_a_panic). Not blocking — just noting the coverage gap is inherent to the design's injection points, not something this PR left on the table for free.🤖 Prompt for 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. In `@crates/ourios-miner/src/cluster.rs` around lines 4730 - 4755, No code change is required: keep rfc0035_f2_salvage_forwards_a_captured_record_to_the_sink as an isolated salvage test, since the real post-capture panic path is not externally injectable. Preserve the existing coverage and acknowledge that end-to-end panic/resume/salvage coverage is intentionally unavailable.
🤖 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-bench/src/soak.rs`:
- Around line 328-330: Update the encode_workers validation in the soak
configuration setup to reject values above a documented operational maximum
before constructing EncodePool. Ensure the maximum prevents queue-capacity
multiplication overflow and unreasonable OS thread creation, while preserving
the existing rejection of zero and passing valid values through unchanged.
In `@crates/ourios-ingester/src/receiver/pipeline.rs`:
- Around line 424-434: Add a focused current-thread Tokio rotation test
alongside the existing multi-thread rotation coverage, using
#[tokio::test(flavor = "current_thread")] and triggering a rotation through the
relevant receiver pipeline setup. Assert the rotation completes successfully so
the else branch calling drain_and_hook directly is exercised.
---
Nitpick comments:
In `@crates/ourios-ingester/src/receiver/pipeline.rs`:
- Around line 424-434: Extract the flavor-guarded drain-and-hook closure into a
helper method named quiesce_and_fire_rotation_hook, preserving the existing
MultiThread block_in_place behavior and inline dispatch for current-thread
runtimes. Replace the inline branch in ingest_bound with a call to this helper,
passing miner and prev.
In `@crates/ourios-miner/src/cluster.rs`:
- Line 192: The mined_capture_salvages counter is not exposed through the
existing metrics surface, so production alerts cannot observe it. Wire
mined_capture_salvages_total() into the same OTel/Prometheus registration or
export path used by MinerMetrics counters such as merges_total and
parse_failures_total, preserving its counter semantics and existing accessor.
- Around line 4730-4755: No code change is required: keep
rfc0035_f2_salvage_forwards_a_captured_record_to_the_sink as an isolated salvage
test, since the real post-capture panic path is not externally injectable.
Preserve the existing coverage and acknowledge that end-to-end
panic/resume/salvage coverage is intentionally unavailable.
🪄 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: 9633cca7-aff9-4ab5-81e0-c92b90ece765
📒 Files selected for processing (7)
crates/ourios-bench/src/main.rscrates/ourios-bench/src/soak.rscrates/ourios-ingester/src/receiver/pipeline.rscrates/ourios-ingester/tests/it/main.rscrates/ourios-ingester/tests/it/rfc0035_f2_miner_panic_salvage.rscrates/ourios-miner/src/cluster.rscrates/ourios-server/src/receiver.rs
…ode_workers F5: the rotation branch's quiesce + hook can block a runtime worker for seconds (queued encodes + the hook's store I/O); wrap them in block_in_place so the runtime relocates other tasks — flavor-guarded, since block_in_place panics on a current-thread runtime (the crash fixtures run one and must not gain a rotation landmine). F6: the soak harness gets its own --encode-workers knob (validated > 0, default = host parallelism) mirroring the server's receiver.encode_workers instead of overloading worker_threads, so the two axes — tokio load runtime vs encode OS threads — are tunable independently. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01WQY9wfrfRggqSpMLH8Xj3Y
7dc8377 to
3a7e8cb
Compare
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 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/receiver/pipeline.rs`:
- Around line 428-434: Add a focused test for the rotation path containing the
runtime_flavor check, run under a Tokio current_thread runtime so the inline
drain_and_hook branch executes without block_in_place. Assert that rotation
completes successfully and preserves the expected hook/drain behavior.
🪄 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: c73d201b-7830-487a-82e9-6906305c00d6
📒 Files selected for processing (3)
crates/ourios-bench/src/main.rscrates/ourios-bench/src/soak.rscrates/ourios-ingester/src/receiver/pipeline.rs
🚧 Files skipped from review as they are similar to previous changes (2)
- crates/ourios-bench/src/soak.rs
- crates/ourios-bench/src/main.rs
…d rotation test Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01WQY9wfrfRggqSpMLH8Xj3Y
…rted telemetry Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01WQY9wfrfRggqSpMLH8Xj3Y
What
The required changes from #577's compensating adversarial review (the gate that the accidental out-of-band landing skipped — see #577's closing record):
ingest_bound) nor leak aCapturedrecord across the unwind (slot settles on every exit; salvage counter, expected 0). Integration test: real pooled pipeline, record 3 of 4 panics → the 2 pre-panic records reach Parquet, the next batch acks, slot clean.flush_then_snapshotpath — one arm proves buffered records ≤ mark flush before the stamp; the sabotage arm proves the stamp is skipped when flush can't drain (and found+fixed a race in its own first assertion by quiescing first).block_in_place(flavor-guarded — panics on current-thread runtimes otherwise); soak gets its own--encode-workersknob mirroring the server's.Verification
fmt / clippy (workspace, all features) / full nextest 1197 passed, 0 failed /
cargo doczero warnings (the check whose absence caused the last CI failure). Phased commits ≤5 files.Invariants
Strengthens §3.4-adjacent durability (F2 restores a pre-split survived failure mode; F3 pins the flush contract) and makes the §3.1 barrier documentation truthful. No on-disk change.
🤖 Generated with Claude Code
https://claude.ai/code/session_01WQY9wfrfRggqSpMLH8Xj3Y
Summary by CodeRabbit
New Features
--encode-workersoption for soak testing to control encode-pool concurrency (defaults to the existing worker-thread count when omitted).Bug Fixes
Tests