Repository navigation
feat: Framer + CriAggregator — zero-copy, Kani-proven line splitting (Steps 4-5) - #311
Conversation
|
Caution Review failedThe pull request is closed. ℹ️ Recent review info⚙️ Run configurationConfiguration used: Organization UI Review profile: ASSERTIVE Plan: Pro Run ID: 📒 Files selected for processing (3)
WalkthroughAdds three new public modules to logfwd-core: Comment |
020f50c to
01253d0
Compare
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
There was a problem hiding this comment.
Requesting changes due to two correctness issues introduced in this PR:
-
crates/logfwd-core/src/framer.rs(around line 93): whenMAX_LINES_PER_FRAMEis reached, the loop still advancesstarton later newlines. As a result, extra complete lines are neither emitted nor preserved in remainder (data loss). A concrete trigger is input with more than 4096 newline-terminated lines. -
crates/logfwd-core/src/framer.rs(around line 194): CRI flags parsing treats any token starting withF(e.g.FOO) as full. This can incorrectly flush pending partial aggregates on malformed CRI input. Please require exactF/Pflag tokens (or reject malformed lines).
What is this? | From workflow: AI: PR Review
Give us feedback! React with 🚀 if perfect, 👍 if helpful, 👎 if not.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 4
🤖 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-core/src/aggregator.rs`:
- Around line 20-31: The doc comment for the aggregator currently describes
returning Some/None but the public API now returns AggregateResult; update the
documentation in crates/logfwd-core/src/aggregator.rs to describe the
AggregateResult enum and its variants (AggregateResult::Complete(payload) for
finished messages and AggregateResult::Pending for buffered partial lines),
adjust the lifecycle examples to use Complete/Pending instead of Some/None, and
ensure any example text mentions the zero-copy fast path for Complete to keep
behavior accurate (reference AggregateResult and the aggregator function that
produces it).
In `@crates/logfwd-core/src/framer.rs`:
- Around line 157-159: The CRI flag parsing in frame (which uses NewlineFramer
and returns CriFrameOutput) currently treats any non-'F' first byte as partial
and accepts multi-byte/missing flags; change the parser so it only accepts a
single-byte ASCII flag that is exactly 'F' (full) or 'P' (partial) and rejects
(i.e., skip) lines where the flag is missing, multi-byte, or any other value;
update the is_full/is_partial decision in the frame parsing code (and the same
logic around the block at/near the existing check referenced by lines ~192-197)
to skip malformed CRI lines instead of coercing them to partial, preserving
aggregation boundaries.
- Around line 130-132: CriFrameOutput currently uses heap-allocating
Vec<CriLineInfo> (and Vec::with_capacity in frame()), which violates the
"fixed-size, no heap allocation" contract; change CriFrameOutput to mirror
FrameOutput by using a fixed-capacity array/buffer and an explicit count field
instead of Vec: replace lines: Vec<CriLineInfo> with something like lines:
[CriLineInfo; MAX_CRI_LINES_PER_FRAME] plus a lines_len: usize, remove any
Vec::with_capacity/new allocations in frame() and update push/iteration code to
use the count and indexed access, and ensure all places that constructed or
returned CriFrameOutput use the new fixed-size struct and the
MAX_CRI_LINES_PER_FRAME constant.
- Around line 91-101: In NewlineFramer::frame the loop currently skips emitting
empty lines and continues advancing start past un-emitted data when output.count
reaches MAX_LINES_PER_FRAME; update the loop so blank lines are emitted (change
the i > start check so a zero-length line at i==start is recorded into
output.line_ranges/output.count) and when output.count == MAX_LINES_PER_FRAME
stop processing further bytes instead of advancing start (break the loop and
leave output.remainder_offset pointing at start so the caller receives the
remaining data). Make these changes in the function handling input, i, start,
output.count, MAX_LINES_PER_FRAME, output.line_ranges and
output.remainder_offset.
🪄 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: Organization UI
Review profile: ASSERTIVE
Plan: Pro
Run ID: 9298e702-e84d-4737-aab6-39bc0c512f99
📒 Files selected for processing (3)
crates/logfwd-core/src/aggregator.rscrates/logfwd-core/src/framer.rscrates/logfwd-core/src/lib.rs
01253d0 to
3ce71ba
Compare
There was a problem hiding this comment.
♻️ Duplicate comments (1)
crates/logfwd-core/src/framer.rs (1)
93-103:⚠️ Potential issue | 🟠 MajorLines are silently dropped when
MAX_LINES_PER_FRAMEis reached.When
output.count == MAX_LINES_PER_FRAME, the loop continues advancingstartwithout emitting lines. The finalremainder_offset = startthen skips those un-emitted lines entirely. The caller loses data.The fix: break the loop when capacity is exhausted, leaving
remainder_offsetat the first byte of the un-emitted line.🐛 Suggested fix
while i < input.len() { if input[i] == b'\n' { - if i > start && output.count < MAX_LINES_PER_FRAME { + if output.count == MAX_LINES_PER_FRAME { + // Capacity full — stop and leave remainder at start + break; + } + if i > start { output.line_ranges[output.count] = (start, i); output.count += 1; } start = i + 1; } i += 1; }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@crates/logfwd-core/src/framer.rs` around lines 93 - 103, The loop in framer.rs that scans input by newline can advance start past lines when output.count reaches MAX_LINES_PER_FRAME, causing un-emitted lines to be dropped; change the logic in the while loop that updates output.line_ranges and start so that when output.count == MAX_LINES_PER_FRAME you stop emitting further lines and break out of the loop (do not update start to skip the first un-emitted line), leaving output.remainder_offset pointing at the start of that un-emitted line; touch the variables/locations referenced (input, start, i, output.count, output.line_ranges, MAX_LINES_PER_FRAME, output.remainder_offset) to ensure the loop exits early instead of continuing to advance start.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Duplicate comments:
In `@crates/logfwd-core/src/framer.rs`:
- Around line 93-103: The loop in framer.rs that scans input by newline can
advance start past lines when output.count reaches MAX_LINES_PER_FRAME, causing
un-emitted lines to be dropped; change the logic in the while loop that updates
output.line_ranges and start so that when output.count == MAX_LINES_PER_FRAME
you stop emitting further lines and break out of the loop (do not update start
to skip the first un-emitted line), leaving output.remainder_offset pointing at
the start of that un-emitted line; touch the variables/locations referenced
(input, start, i, output.count, output.line_ranges, MAX_LINES_PER_FRAME,
output.remainder_offset) to ensure the loop exits early instead of continuing to
advance start.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro
Run ID: 2605e128-9dad-48e4-aff0-d6d34fb00a82
📒 Files selected for processing (5)
crates/logfwd-core/src/aggregator.rscrates/logfwd-core/src/byte_search.rscrates/logfwd-core/src/cri.rscrates/logfwd-core/src/framer.rscrates/logfwd-core/src/lib.rs
3ce71ba to
66f9605
Compare
|
Implemented and pushed fixes for all 3 unresolved review threads in Addressed with code changes:
Replied instead of fixing:
Validation run
Note 🔒 Integrity filtering filtered 1 itemIntegrity filtering activated and filtered the following item during workflow execution.
What is this? | From workflow: AI: Address PR Review Feedback Give us feedback! React with 🚀 if perfect, 👍 if helpful, 👎 if not. |
New modules for the zero-copy pipeline:
framer.rs — identifies line boundaries without memchr or Vec:
- NewlineFramer: splits on \n, returns fixed-size FrameOutput
- CriFramer: splits on \n, extracts CRI field ranges (zero-copy)
- find_byte: plain loop (no memchr) for Kani provability
- LLVM auto-vectorizes the byte scan loops
- 4 Kani proofs: no-panic (32 bytes), ranges-valid, remainder-correct, find_byte
aggregator.rs — CRI P/F partial line merging:
- F-only fast path: zero-copy (returns reference to input)
- P accumulation: copies into internal Vec (frees read buffer)
- P+F emission: concatenated result from internal buffer
- max_message_size enforced on all paths (including F-only)
- 3 Kani proofs: F-only max size, P+F max size, P returns Pending
Design choices:
- No memchr dependency — plain byte loops are Kani-provable AND
LLVM auto-vectorizes them (verified with vectorization_canary pattern)
- Fixed-size FrameOutput (4096 entries, 64KB) — no heap allocation
- CriFramer delegates line splitting to NewlineFramer, adds field parsing
Total Kani proofs in codebase: 22 (13 prior + 4 framer + 3 aggregator + 2 CRI)
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
3259622 to
17e3d99
Compare
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
crates/logfwd-core/src/cri.rs (1)
98-112: 🧹 Nitpick | 🔵 Trivial
CriReassemblerF-only path copies (unlikeCriAggregator).Lines 100-111 always copy into
partial_buf, even when no partials are pending. This differs fromCriAggregator::feedwhich returns a zero-copy slice for F-only.The difference stems from lifetime constraints:
CriReassembler::feedtakes&CriLine<'_>where the message borrows external data, so returning it directly would tie the output to the caller's input lifetime.CriAggregatoruses&'a mut self, message: &'a [u8]to enable zero-copy.Consider documenting this design difference or consolidating if callers can be migrated.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@crates/logfwd-core/src/cri.rs` around lines 98 - 112, CriReassembler::feed currently always copies F-only messages into partial_buf (via partial_buf.extend_from_slice) because it accepts cri: &CriLine<'_> whose message borrows an external, shorter lifetime, unlike CriAggregator::feed which accepts message: &'a [u8] and can return a zero-copy slice; pick one of two fixes: either document this design difference in the CriReassembler::feed doc comment explaining why F-only lines are copied (referencing CriReassembler::feed, partial_buf, max_line_size and contrast to CriAggregator::feed), or change the API so CriReassembler::feed takes cri with a matching output lifetime (e.g. accept the message as &'a [u8] or change the CriLine lifetime parameter) allowing you to return &self-like slices and avoid the copy (update signatures of callers accordingly); implement the approach that matches call-site ownership (prefer documentation if callers cannot be migrated).
♻️ Duplicate comments (4)
crates/logfwd-core/src/framer.rs (3)
132-137:⚠️ Potential issue | 🟠 Major
CriFrameOutputheap-allocates despite "no heap allocation" doc.Line 132 claims "Fixed-size, no heap allocation" but line 134 uses
Vec<CriLineInfo>. Line 163 callsVec::with_capacity()on everyframe()call.
CriLineInfocontains only(usize, usize)tuples andbool— allCopytypes. It can use a fixed array likeFrameOutputdoes.This was flagged previously. Based on learnings: "Avoid per-line heap allocations in hot paths (scanner, CRI parser)".
♻️ Use fixed-size array
+impl Copy for CriLineInfo {} +impl Clone for CriLineInfo { + fn clone(&self) -> Self { *self } +} + /// Output of CRI framing. Fixed-size, no heap allocation for line storage. pub struct CriFrameOutput { - lines: Vec<CriLineInfo>, // CriLineInfo is non-Copy (contains ranges), use Vec + lines: [CriLineInfo; MAX_CRI_LINES_PER_FRAME], + count: usize, /// Byte offset of the first unprocessed byte. pub remainder_offset: usize, }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@crates/logfwd-core/src/framer.rs` around lines 132 - 137, CriFrameOutput currently uses a Vec<CriLineInfo> and calls Vec::with_capacity() in frame(), which contradicts the "no heap allocation" doc and causes per-frame heap work; change CriFrameOutput to store a fixed-size array (e.g. [CriLineInfo; N]) plus a length counter (like FrameOutput does) and make CriLineInfo derive/copy types (implement Copy/Clone if needed) so you can push entries without allocation; update the frame() implementation to write into the array and increment the count instead of creating a Vec, and adjust any consumers that iterate lines to use the length counter and array slice.
91-105:⚠️ Potential issue | 🔴 CriticalLines are silently dropped when
MAX_LINES_PER_FRAMEis reached.When
output.counthitsMAX_LINES_PER_FRAME, the loop continues advancingstartpast complete lines without recording them. On exit,remainder_offset = startskips those lines entirely — they're neither inline_rangesnor in the remainder for the next cycle.This was flagged in a previous review but remains unaddressed.
🐛 Fix: break when output is full
while i < input.len() { if input[i] == b'\n' { - if i > start && output.count < MAX_LINES_PER_FRAME { + if output.count == MAX_LINES_PER_FRAME { + // Output full — leave remaining bytes for next frame + break; + } + if i > start { output.line_ranges[output.count] = (start, i); output.count += 1; } start = i + 1; } i += 1; }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@crates/logfwd-core/src/framer.rs` around lines 91 - 105, The loop in framer that scans input bytes currently continues advancing start after MAX_LINES_PER_FRAME is reached, causing completed lines to be skipped; update the loop so that when input[i] == b'\n' and output.count == MAX_LINES_PER_FRAME you stop processing further lines (break/return) without advancing start so those remaining lines become the remainder. Concretely: in the branch handling newline, check output.count against MAX_LINES_PER_FRAME before assigning output.line_ranges/output.count and before setting start; if full, set output.remainder_offset = start (or leave start unchanged) and break/return immediately so no lines are silently dropped; adjust use of output.line_ranges, output.count and output.remainder_offset accordingly.
194-199:⚠️ Potential issue | 🟠 MajorMalformed CRI flags accepted instead of rejected.
Only the first byte is checked (
line[flags_start] == b'F'). Invalid flags likeX,FP, or multi-byte flags parse "successfully" withis_full = false. This corrupts downstream aggregation.Previously flagged but unaddressed.
🛠️ Validate flag strictly
let flags_start = sp2 + 1; - let (is_full, msg_offset) = if let Some(sp3) = find_byte(line, b' ', flags_start) { - (line[flags_start] == b'F', sp3 + 1) - } else { - (line.get(flags_start) == Some(&b'F'), line.len()) - }; + let flag = *line.get(flags_start)?; + let is_full = match flag { + b'F' => true, + b'P' => false, + _ => return None, // Invalid flag + }; + let msg_offset = match line.get(flags_start + 1) { + Some(b' ') => flags_start + 2, + None => line.len(), + _ => return None, // Multi-byte flag or missing space + };crates/logfwd-core/src/aggregator.rs (1)
20-31:⚠️ Potential issue | 🟡 MinorDoc comments still reference
Some/Noneinstead ofAggregateResult.The examples mention
NoneandSome("...")but the actual API returnsAggregateResult::PendingandAggregateResult::Complete. This was flagged previously and marked addressed, but the text remains inconsistent.📝 Suggested fix
/// Feed CRI message bytes with their P/F flag. The aggregator returns -/// `Some(message)` when a complete message is ready (on F lines), or -/// `None` for P lines (buffered internally). +/// `AggregateResult::Complete(message)` when a complete message is ready +/// (on F lines), or `AggregateResult::Pending` for P lines (buffered internally). /// /// # Lifecycle /// /// ```text -/// P "hello " → None (buffered) -/// P "world" → None (buffered) -/// F "!" → Some("hello world!") -/// F "simple" → Some("simple") // zero-copy fast path +/// P "hello " → Pending (buffered) +/// P "world" → Pending (buffered) +/// F "!" → Complete("hello world!") +/// F "simple" → Complete("simple") // zero-copy fast path /// ```🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@crates/logfwd-core/src/aggregator.rs` around lines 20 - 31, Doc comments in aggregator.rs still use Some/None in the examples but the API returns AggregateResult::Pending and AggregateResult::Complete; update the example lines inside the doc block (the P/F lifecycle example) to replace "None" with "Pending (buffered)" and "Some(\"...\")" with "Complete(\"...\")" (or explicitly `AggregateResult::Pending` / `AggregateResult::Complete` if you prefer fully-qualified names) so the examples match the actual return type used by the aggregator.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Outside diff comments:
In `@crates/logfwd-core/src/cri.rs`:
- Around line 98-112: CriReassembler::feed currently always copies F-only
messages into partial_buf (via partial_buf.extend_from_slice) because it accepts
cri: &CriLine<'_> whose message borrows an external, shorter lifetime, unlike
CriAggregator::feed which accepts message: &'a [u8] and can return a zero-copy
slice; pick one of two fixes: either document this design difference in the
CriReassembler::feed doc comment explaining why F-only lines are copied
(referencing CriReassembler::feed, partial_buf, max_line_size and contrast to
CriAggregator::feed), or change the API so CriReassembler::feed takes cri with a
matching output lifetime (e.g. accept the message as &'a [u8] or change the
CriLine lifetime parameter) allowing you to return &self-like slices and avoid
the copy (update signatures of callers accordingly); implement the approach that
matches call-site ownership (prefer documentation if callers cannot be
migrated).
---
Duplicate comments:
In `@crates/logfwd-core/src/aggregator.rs`:
- Around line 20-31: Doc comments in aggregator.rs still use Some/None in the
examples but the API returns AggregateResult::Pending and
AggregateResult::Complete; update the example lines inside the doc block (the
P/F lifecycle example) to replace "None" with "Pending (buffered)" and
"Some(\"...\")" with "Complete(\"...\")" (or explicitly
`AggregateResult::Pending` / `AggregateResult::Complete` if you prefer
fully-qualified names) so the examples match the actual return type used by the
aggregator.
In `@crates/logfwd-core/src/framer.rs`:
- Around line 132-137: CriFrameOutput currently uses a Vec<CriLineInfo> and
calls Vec::with_capacity() in frame(), which contradicts the "no heap
allocation" doc and causes per-frame heap work; change CriFrameOutput to store a
fixed-size array (e.g. [CriLineInfo; N]) plus a length counter (like FrameOutput
does) and make CriLineInfo derive/copy types (implement Copy/Clone if needed) so
you can push entries without allocation; update the frame() implementation to
write into the array and increment the count instead of creating a Vec, and
adjust any consumers that iterate lines to use the length counter and array
slice.
- Around line 91-105: The loop in framer that scans input bytes currently
continues advancing start after MAX_LINES_PER_FRAME is reached, causing
completed lines to be skipped; update the loop so that when input[i] == b'\n'
and output.count == MAX_LINES_PER_FRAME you stop processing further lines
(break/return) without advancing start so those remaining lines become the
remainder. Concretely: in the branch handling newline, check output.count
against MAX_LINES_PER_FRAME before assigning output.line_ranges/output.count and
before setting start; if full, set output.remainder_offset = start (or leave
start unchanged) and break/return immediately so no lines are silently dropped;
adjust use of output.line_ranges, output.count and output.remainder_offset
accordingly.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro
Run ID: fb43c1c4-fe01-4960-b255-42bc8a90d8c2
📒 Files selected for processing (6)
crates/logfwd-core/src/aggregator.rscrates/logfwd-core/src/byte_search.rscrates/logfwd-core/src/cri.rscrates/logfwd-core/src/framer.rscrates/logfwd-core/src/lib.rsdev-docs/PROOF_AUDIT.md
- Remove CriFramer/CriFrameOutput/CriLineInfo from framer.rs — CRI framing architecture deferred to #313 (unified SIMD benchmark). CRI parsing remains in cri.rs (parse_cri_line + CriReassembler). - Fix aggregator doc comments to reference AggregateResult instead of Option (addresses CodeRabbit review feedback). - Update PROOF_AUDIT.md: add content_correct oracle proof, close NewlineFramer correctness gap, bump proof count to 31. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Summary
Steps 4-5 of the zero-copy pipeline plan. Adds two new proven modules to logfwd-core:
framer.rs — Line boundary detection
Identifies line boundaries without copying data. Returns byte ranges into the input buffer.
NewlineFramer: splits on\nusing plain byte loops (no memchr)CriFramer: splits on\n, then extracts CRI timestamp/stream/flags/message rangesfind_byte: plain loop replacement for memchr, Kani-provableNo memchr, no Vec, no heap allocation in the hot path. LLVM auto-vectorizes the byte scan loops. Fixed-size
FrameOutput(4096 lines, 64KB stack).aggregator.rs — CRI P/F partial line merging
max_message_sizeenforced on ALL paths including F-onlyKani proofs (7 new, 22 total)
Why no memchr?
memchr uses SIMD internally but Kani can't model it (issue #632). By writing plain byte loops, we get:
Test plan
cargo clippy -- -D warningscleancargo fmt --checkcleanPart of #303 (zero-copy pipeline). Parent: #262.
🤖 Generated with Claude Code