fix: CodeRabbit feedback + formal verification (proptest, snapshots, bug fixes)#745
Merged
Merged
Conversation
…l and output sinks CodeRabbit fixes (PR #701 late feedback): - worker_pool: MAX_ATTEMPTS off-by-one (was 1+4=5, now 1+3=4 total) - worker_pool: assert! with descriptive message including actual value in new() - worker_pool: promote debug_assert!(submit after drain) to real guard with rejection ack - worker_pool: RetryAfter delay reset bug — reset to initial delay, not doubling - worker_pool: add num_rows + submitted_at to AckItem for real output latency metrics - worker_pool: log ack channel failures instead of silently discarding - pipeline: record_batch moved to apply_pool_ack with actual delivery latency - elasticsearch: zero-copy batch_buf move (std::mem::replace) — eliminates clone per send - elasticsearch: debug_assert!(ends_with) → runtime Err for release-mode safety - .coderabbit.yaml: move long instructions to docs/ci/*.md (fixes schema 10k char limit) - fanout/json_lines: move impl OutputSink blocks above #[cfg(test)] modules Formal verification additions: - loki: fix timestamp overflow — checked_add with truncation on u64::MAX - loki: proptest for minimal-displacement invariant (duplicate timestamps → consecutive) - loki: stream key encoding round-trip test with special chars (=, comma, quotes, backslash) - loki: unit test for overflow truncation behavior - elasticsearch: 4 insta snapshot tests (basic batch, nullable, special chars, all-null) - lib: 5 proptest tests for write_row_json (valid JSON, _int/_str/_float suffix stripping, no newlines) Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
WalkthroughThis PR moves extensive pre-merge rules out of Possibly related PRs
Caution Pre-merge checks failedPlease resolve all errors before merging. Addressing warnings is optional.
❌ Failed checks (3 errors, 2 warnings)
Comment |
There was a problem hiding this comment.
Actionable comments posted: 10
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In @.coderabbit.yaml:
- Around line 39-49: Update the condensed exemption list in the instructions
block so it matches the full guide by adding BTreeMap back into the heap-heavy
exemption; specifically edit the instructions string that currently lists
"Vec/HashMap" to read "Vec/HashMap/BTreeMap" (or equivalent) so the rule in
.coderabbit.yaml aligns with docs/ci/formal-verification-coverage.md and does
not false-block heap-heavy uses of BTreeMap.
- Around line 27-33: The condensed "error" check in .coderabbit.yaml removed the
typestate escape hatch—update the "Pass if" / error-mode summary text to
explicitly preserve the exception that permits enum+match state machines when
the PR documents why the typestate pattern is impractical; edit the rule (the
"Pass if" block / "error" summary) to add a clause like "except when the PR
provides justification that typestate is impractical" or "unless justified in PR
with documentation", so the checker still allows documented exceptions while
keeping the other constraints (no vague .unwrap/.expect, avoid clones in hot
loops, SAFETY comments, doc comments, etc.).
In `@crates/logfwd-output/src/elasticsearch.rs`:
- Around line 984-1086: The tests currently exercise ElasticsearchSink via
make_sync_sink but not the async path that ElasticsearchSinkFactory /
ElasticsearchAsyncSink now implements (which injects `@timestamp` and uses the new
"}\n" line ending); add corresponding snapshots that construct an async
serializer (use ElasticsearchSinkFactory::new or instantiate
ElasticsearchAsyncSink) and run the same RecordBatch through its serialization
path, capturing the output buffer and asserting snapshots (e.g., mirror
snapshot_basic_multi_type_batch, snapshot_nullable_columns,
snapshot_special_char_strings, snapshot_all_null_fields but using
ElasticsearchAsyncSink/ElasticsearchSinkFactory), or instead refactor the common
serialization logic into a shared function (e.g., serialize_record_batch or
similar) and call that from both the sync and async tests so the new behavior is
covered.
In `@crates/logfwd-output/src/loki.rs`:
- Around line 75-89: The helper sort_and_dedup_timestamps currently truncates
entries on overflow but doesn't inform callers, causing send_batch/do_send to
report the original batch size and silently lose rows; change
sort_and_dedup_timestamps to return a Result<usize, Error> or at minimum the
retained count (e.g., Ok(retained_count)) instead of void, and update callers
(send_batch and do_send) to read that return value and either (a) subtract the
dropped count from the metrics reported or (b) treat the overflow as a rejected
batch and error out so metrics reflect actual payload; adjust signatures and
callers accordingly (sort_and_dedup_timestamps, send_batch, do_send) to
propagate the count/error and ensure recorded success/row counts match the
actual entries sent.
In `@crates/logfwd/src/pipeline.rs`:
- Around line 611-619: The ack path is zeroing scan_ns/transform_ns so
successful batches no longer contribute to
scan_nanos_total/transform_nanos_total; fix by propagating the timing values
with the WorkItem or recording them in flush_batch() instead of dropping them:
add scan_ns and transform_ns fields to the WorkItem created in the pool.submit
call ( alongside submitted_at ), populate them from record_batch() where they
are currently owned, and then in flush_batch()/the ack path use those fields to
update scan_nanos_total and transform_nanos_total (alternatively, call the
existing record logic from flush_batch() to record timings there rather than
hard-coding zeros ). Ensure references to WorkItem, record_batch, flush_batch,
submitted_at, scan_ns, transform_ns, scan_nanos_total, and transform_nanos_total
are updated consistently.
In `@crates/logfwd/src/worker_pool.rs`:
- Around line 175-196: The submit() path only checks self.cancel.is_cancelled(),
so after a graceful drain callers can still spawn new workers; add a boolean
flag (e.g. self.drained) that drain() sets to true for the normal/graceful
shutdown path and update submit() to reject when either
self.cancel.is_cancelled() or self.drained is true, using the same ack/error
handling used now (keep the eprintln and AckItem rejection logic). Also ensure
drain() sets self.drained = true in the graceful-drain branch (not just the
forced-timeout branch) so post-drain submissions are consistently rejected.
- Around line 417-429: The worker completion path discards the result of
ack_tx.send(AckItem { ... }) so if the receiver is closed we lose tickets
silently; update the WorkerMsg::Work completion block to check the Result from
ack_tx.send(...) and log a warning (including worker id, tickets, success,
num_rows, and submitted_at or the AckItem) when send returns Err so we get
diagnostics but don’t panic. Locate the send call in the WorkerMsg::Work branch
(after process_item(...) in worker_pool.rs) and replace the discarded send with
a match or if let Err(e) = ack_tx.send(ack). { log::warn!(target: "...", "ack
send failed for worker {}: error={:?}, ack={:?}", id, e, ack); } ensuring the
code compiles with the existing logging facility.
In `@docs/ci/crate-boundary-and-dependency-integrity.md`:
- Around line 52-64: The doc currently states Rule 4 as an absolute invariant
but the codebase still contains violations (e.g., test panic!() calls in
crates/logfwd-core/src/aggregator.rs and production .unwrap()s in
crates/logfwd-core/src/structural.rs); update the text in
docs/ci/crate-boundary-and-dependency-integrity.md to clarify that the deny
rules for clippy::unwrap_used, clippy::panic, and clippy::indexing_slicing are
review-enforced for new changes (or that existing violations must be cleaned up
before converting to compiler denies), and explicitly state that PRs introducing
new .unwrap(), panic!(), or unchecked slice indexing in logfwd-core will be
flagged and must use .get(i)? or .get(i).ok_or(Error::OutOfBounds)? instead.
In `@docs/ci/documentation-thoroughly-updated.md`:
- Around line 153-164: The docs claim all reference files are versioned, which
is inaccurate; update the paragraph in the References section to reflect current
practice: state that only Arrow and DataFusion files (e.g., arrow-v54.md,
datafusion-v45.md) are versioned today while other guides
(tokio-async-patterns.md, opentelemetry-otlp.md, kani-verification.md,
notify-memchr-zstd.md) are maintained without versioned filenames; instruct
reviewers to consult dev-docs/references/ for the appropriate file (listing the
example files) and remove the requirement to rename non-versioned files on major
dependency bumps.
In `@docs/ci/formal-verification-coverage.md`:
- Around line 31-37: Update the coverage guideline to require explicit
state-machine invariant proofs for aggregator.rs: stateful protocol handling of
P/F flags and aggregation MUST include Kani proofs that invariants hold across
all transitions (e.g., add a proof like verify_aggregator_state_invariants that
exercises state transitions via kani::any() inputs over bounded traces/slices),
in addition to the existing no-panic (verify_<fn>_no_panic) and oracle guidance;
reference the aggregator.rs module and suggest using bounded trace length and
symbolic inputs to exhaustively check transition invariants for every state and
transition.
🪄 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: Repository YAML (base), Organization UI (inherited)
Review profile: ASSERTIVE
Plan: Pro
Run ID: 644c7bc0-24da-4493-aac4-b8584ce3a409
⛔ Files ignored due to path filters (5)
Cargo.lockis excluded by!**/*.lockcrates/logfwd-output/src/snapshots/logfwd_output__elasticsearch__snapshot_tests__all_null_fields.snapis excluded by!**/*.snapcrates/logfwd-output/src/snapshots/logfwd_output__elasticsearch__snapshot_tests__basic_multi_type.snapis excluded by!**/*.snapcrates/logfwd-output/src/snapshots/logfwd_output__elasticsearch__snapshot_tests__nullable_columns.snapis excluded by!**/*.snapcrates/logfwd-output/src/snapshots/logfwd_output__elasticsearch__snapshot_tests__special_char_strings.snapis excluded by!**/*.snap
📒 Files selected for processing (14)
.coderabbit.yamlcrates/logfwd-output/Cargo.tomlcrates/logfwd-output/src/elasticsearch.rscrates/logfwd-output/src/fanout.rscrates/logfwd-output/src/json_lines.rscrates/logfwd-output/src/lib.rscrates/logfwd-output/src/loki.rscrates/logfwd/src/pipeline.rscrates/logfwd/src/worker_pool.rsdocs/ci/crate-boundary-and-dependency-integrity.mddocs/ci/documentation-thoroughly-updated.mddocs/ci/formal-verification-coverage.mddocs/ci/high-quality-rust-practices.mddocs/ci/maintainer-fitness.md
Comment on lines
+984
to
+1086
| fn make_sync_sink() -> ElasticsearchSink { | ||
| ElasticsearchSink::new( | ||
| "test".to_string(), | ||
| "http://localhost:9200".to_string(), | ||
| "test-index".to_string(), | ||
| vec![], | ||
| make_stats(), | ||
| ) | ||
| } | ||
|
|
||
| /// Snapshot: basic multi-type batch (level_str, status_int, duration_float). | ||
| /// Regression guard: field name normalization + type serialization. | ||
| #[test] | ||
| fn snapshot_basic_multi_type_batch() { | ||
| let schema = Arc::new(Schema::new(vec![ | ||
| Field::new("level_str", DataType::Utf8, false), | ||
| Field::new("status_int", DataType::Int64, false), | ||
| Field::new("duration_ms_float", DataType::Float64, false), | ||
| ])); | ||
| let batch = RecordBatch::try_new( | ||
| schema, | ||
| vec![ | ||
| Arc::new(StringArray::from(vec!["ERROR", "INFO", "WARN"])), | ||
| Arc::new(Int64Array::from(vec![500i64, 200, 404])), | ||
| Arc::new(Float64Array::from(vec![125.5f64, 3.2, 87.0])), | ||
| ], | ||
| ) | ||
| .unwrap(); | ||
|
|
||
| let mut sink = make_sync_sink(); | ||
| sink.serialize_batch(&batch).unwrap(); | ||
| let output = String::from_utf8(sink.batch_buf.clone()).unwrap(); | ||
| insta::assert_snapshot!("basic_multi_type", output); | ||
| } | ||
|
|
||
| /// Snapshot: nullable columns — null values must appear as JSON null. | ||
| #[test] | ||
| fn snapshot_nullable_columns() { | ||
| let schema = Arc::new(Schema::new(vec![ | ||
| Field::new("msg_str", DataType::Utf8, true), | ||
| Field::new("code_int", DataType::Int64, true), | ||
| ])); | ||
| let batch = RecordBatch::try_new( | ||
| schema, | ||
| vec![ | ||
| Arc::new(StringArray::from(vec![Some("hello"), None, Some("world")])), | ||
| Arc::new(Int64Array::from(vec![Some(1i64), Some(2), None])), | ||
| ], | ||
| ) | ||
| .unwrap(); | ||
|
|
||
| let mut sink = make_sync_sink(); | ||
| sink.serialize_batch(&batch).unwrap(); | ||
| let output = String::from_utf8(sink.batch_buf.clone()).unwrap(); | ||
| insta::assert_snapshot!("nullable_columns", output); | ||
| } | ||
|
|
||
| /// Snapshot: strings with special JSON characters (quotes, backslash, newlines). | ||
| /// Regression guard: escape_json correctness. | ||
| #[test] | ||
| fn snapshot_special_char_strings() { | ||
| let schema = Arc::new(Schema::new(vec![Field::new( | ||
| "msg_str", | ||
| DataType::Utf8, | ||
| false, | ||
| )])); | ||
| let batch = RecordBatch::try_new( | ||
| schema, | ||
| vec![Arc::new(StringArray::from(vec![ | ||
| r#"say "hello" world"#, | ||
| "line1\nline2\ttab", | ||
| r"back\slash", | ||
| "control\x00char", | ||
| ]))], | ||
| ) | ||
| .unwrap(); | ||
|
|
||
| let mut sink = make_sync_sink(); | ||
| sink.serialize_batch(&batch).unwrap(); | ||
| let output = String::from_utf8(sink.batch_buf.clone()).unwrap(); | ||
| insta::assert_snapshot!("special_char_strings", output); | ||
| } | ||
|
|
||
| /// Snapshot: single row with all-null nullable fields produces valid output. | ||
| #[test] | ||
| fn snapshot_all_null_fields() { | ||
| let schema = Arc::new(Schema::new(vec![ | ||
| Field::new("msg_str", DataType::Utf8, true), | ||
| Field::new("code_int", DataType::Int64, true), | ||
| ])); | ||
| let batch = RecordBatch::try_new( | ||
| schema, | ||
| vec![ | ||
| Arc::new(StringArray::from(vec![None as Option<&str>])), | ||
| Arc::new(Int64Array::from(vec![None as Option<i64>])), | ||
| ], | ||
| ) | ||
| .unwrap(); | ||
|
|
||
| let mut sink = make_sync_sink(); | ||
| sink.serialize_batch(&batch).unwrap(); | ||
| let output = String::from_utf8(sink.batch_buf.clone()).unwrap(); | ||
| insta::assert_snapshot!("all_null_fields", output); |
There was a problem hiding this comment.
🧹 Nitpick | 🔵 Trivial
Snapshot the async Elasticsearch serializer too.
All four new tests build ElasticsearchSink, but the default single-output Elasticsearch path goes through ElasticsearchSinkFactory/ElasticsearchAsyncSink. That async serializer is the code changed above, and it adds behavior these snapshots never exercise (@timestamp injection plus the new }\n invariant). Either snapshot the async sink as well, or extract one shared serializer and test that directly.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@crates/logfwd-output/src/elasticsearch.rs` around lines 984 - 1086, The tests
currently exercise ElasticsearchSink via make_sync_sink but not the async path
that ElasticsearchSinkFactory / ElasticsearchAsyncSink now implements (which
injects `@timestamp` and uses the new "}\n" line ending); add corresponding
snapshots that construct an async serializer (use ElasticsearchSinkFactory::new
or instantiate ElasticsearchAsyncSink) and run the same RecordBatch through its
serialization path, capturing the output buffer and asserting snapshots (e.g.,
mirror snapshot_basic_multi_type_batch, snapshot_nullable_columns,
snapshot_special_char_strings, snapshot_all_null_fields but using
ElasticsearchAsyncSink/ElasticsearchSinkFactory), or instead refactor the common
serialization logic into a shared function (e.g., serialize_record_batch or
similar) and call that from both the sync and async tests so the new behavior is
covered.
Comment on lines
+175
to
+196
| if self.cancel.is_cancelled() { | ||
| // Pool has been drained — reject the item immediately rather than | ||
| // silently losing it. This keeps the at-least-once invariant intact | ||
| // even for callers that mistakenly submit after drain. | ||
| eprintln!("worker_pool: submit after drain — rejecting batch immediately"); | ||
| let ticket_count = item.tickets.len(); | ||
| if self | ||
| .ack_tx | ||
| .send(AckItem { | ||
| tickets: item.tickets, | ||
| success: false, | ||
| num_rows: item.num_rows, | ||
| submitted_at: item.submitted_at, | ||
| }) | ||
| .is_err() | ||
| { | ||
| eprintln!( | ||
| "worker_pool: ack channel closed, batch lost permanently (ticket_count={ticket_count})" | ||
| ); | ||
| } | ||
| return; | ||
| } |
There was a problem hiding this comment.
Submit-after-drain rejection still misses the clean-drain case.
submit() only checks self.cancel.is_cancelled(), but drain() sets that token only on the timeout/forced-shutdown path. After a graceful drain, a later submit() can spawn fresh workers again instead of rejecting the batch, so the new runtime guard does not actually cover normal post-drain use.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@crates/logfwd/src/worker_pool.rs` around lines 175 - 196, The submit() path
only checks self.cancel.is_cancelled(), so after a graceful drain callers can
still spawn new workers; add a boolean flag (e.g. self.drained) that drain()
sets to true for the normal/graceful shutdown path and update submit() to reject
when either self.cancel.is_cancelled() or self.drained is true, using the same
ack/error handling used now (keep the eprintln and AckItem rejection logic).
Also ensure drain() sets self.drained = true in the graceful-drain branch (not
just the forced-timeout branch) so post-drain submissions are consistently
rejected.
Code fixes:
- loki.rs: sort_and_dedup_timestamps returns retained count; serialize_loki_json
returns (payload, retained_rows) and send_batch uses retained count for metrics
so overflow-truncated entries aren't reported as delivered
- worker_pool.rs: add scan_ns/transform_ns to WorkItem/AckItem so timing metrics
are preserved through the async ack path; log warning when ack_tx.send fails
instead of silently discarding with let _ =
- pipeline.rs: populate scan_ns/transform_ns in WorkItem; use ack.scan_ns and
ack.transform_ns in apply_pool_ack instead of hardcoded zeros
Doc fixes:
- .coderabbit.yaml: add BTreeMap to heap-heavy exemption; add typestate escape
hatch note ("except when PR justifies why typestate is impractical")
- docs/ci/formal-verification-coverage.md: add aggregator.rs state-machine
invariant proof requirement (verify_aggregator_state_invariants)
- docs/ci/documentation-thoroughly-updated.md: clarify which reference files are
versioned (arrow, datafusion) vs unversioned (tokio, otlp, kani)
- docs/ci/crate-boundary-and-dependency-integrity.md: clarify Rule 4 applies to
new changes only, not existing violations
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
There was a problem hiding this comment.
Actionable comments posted: 1
♻️ Duplicate comments (1)
crates/logfwd/src/worker_pool.rs (1)
183-206:⚠️ Potential issue | 🟠 MajorSubmit-after-drain still allows new workers after graceful shutdown.
The check at line 183 only guards against the forced-cancel path. After
drain()completes gracefully (no timeout),self.cancel.is_cancelled()returns false, and subsequentsubmit()calls will spawn fresh workers via lines 234-240.A
self.drained: boolflag set at the end ofdrain()(both branches) would close this gap.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@crates/logfwd/src/worker_pool.rs` around lines 183 - 206, The submit-after-drain bug: add a self.drained: bool field to WorkerPool, set it true at the end of drain() in both the timeout and graceful-complete branches, and have submit() check this flag in addition to self.cancel.is_cancelled() (e.g. if self.drained { reject/ack as before }) to prevent spawn of new workers after a successful graceful drain; ensure access to self.drained is synchronized the same way other state is (or updated while holding the existing lock) and reuse the same rejection/ack logic used when cancel.is_cancelled() is true.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@docs/ci/crate-boundary-and-dependency-integrity.md`:
- Around line 20-22: The fenced code block showing the cargo build command lacks
a language identifier; update the markdown in
docs/ci/crate-boundary-and-dependency-integrity.md by adding a language tag
(e.g., "shell" or "bash") to the opening triple-backtick for the block
containing "cargo build -p logfwd-core --target thumbv6m-none-eabi" so the block
becomes a fenced code block with an explicit language identifier.
---
Duplicate comments:
In `@crates/logfwd/src/worker_pool.rs`:
- Around line 183-206: The submit-after-drain bug: add a self.drained: bool
field to WorkerPool, set it true at the end of drain() in both the timeout and
graceful-complete branches, and have submit() check this flag in addition to
self.cancel.is_cancelled() (e.g. if self.drained { reject/ack as before }) to
prevent spawn of new workers after a successful graceful drain; ensure access to
self.drained is synchronized the same way other state is (or updated while
holding the existing lock) and reuse the same rejection/ack logic used when
cancel.is_cancelled() is true.
🪄 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: Repository YAML (base), Organization UI (inherited)
Review profile: ASSERTIVE
Plan: Pro
Run ID: 6283629f-0c27-4eff-a8c9-01b1aa37c088
📒 Files selected for processing (7)
.coderabbit.yamlcrates/logfwd-output/src/loki.rscrates/logfwd/src/pipeline.rscrates/logfwd/src/worker_pool.rsdocs/ci/crate-boundary-and-dependency-integrity.mddocs/ci/documentation-thoroughly-updated.mddocs/ci/formal-verification-coverage.md
Comment on lines
+20
to
+22
| ``` | ||
| cargo build -p logfwd-core --target thumbv6m-none-eabi | ||
| ``` |
There was a problem hiding this comment.
🧹 Nitpick | 🔵 Trivial
Add language identifier to fenced code block.
Static analysis flags MD040: fenced code blocks should have a language specified.
Suggested fix
-```
+```shell
cargo build -p logfwd-core --target thumbv6m-none-eabi</details>
<!-- suggestion_start -->
<details>
<summary>📝 Committable suggestion</summary>
> ‼️ **IMPORTANT**
> Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
```suggestion
🧰 Tools
🪛 markdownlint-cli2 (0.22.0)
[warning] 20-20: Fenced code blocks should have a language specified
(MD040, fenced-code-language)
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@docs/ci/crate-boundary-and-dependency-integrity.md` around lines 20 - 22, The
fenced code block showing the cargo build command lacks a language identifier;
update the markdown in docs/ci/crate-boundary-and-dependency-integrity.md by
adding a language tag (e.g., "shell" or "bash") to the opening triple-backtick
for the block containing "cargo build -p logfwd-core --target
thumbv6m-none-eabi" so the block becomes a fenced code block with an explicit
language identifier.
This was referenced Apr 3, 2026
Merged
This was referenced Apr 5, 2026
Closed
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
1 participant
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.
Summary
Addresses all late CodeRabbit feedback on PR #701 and adds a comprehensive formal verification layer discovered through research agent analysis.
CodeRabbit Fixes (PR #701 late feedback)
assert!now includes the actualmax_workersvalue in the messagedebug_assert!(submit after drain)promoted to real runtime guard that rejects the batch via AckItem (preserves at-least-once invariant in release builds)num_rows+submitted_atadded to AckItem — output latency metrics now reflect actual delivery time, not near-zero submission timerecord_batch()moved toapply_pool_ack()so metrics use real delivery latencyself.batch_buf.clone()→std::mem::replace— eliminates per-send buffer copy (0→64KB allocation per batch avoided)debug_assert!(ends_with b"}\\n")→ runtimeErr— prevents silent usize underflow/truncate corruption in release buildsinstructionsfields moved todocs/ci/*.md(fixes 10,000-char schema limit error)impl OutputSinkblocks moved above#[cfg(test)]modules (convention compliance)Formal Verification Additions
Loki:
sort_and_dedup_timestamps—checked_addwith graceful truncation at u64::MAX instead of wrapping to 0[base, base+1, ..., base+N-1]=,,,",\) that were lossy in the oldk=v,...formatElasticsearch:
src/snapshots/— CI fails if serialization format changesOutput lib (
write_row_json):_int/_str/_floatsuffix stripping + round-trip, no internal newlines in NDJSON bulk formatTest plan
cargo test -p logfwd-output -p logfwd— all pass (116 logfwd-output tests, 51 logfwd tests)cargo fmt— no changescargo clippy— no warningscargo insta accept🤖 Generated with Claude Code