feat(server): wire the RFC 0014 data write path live (rfc0013 green .6) - #245
Merged
Conversation
The production data path now flows OTLP → WAL → miner → ParquetRecordSink → object store. The miner emits through a SharedParquetSink (a cloneable Arc<Mutex<ParquetRecordSink>> handle); the pipeline drives the flush triggers the sink can't observe itself — flush_all on WAL segment rotation (RFC0014.3) and flush_aged on a batch-window age sweep (RFC0014.2). The store is a LocalFileSystem-backed Store rooted at bucket_root; S3 selection (RFC 0004) is the RFC 0014 §7 follow-on. WAL-durability invariant (CLAUDE.md §3.4, hazard #3). The sink buffer is an in-memory accelerator, never the durability of record — records are WAL-fsync'd before they reach it. No-loss across a crash is preserved by a single ordering rule applied at every miner-snapshot cadence point (post-recovery, rotation, graceful shutdown): flush the sink *before* writing the snapshot. That keeps the miner's snapshot horizon at or below the sink's flushed horizon, so startup recovery's miner-gated replay re-emits every un-flushed acknowledged record into a fresh sink (the crash discards only the volatile buffer). At-least-once: records flushed just before a crash may be re-flushed on restart; no record is lost. Multi-tenancy (§3.7) is unchanged — buffers are keyed by PartitionKey, which carries tenant_id. RFC0013.6 (WAL stays local) is greened end to end through the served binary in ourios-server/tests/rfc0013_6_wal_stays_local.rs: it ingests over HTTP, SIGTERMs (the shutdown drain flushes), and asserts only Parquet/manifest objects land under bucket_root while the WAL *.wal segments stay on the disjoint local wal_root. The stub in ourios-parquet/tests is redirected there (that crate has no WAL or server to observe). All eight RFC 0013 §5 scenarios now pass → RFC 0013 green. RFC0014.5 (crash no-loss) stays #[ignore]d pending its dedicated crash fixture (the next PR), which exercises this same recovery path under a real SIGKILL. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
|
Warning Review limit reached
More reviews will be available in 37 minutes and 47 seconds. Learn how PR review limits work. Your organization has used up its prepaid credits, and credit purchases are no longer available. Enable the review add-on in the billing tab to keep reviews running — you're only billed for reviews past your plan's rate limits ($0.25/file). ⌛ How to resolve this issue?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 credits. 🚦 How do rate limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan refill rate. 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, the refill rate gradually slows as usage increases. The highest same-day bursts are limited more strictly. 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 (2)
📝 WalkthroughWalkthroughAdds a cloneable ChangesRFC 0013 .6: WAL-local Parquet flush integration
Sequence Diagram(s)sequenceDiagram
participant Client as OTLP HTTP Client
participant Receiver as ourios-server receiver
participant MinerCluster
participant SharedParquetSink
participant AgeSweep as age-sweep task
participant ObjectStore as Object Store bucket_root
participant WAL as WAL wal_root
Client->>Receiver: POST /v1/logs (OTLP protobuf)
Receiver->>WAL: append to WAL segment
Receiver->>MinerCluster: mine record
MinerCluster->>SharedParquetSink: emit(MinedRecord)
AgeSweep->>SharedParquetSink: periodic flush_aged()
SharedParquetSink->>ObjectStore: write Parquet partitions
Note over Receiver: SIGTERM received
Receiver->>AgeSweep: abort + await flush_tick
Receiver->>SharedParquetSink: flush_all() via flush_then_snapshot
alt buffered_records == 0
SharedParquetSink->>ObjectStore: final Parquet flush
Receiver->>WAL: write_snapshots(offset)
else flush incomplete
Note over Receiver: skip snapshot, records remain in WAL for recovery
end
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 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 |
Contributor
There was a problem hiding this comment.
Pull request overview
Wires the production data write path into ourios-server so mined records are buffered and flushed to Parquet via the RFC 0013 Store, while ensuring WAL segments remain on local disk (greening RFC0013.6 and marking RFC 0013 as green).
Changes:
- Add a shared, cloneable
SharedParquetSinkso the miner canemitwhile the pipeline/server drives flush triggers (rotation, age sweep, shutdown). - Wire
Store::local(bucket_root)+ParquetRecordSinkinto the receiver role, with flush-on-rotation, periodic age sweep, and drain-on-shutdown. - Add an end-to-end served-binary test for RFC0013.6 and update RFC 0013 documentation/status accordingly.
Reviewed changes
Copilot reviewed 6 out of 6 changed files in this pull request and generated 2 comments.
Show a summary per file
| File | Description |
|---|---|
| docs/rfcs/0013-object-storage.md | Marks RFC 0013 as green and updates status note to reflect completed scenarios, including RFC0013.6. |
| crates/ourios-server/tests/rfc0013_6_wal_stays_local.rs | New end-to-end served-binary test asserting WAL stays under wal_root while Parquet/manifest land under bucket_root. |
| crates/ourios-server/src/receiver.rs | Wires Store + SharedParquetSink into receiver startup/recovery/rotation/shutdown and adds periodic age-sweep flushing. |
| crates/ourios-server/src/main.rs | Passes bucket_root into the receiver config for the write path. |
| crates/ourios-parquet/tests/rfc0013_object_store.rs | Removes the RFC0013.6 stub here and documents that it’s now covered end-to-end in ourios-server. |
| crates/ourios-ingester/src/record_sink.rs | Introduces SharedParquetSink (Arc+Mutex wrapper) and adds a unit test for shared buffering + flushing across clones. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
… root Addresses Copilot review on #245: - `spawn_age_sweep` ran `flush_aged` (Parquet encode + blocking store I/O) directly on a runtime worker; move it to `spawn_blocking` so periodic sweeps never stall the receiver (esp. against S3). - `Store::local` errors if `bucket_root` is missing, which regressed the server's ability to start on a not-yet-created `OURIOS_BUCKET_ROOT`; `create_dir_all` the root before opening the store. The RFC0013.6 test now omits its own `create_dir_all`, so it covers this path. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Closes a no-loss hole Copilot flagged on #245. `flush_all` retains a partition whose store write fails (the WAL is the durability of record), but the cadence points wrote the miner snapshot unconditionally afterward. With the store unavailable that advances the snapshot horizon past records that never reached object storage — and recovery suppresses frames at or below the horizon, so on the next start they are never re-emitted into a fresh sink. That is data loss precisely when the store is down. Fix: a `flush_then_snapshot` helper flushes, then writes the snapshot only if `buffered_records() == 0`. If the sink didn't drain it skips the snapshot (logged), so the horizon can't outrun the flushed horizon and recovery re-mines the un-flushed records. Applied at all three cadence points (post-recovery, rotation, shutdown). The rotation hook — which runs on the request path — wraps the blocking flush + snapshot in `block_in_place` so it doesn't stall a runtime worker. Tested: a fault-injection unit test replaces the store root with a file so `put_blocking` fails, then asserts the snapshot is skipped and the records stay buffered (not lost); plus the drained happy path. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
…shes Addresses Copilot review round 3 on #245: - Age sweep: set the interval's missed-tick behavior to `Delay` so a slow sweep (S3) can't trigger catch-up bursts of back-to-back flushes. - `flush_then_snapshot` doc: the bool is "did the sink drain" — clarified that a `write_snapshots` failure still returns `true` (the data reached the store; the snapshot is a rebuildable-cache miss), resolving the contradiction with the old "snapshot written" wording. - Wrap the post-recovery and shutdown cadence flushes in `block_in_place` too (not just the rotation hook), so their blocking Parquet/store I/O doesn't stall a runtime worker. - `SharedParquetSink` doc: correct the "short critical sections" claim — `flush_all`/`flush_aged` hold the mutex across encode + `put_blocking`, so they are blocking sections (benign sub-ms on the local backend). Moving the encode+put outside the lock is tracked for the S3 backend (RFC 0014 §7 / RFC 0013), where PUTs are slow enough to matter. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
![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
🤖 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-server/tests/rfc0013_6_wal_stays_local.rs`:
- Around line 66-83: The http_post_logs function uses .ok() to silently discard
the result of stream.flush(), which makes debugging harder if the flush fails.
Replace the .ok() call on the stream.flush().await line with .expect() and
provide a descriptive error message like "flush stream" to ensure failures are
explicit and visible during test execution.
- Around line 85-103: The files_under function silently ignores filesystem
errors using the `else { continue; }` pattern and `.flatten()` which skips
errored entries. Instead of continuing silently on errors, make the function
fail fast by using expect() or unwrap() on the read_dir result to panic with a
clear message when directory reading fails, and similarly handle any errors from
entry enumeration rather than silently skipping them with flatten(). This
ensures that filesystem problems in the test environment (setup failures,
permissions, races) are caught immediately with clear error messages rather than
causing confusing failures later.
🪄 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: adfe3809-37bf-4cad-810f-575686493c52
📒 Files selected for processing (6)
crates/ourios-ingester/src/record_sink.rscrates/ourios-parquet/tests/rfc0013_object_store.rscrates/ourios-server/src/main.rscrates/ourios-server/src/receiver.rscrates/ourios-server/tests/rfc0013_6_wal_stays_local.rsdocs/rfcs/0013-object-storage.md
Addresses review round 4 on #245: - Shutdown awaits the age-sweep task instead of aborting it. Aborting the async task would orphan an in-flight `spawn_blocking` flush still holding the sink mutex (which the shutdown drain would then wait on anyway); the task already observes the `shutdown` watch signal, so awaiting lets the in-flight flush finish and the task exit cleanly (Copilot). - RFC0013.6 test: `http_post_logs` `.expect()`s the flush, and `files_under` fails fast on `read_dir`/entry errors rather than silently skipping — in a controlled temp tree those are real problems (CodeRabbit). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Round 5 (Copilot): a failed snapshot write doesn't force a *full* replay — prior snapshots may survive and partial per-tenant writes may have landed. Reword the cadence log from "next start full-replays" to "next start may replay more from the WAL" so it doesn't mislead operators. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This was referenced Jun 23, 2026
This was referenced Jun 29, 2026
Merged
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.
What
Wires the production data write path live in the server: OTLP → WAL → miner →
ParquetRecordSink→ object store. This greens RFC0013.6 (WAL stays local) → RFC 0013 is now green (8/8), and lays the recovery groundwork the RFC0014.5 crash test will exercise next.SharedParquetSink(ourios-ingester): a cloneableArc<Mutex<ParquetRecordSink>>handle. The mineremits through it; the pipeline drives the flush triggers the sink can't observe itself.ourios-server): builds aLocalFileSystem-backedStoreatbucket_root, injects the sink into the miner, force-flushes on WAL rotation (RFC0014.3), runs an age-sweep tick (RFC0014.2), and drains on graceful shutdown. S3 selection (RFC 0004) is the RFC 0014 §7 follow-on.Invariants addressed
WAL-before-ack / no-loss (
CLAUDE.md§3.4, hazard #3). The sink buffer is an in-memory accelerator, never the durability of record — records are WAL-fsync'd before they reach it. No-loss across a crash is preserved by one ordering rule at every miner-snapshot cadence point (post-recovery, rotation, shutdown): flush the sink before writing the snapshot. That keeps the miner's snapshot horizon ≤ the sink's flushed horizon, so startup recovery's miner-gated replay re-emits every un-flushed acknowledged record into a fresh sink. Semantics are at-least-once: records flushed just before a crash may be re-flushed on restart; none are lost. (Exactly-once dedup is a separate downstream concern, not in RFC 0014's no-loss scope.)Multi-tenancy (§3.7). Unchanged — buffers are keyed by
PartitionKey, which carriestenant_id; no buffer or flush crosses tenants.Tests
ourios-server/tests/rfc0013_6_wal_stays_local.rs): ingest over HTTP, SIGTERM, assert only Parquet/manifest objects land underbucket_rootand the WAL*.walsegments stay on the disjoint localwal_root; the flushed Parquet rounds back out (separation isn't vacuous). Theourios-parquetstub is redirected here (that crate has no WAL/server to observe).SharedParquetSinkunit test (shared buffer across clones; flush via the handle drains it).RFC0014.5 (crash no-loss) stays
#[ignore]d pending its dedicated SIGKILL crash fixture (next PR), which exercises this same recovery path.Verification
cargo fmt --all --check,cargo clippy -p ourios-ingester -p ourios-parquet -p ourios-server --all-targets --all-features -D warnings, the three crates' test suites, andmdbook build— all green locally.🤖 Generated with Claude Code
Summary by CodeRabbit
New Features
Documentation