[strawgate]

Summary

• OutputWorkerPool (crates/logfwd/src/worker_pool.rs): Dynamic async worker pool with MRU-first work consolidation — keeps work on fewest active workers so idle ones self-terminate and close HTTP connections. Scales up to max_workers under burst, back-pressures via send().await at capacity. 6 Kani proofs verify dispatch invariants.
• Sink trait (crates/logfwd-output/src/sink.rs): Rewritten to use Pin<Box<dyn Future>> for dyn-compatibility (enables Box<dyn Sink>). Added SinkFactory (one per output type, holds shared Arc<reqwest::Client>) and SyncSinkAdapter (wraps legacy OutputSink with block_in_place).
• ElasticsearchAsyncSink + ElasticsearchSinkFactory: reqwest-based, gzip compression, @timestamp injection, RetryAfter on 429 (parses Retry-After header), errors:true partial-failure detection. Old ureq ElasticsearchSink kept for backward compat during migration.
• LokiAsyncSink + LokiSinkFactory: reqwest-based, per-stream timestamp sort+dedup (Loki monotonic ordering requirement), label extraction from config and record columns, tenant ID support. Proptest covers the ordering invariant.

Architecture

Pipeline → OutputWorkerPool.submit(WorkItem)
                │
                ▼  MRU-first dispatch
         ┌──────────────┐
         │  Worker (1)  │ ← owns Box<dyn Sink> (ElasticsearchAsyncSink)
         │  Worker (2)  │   created by Arc<dyn SinkFactory>
         │  ...idle...  │ → hits idle_timeout → closes HTTP conn
         └──────────────┘
                │ AckItem channel
                ▼
         Pipeline advances file checkpoints

What's next (follow-up tasks)

• Task 6: Wire OutputWorkerPool into pipeline.rs (replace block_in_place(output.send_batch(...)))
• Issues Convert OtlpSink to async reqwest (migrate from ureq) #689–693: Async migration for OTLP, TCP, UDP, Stdout, JsonLines (assigned to Copilot)

Test plan

• cargo test -p logfwd -p logfwd-output — all pass
• cargo clippy -p logfwd -p logfwd-output -- -D warnings — clean
• cargo fmt — applied
• cargo kani -p logfwd — run after review (Kani takes ~5 min)
• Wire pool into pipeline.rs and run integration tests

🤖 Generated with Claude Code

[coderabbitai]

Warning

.coderabbit.yaml has a parsing error

The CodeRabbit configuration file in this repository has a parsing error and default settings were used instead. Please fix the error(s) in the configuration file. You can initialize chat with CodeRabbit to get help with the configuration file.

💥 Parsing errors (1)

Validation error: Instructions must be under 10000 characters at "reviews.pre_merge_checks.custom_checks[1].instructions"

⚙️ Configuration instructions

• Please see the configuration documentation for more information.
• You can also validate your configuration using the online YAML validator.
• If your editor has YAML language server enabled, you can add the path at the top of this file to enable auto-completion and validation: # yaml-language-server: $schema=https://coderabbit.ai/integrations/schema.v2.json

Walkthrough

Introduces async output handling through a new OutputWorkerPool that manages worker tasks for batch submission with retry logic, adds dynamic trait-object compatibility to the Sink trait via boxed futures, and implements HTTP-based async sinks for Elasticsearch bulk indexing and Loki log push. The pipeline is refactored to route output through the worker pool instead of direct sink submission. Dependencies are added for async runtime (tokio), HTTP client (reqwest), and compression (flate2). A .coderabbit.yaml configuration is added with custom checks for Rust practices, formal verification coverage, documentation synchronization, and crate boundary integrity. Comprehensive test coverage is added for fanout error handling, JSON serialization edge cases, and worker pool lifecycle.

Possibly related PRs

• feat: Async pipeline — select! loop, input threads, bounded channels (#204) #221: Modifies the async pipeline (Pipeline::run_async) and async/sync sink trait signatures that this PR refactors into a worker pool and object-safe factory pattern.

• fix: Pipeline continues on all errors instead of halting #208: Modifies crates/logfwd/src/pipeline.rs output error handling semantics to continue-on-error; this PR implements pool-based output error routing as an alternative approach.

• refactor: Rust best practices — workspace lints, resolver 3, code fixes #567: Modifies the output/sink surface in crates/logfwd-output (SendResult enum and sink-related APIs) that this PR extends with factory and adapter types.

---

Caution

Pre-merge checks failed

Please resolve all errors before merging. Addressing warnings is optional.

• Ignore

❌ Failed checks (3 errors, 2 warnings)

Check name	Status	Explanation	Resolution
High-Quality Rust Practices	❌ Error	Public API function OutputWorkerPool::new() panics on invalid input instead of returning Result type.	Change OutputWorkerPool::new() to return io::Result and replace assert! with descriptive io::Error.
Formal Verification Coverage	❌ Error	PR fails formal verification coverage: sort_and_dedup_timestamps lacks required Kani proof, OutputWorkerPool::new uses assert!() instead of Result, VERIFICATION.md not updated.	Add Kani proof for sort_and_dedup_timestamps, change OutputWorkerPool::new to return Result, update VERIFICATION.md with dispatch_step proofs.
Crate Boundary And Dependency Integrity	❌ Error	OutputWorkerPool::new() uses assert!() for invalid input instead of returning Result, violating the public API rule requiring Result/Option instead of panics.	Change OutputWorkerPool::new() to return io::Result and return Err for max_workers == 0 instead of asserting.
Documentation Thoroughly Updated	⚠️ Warning	PR introduces major architectural changes (OutputWorkerPool, MRU dispatch, async Sink redesign, SinkFactory pattern) without updating any documentation files.	Update ARCHITECTURE.md section 7, add ADR to DESIGN.md, add worker_pool.rs to VERIFICATION.md, document new dependencies in CRATE_RULES.md, add struct field doc comments, mark Phase 5 tasks complete in PHASES.md.
Maintainer Fitness	⚠️ Warning	PR has three unresolved correctness issues: MAX_ATTEMPTS off-by-one bug, OutputWorkerPool::new() using assert! instead of Result, and false panic-handling claim in docstring.	Fix MAX_ATTEMPTS logic (attempts=1, >=→==), return io::Result from OutputWorkerPool::new(), implement steady-state panic detection or remove false docstring claim, update PHASES.md, separate .coderabbit.yaml.

---

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[coderabbitai]

Actionable comments posted: 18

🤖 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 24-186: The inline custom-check bodies at
reviews.pre_merge_checks.custom_checks[*].instructions are too long and risk
exceeding the 10,000 character schema limit; move each long rulebook into a repo
docs/knowledge-base file (e.g., docs/ci/custom-checks/*.md) and replace the
corresponding instructions value with a short pass/fail rubric (<=10k chars)
that references the external doc; update the YAML entries under
reviews.pre_merge_checks.custom_checks (the specific array items that currently
contain the large Rust guidance) to point to the external doc by name and
include only the concise acceptance criteria and a link to the full guidance.

In `@crates/logfwd-output/src/elasticsearch.rs`:
- Around line 383-388: The timestamp-detection currently inspects raw Arrow
field names (has_timestamp_col) which misses columns that will be serialized by
write_row_json() under normalized JSON names (e.g., "@timestamp" / "_timestamp"
from suffixed fields), causing duplicate/overwritten timestamp keys; update the
check to examine the normalized output names produced for serialization instead
of raw field names — e.g., use the build_col_infos(batch) result (cols) and
detect if any column's computed JSON name equals "@timestamp" or "_timestamp"
(the same names write_row_json() emits) before injecting a generated timestamp;
apply the same change to the other identical block referenced (lines ~397-416)
so both branches use cols-based JSON-name detection.
- Around line 332-333: The code currently stores client: Arc<reqwest::Client> on
ElasticsearchSinkFactory and shares it across workers, which contradicts
SinkFactory's intent of per-worker clients; change the design so
ElasticsearchSinkFactory does NOT hold Arc<reqwest::Client> but instead
constructs a new reqwest::Client for each worker and stores that client on each
ElasticsearchAsyncSink instance (move the client field from the factory into
ElasticsearchAsyncSink), remove Arc wrapping (or keep only config/ClientBuilder
on the factory), and create the per-worker client inside the factory's
create/new_sink method so idle worker shutdown drops its own client and
associated pooled connections (update constructors and any references to use
self.client on ElasticsearchAsyncSink).

In `@crates/logfwd-output/src/lib.rs`:
- Around line 516-523: The call to LokiSinkFactory::new currently hardcodes
tenant_id, static_labels, and label_columns (passing None and empty Vecs)
because OutputConfig does not expose those fields; update the config type (the
struct in logfwd-config) to include tenant_id: Option<String>, static_labels:
Vec<(String,String)> (or HashMap), and label_columns: Vec<String>, deserialize
them from config, then wire those new fields into the LokiSinkFactory::new
invocation instead of None/Vec::new so multi-tenant headers and label-based
stream partitioning are configurable; modify the factory call site that
constructs LokiSinkFactory::new and the configuration struct/parser to propagate
these three values.
- Around line 452-460: The OnceFactory.create implementation permanently takes
the inner sink, so after a worker exits and spawn_worker runs again it finds
None and sync outputs stop; change OnceFactory to either (a) retain enough
configuration or a boxed factory/closure so create() can rebuild a fresh sink on
each call, or (b) stop consuming the original sink and instead wrap the sink in
a shared handle (e.g. Arc<Mutex<...>>) so SyncSinkAdapter::new can be called
repeatedly; update the OnceFactory struct to store the rebuild config/factory or
an Arc-wrapped sink and modify create() accordingly so subsequent spawn_worker
invocations always get a valid sink rather than None.
- Around line 453-455: The create() method uses self.inner.lock().unwrap(),
which panics on a poisoned mutex; change the lock acquisition to handle
poisoning and return an io::Error instead (e.g., replace unwrap() with
lock().map_err(|e| io::Error::new(io::ErrorKind::Other, format!("mutex poisoned:
{}", e)))? or use e.into_inner() to recover) so the function returns a proper
io::Result rather than panicking; ensure subsequent code still calls
guard.take() on the recovered guard variable.

In `@crates/logfwd-output/src/loki.rs`:
- Around line 189-194: The current stream_key built by mapping labels to
format!("{k}={v}") and joining with commas is lossy because label values can
contain commas; replace this with an unambiguous representation (e.g., use the
sorted Vec<(String,String)> itself as the map key or serialize the sorted labels
with a lossless encoder like serde_json::to_string(&labels)) when building
stream_key in the stream_key creation (variable stream_key that uses
labels.iter().map(...).collect::<Vec<_>>().join(",")); make the same change in
the other occurrence referenced (lines around 221-227) so keys are reversible
and distinct streams cannot collide.
- Around line 68-77: The timestamp extraction logic needs a null check and
overflow-safe increment: when reading timestamps from the nullable Arrow column
(the code path that mirrors label handling at line 174), call .is_null(row)
before .value(row) and if null fall back to observed_time_ns; also replace any
unchecked `+ 1` in sort_and_dedup_timestamps with a checked/saturating increment
(e.g., use checked_add(1).unwrap_or(u64::MAX) or saturating_add) so timestamps
do not wrap on u64::MAX. Update the extraction and the sort_and_dedup_timestamps
adjustment to use these checks.

In `@crates/logfwd-output/src/sink.rs`:
- Around line 53-57: The trait method send_batch currently returns a boxed
future per batch which allocates on the hot send path; change the trait to avoid
boxing by introducing an associated generic future type (e.g., add type
SendFuture<'a>: Future<Output = io::Result<SendResult>> + Send + 'a and change
fn send_batch(...) -> Self::SendFuture<'a>) or otherwise return an impl Future
from concrete implementations, then update all implementors of send_batch (and
any callers) to return that concrete future (or use async fn in impls) and run
benchmarks for the send path to verify there is no allocation/regression;
reference send_batch, RecordBatch, BatchMetadata, and SendResult when updating
signatures and implementations.
- Around line 129-130: The shutdown implementation in Sink::shutdown currently
returns Ok(()) and never flushes the wrapped sink, causing buffered legacy sinks
to skip final flush; change the shutdown method in sink.rs to call and await the
inner sink's shutdown/flush so the wrapped sink is flushed on graceful exit (use
the adapter's inner field and forward its shutdown future or call its
synchronous flush and propagate any io::Result error), and ensure the returned
Future resolves to that result instead of unconditionally Ok(()), preserving
Send and lifetime bounds expected by OutputWorkerPool.

In `@crates/logfwd/src/pipeline.rs`:
- Around line 181-213: The code wraps FanOut in OnceFactory but still constructs
OutputWorkerPool with max_workers=4, which conflicts because
OnceFactory::create() only succeeds once; either use a factory that supports
multiple creates (replace OnceFactory::new(Box::new(FanOut::new(...))) with a
multi-create factory for FanOut) or constrain the pool to a single worker when
using OnceFactory (set OutputWorkerPool::new(..., 1, ...) for the branch that
builds OnceFactory). Update the branch that builds
Arc::new(OnceFactory::new(...)) and the subsequent OutputWorkerPool
instantiation (referencing OnceFactory, FanOut, and OutputWorkerPool) so the
chosen factory and max_workers are compatible.

In `@crates/logfwd/src/worker_pool.rs`:
- Around line 260-263: The forced shutdown currently calls self.cancel.cancel()
then self.join_set.shutdown(), which aborts worker tasks mid-flight and prevents
them from sending final AckItem for in-flight WorkItem; instead, change the
shutdown sequence in worker_pool.rs to first close/drain the work channel and
await workers to exit naturally (have receivers return None or use a
timeout-based wait), only after the drain/timeout call self.cancel.cancel() and
then join_set.shutdown() as a fallback; alternatively (or in addition) track
in-flight WorkItem identifiers externally (e.g., an in_flight set updated by the
worker start/finish paths) and mark those as failed before aborting so AckItem
semantics remain consistent.
- Around line 236-240: The shutdown loop currently awaits
handle.tx.send(WorkerMsg::Shutdown), which can block if a worker's bounded
channel is full; change it to use non-blocking try_send(WorkerMsg::Shutdown)
(like submit() uses) and ignore or log the Err(e) result as appropriate so
shutdown signaling never awaits channel capacity. Specifically, locate the
shutdown code that iterates over workers and replace handle.tx.send(...).await
with handle.tx.try_send(WorkerMsg::Shutdown), handling the Result (e.g., map_err
or match) but do not await; keep dropping workers afterwards to close channels.
- Around line 191-208: When dispatching WorkItem ensure no path silently drops
it: update the logic around spawn_worker/workers and sends so failures always
emit a failure ack or return the WorkItem to caller. Specifically, in the code
around spawn_worker(), if spawn_worker() returns Err and workers is empty, call
the WorkItem failure/ack path (or return an Err) instead of falling through;
when awaiting a cloned sender (front.tx.clone(); tx.send(msg).await) handle the
Result and emit a failure ack on Err; replace blocking send(Shutdown).await on
bounded channels with a non-blocking/timeout variant (e.g., try_send or
tokio::time::timeout using graceful_timeout) so shutdown won’t hang; and when
cancel.cancelled() triggers, before breaking the select! loop drain the internal
receiver/queue and ack or fail each buffered WorkItem (so
join_set.shutdown().await won’t leave unacked items). Ensure you reference and
update behaviors in spawn_worker, workers, tx.send, Shutdown sending,
cancel.cancelled, join_set.shutdown, and the WorkItem ack/failure code paths
(and consider OnceFactory behavior) so every dispatch either reaches a worker
channel or produces an explicit failure ack.

In `@dev-docs/bug-hunt-2026-04-02.md`:
- Line 230: Replace the phrase "by accident" in the sentence referencing `type:
null` (the snippet "type: null works in the simple layout (`output: {type:
null}`) by accident, creating a user expectation that breaks in the advanced
list context") with a shorter alternative such as "accidentally" or
"inadvertently" to tighten phrasing and improve readability while keeping the
rest of the sentence and the references to `type: null`, `output: {type: null}`,
"simple layout", and "advanced list context" unchanged.
- Line 90: There are unlabeled fenced code blocks (``` ... ```) in the Markdown
that cause MD040 lint failures; update each triple-backtick block (the ones
currently with no language token) by adding an explicit language label (e.g.,
text) so the fences become ```text (also apply the same change to the other
occurrences noted), ensuring all unlabeled fenced blocks in the document are
labeled.
- Line 11: MD031 violations: fenced code blocks lack surrounding blank lines;
locate each fenced block marker like the triple-backtick token (e.g., ```sh) in
dev-docs/bug-hunt-2026-04-02.md and ensure there is at least one blank line
before and after every fenced code block. Edit occurrences that the reviewer
listed (including the shown ```sh and the other fences at the line ranges called
out) and insert a blank line above and below each fence so markdownlint MD031 is
satisfied and rendering remains consistent.

🪄 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: 054bba76-2f67-4156-918c-d7f60b88669f

📥 Commits

Reviewing files that changed from the base of the PR and between 085c2cf and 7d65fb9.

⛔ Files ignored due to path filters (1)

• Cargo.lock is excluded by !**/*.lock

📒 Files selected for processing (13)

• .coderabbit.yaml
• crates/logfwd-output/Cargo.toml
• crates/logfwd-output/src/elasticsearch.rs
• crates/logfwd-output/src/lib.rs
• crates/logfwd-output/src/loki.rs
• crates/logfwd-output/src/sink.rs
• crates/logfwd-output/tests/elasticsearch_arrow_ipc.rs
• crates/logfwd-output/tests/elasticsearch_integration.rs
• crates/logfwd/Cargo.toml
• crates/logfwd/src/lib.rs
• crates/logfwd/src/pipeline.rs
• crates/logfwd/src/worker_pool.rs
• dev-docs/bug-hunt-2026-04-02.md

[coderabbitai]

⚠️ Potential issue | 🟠 Major

Keep each custom-check rulebook under the schema limit.

reviews.pre_merge_checks.custom_checks[*].instructions is capped at 10,000 characters, and several of these inline blocks are long enough to exceed that. Once the file stops validating against the schema, the checks become unreliable or stop loading entirely. Move the long guidance into repo docs/knowledge-base files and keep each inline check to a short pass/fail rubric.

Also applies to: 187-340, 341-495, 496-646, 647-807

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In @.coderabbit.yaml around lines 24 - 186, The inline custom-check bodies at
reviews.pre_merge_checks.custom_checks[*].instructions are too long and risk
exceeding the 10,000 character schema limit; move each long rulebook into a repo
docs/knowledge-base file (e.g., docs/ci/custom-checks/*.md) and replace the
corresponding instructions value with a short pass/fail rubric (<=10k chars)
that references the external doc; update the YAML entries under
reviews.pre_merge_checks.custom_checks (the specific array items that currently
contain the large Rust guidance) to point to the external doc by name and
include only the concise acceptance criteria and a link to the full guidance.

[coderabbitai]

⚠️ Potential issue | 🔴 Critical

OnceFactory cannot survive worker respawn.

create() permanently consumes the only sync sink. After that worker self-terminates on idle, the next spawn_worker() hits None, so sync outputs stop recovering after the first idle cycle. Either keep OnceFactory-backed workers alive or store enough config to rebuild the sink.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@crates/logfwd-output/src/lib.rs` around lines 452 - 460, The
OnceFactory.create implementation permanently takes the inner sink, so after a
worker exits and spawn_worker runs again it finds None and sync outputs stop;
change OnceFactory to either (a) retain enough configuration or a boxed
factory/closure so create() can rebuild a fresh sink on each call, or (b) stop
consuming the original sink and instead wrap the sink in a shared handle (e.g.
Arc<Mutex<...>>) so SyncSinkAdapter::new can be called repeatedly; update the
OnceFactory struct to store the rebuild config/factory or an Arc-wrapped sink
and modify create() accordingly so subsequent spawn_worker invocations always
get a valid sink rather than None.

[coderabbitai]

⚠️ Potential issue | 🟡 Minor

Add blank lines around fenced code blocks to satisfy markdownlint (MD031).

Multiple fences are missing required surrounding blank lines; this will keep docs lint red and can affect renderer consistency.

Also applies to: 27-27, 42-42, 54-54, 66-66, 67-67, 80-80, 82-82, 83-83, 90-90, 102-102, 104-104, 106-106, 124-124, 126-126, 144-144, 164-164, 176-176, 188-188, 189-189, 194-194, 212-212, 220-220, 221-221, 241-241, 256-256, 264-264, 274-274, 276-276, 277-277, 295-295

🧰 Tools

🪛 markdownlint-cli2 (0.22.0)

[warning] 11-11: Fenced code blocks should be surrounded by blank lines

(MD031, blanks-around-fences)

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@dev-docs/bug-hunt-2026-04-02.md` at line 11, MD031 violations: fenced code
blocks lack surrounding blank lines; locate each fenced block marker like the
triple-backtick token (e.g., ```sh) in dev-docs/bug-hunt-2026-04-02.md and
ensure there is at least one blank line before and after every fenced code
block. Edit occurrences that the reviewer listed (including the shown ```sh and
the other fences at the line ranges called out) and insert a blank line above
and below each fence so markdownlint MD031 is satisfied and rendering remains
consistent.

[coderabbitai]

⚠️ Potential issue | 🟡 Minor

Specify a language for unlabeled fenced blocks (MD040).

Please label these blocks (e.g., text) so syntax handling is explicit and lint passes.

Also applies to: 194-194, 221-221, 277-277

🧰 Tools

🪛 markdownlint-cli2 (0.22.0)

[warning] 90-90: Fenced code blocks should be surrounded by blank lines

(MD031, blanks-around-fences)

---

[warning] 90-90: 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 `@dev-docs/bug-hunt-2026-04-02.md` at line 90, There are unlabeled fenced code
blocks (``` ... ```) in the Markdown that cause MD040 lint failures; update each
triple-backtick block (the ones currently with no language token) by adding an
explicit language label (e.g., text) so the fences become ```text (also apply
the same change to the other occurrences noted), ensuring all unlabeled fenced
blocks in the document are labeled.

[coderabbitai]

🧹 Nitpick | 🔵 Trivial

Tighten phrasing for readability.

“by accident” is slightly wordy in this sentence; a shorter alternative improves scanability.

🧰 Tools

🪛 LanguageTool

[style] ~230-~230: ‘by accident’ might be wordy. Consider a shorter alternative.
Context: ... simple layout (output: {type: null}) by accident, creating a user expectation that break...

(EN_WORDINESS_PREMIUM_BY_ACCIDENT)

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@dev-docs/bug-hunt-2026-04-02.md` at line 230, Replace the phrase "by
accident" in the sentence referencing `type: null` (the snippet "type: null
works in the simple layout (`output: {type: null}`) by accident, creating a user
expectation that breaks in the advanced list context") with a shorter
alternative such as "accidentally" or "inadvertently" to tighten phrasing and
improve readability while keeping the rest of the sentence and the references to
`type: null`, `output: {type: null}`, "simple layout", and "advanced list
context" unchanged.

[coderabbitai]

Actionable comments posted: 3

♻️ Duplicate comments (3)

crates/logfwd/src/worker_pool.rs (3)

233-239: ⚠️ Potential issue | 🔴 Critical

Don't await Shutdown on a bounded worker queue.

send(WorkerMsg::Shutdown).await can block Phase 1 behind a full worker channel, so graceful_timeout never starts. Use try_send() or put the shutdown signal under the same timeout budget instead of awaiting queue capacity during drain.

As per coding guidelines "Follow tokio async patterns including runtime, bounded channels, CancellationToken, and select! safety as documented in dev-docs/references/tokio-async-patterns.md".

🤖 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 233 - 239, The drain()
implementation currently awaits handle.tx.send(WorkerMsg::Shutdown).await which
can block if a worker's bounded channel is full and thus prevents starting the
graceful_timeout; change the shutdown signalling to a non-blocking send so Phase
1 cannot be stalled: in drain() replace the awaited send with
handle.tx.try_send(WorkerMsg::Shutdown) and handle the Result (ignore or log
Full/Closed errors appropriately) so the loop completes immediately and the
graceful_timeout can start; reference symbols: drain, WorkerMsg::Shutdown,
handle.tx.

---

191-208: ⚠️ Potential issue | 🔴 Critical

Handle every failed ownership transfer in submit().

WorkItem can still vanish in Step 2/3: spawn_worker() can fail when no live worker exists, the new-worker try_send result is ignored, and tx.send(msg).await can return Err if the chosen worker exits after Step 1. In all three cases the tickets never become an AckItem, so the pipeline cannot reject them.

As per coding guidelines "Follow tokio async patterns including runtime, bounded channels, CancellationToken, and select! safety as documented in dev-docs/references/tokio-async-patterns.md".

🤖 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 191 - 208, In submit(),
currently failures in spawn_worker(), the ignored try_send() result, and a
potential Err from tx.send(msg).await can cause WorkItem to vanish and its
ticket never become an AckItem; update submit() to handle all three failure
paths: if spawn_worker() returns Err, if handle.tx.try_send(msg) returns Err, or
if tx.send(msg).await returns Err, then convert the pending WorkItem ticket into
an AckItem (or otherwise signal rejection) and return appropriately so the
pipeline can observe the failure. Locate logic around submit(), spawn_worker(),
the new-worker try_send call and the async tx.send(msg).await, and ensure every
send/fork failure path explicitly transitions the ticket to an AckItem (or
enqueues a rejection) instead of silently dropping it. Ensure you
clone/send/return without holding &mut self across await as before.

---

256-263: ⚠️ Potential issue | 🔴 Critical

Forced drain still strands queued and in-flight tickets.

After timeout, cancel.cancel() plus JoinSet::shutdown() can drop the active WorkItem and any buffered WorkerMsg::Work without synthesizing failure acks. That breaks the drain invariant in Lines 24-25 and can leave tickets stuck in Sending. Add failure-on-drop handling for the active item and drain rx into AckItem { success: false } before the task exits or is aborted.

As per coding guidelines "Follow tokio async patterns including runtime, bounded channels, CancellationToken, and select! safety as documented in dev-docs/references/tokio-async-patterns.md".

Also applies to: 320-335

🤖 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 256 - 263, After the
forced-drain path that calls self.cancel.cancel() and
self.join_set.shutdown().await, ensure we synthesize failure acknowledgements
for any work that would otherwise be dropped: implement failure-on-drop for the
active WorkItem (so dropping it invokes an AckItem { success: false }) and,
before exiting the drain task (or before shutdown/abort), drain the worker rx
channel and send AckItem { success: false } for every buffered WorkerMsg::Work;
update the logic around cancel.cancel(), JoinSet::shutdown(), and the
WorkItem/WorkerMsg handling to guarantee all in-flight and queued tickets
transition from Sending to a terminal failed ack instead of being silently
dropped.

🤖 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/src/worker_pool.rs`:
- Around line 367-419: MAX_ATTEMPTS is off-by-one: with the current checks (if
attempts >= MAX_ATTEMPTS) we perform 1 initial send plus MAX_ATTEMPTS retries,
so change MAX_ATTEMPTS from 4 to 3 and update its comment to "1 initial + 3
retries"; adjust the const MAX_ATTEMPTS declaration and its inline comment where
defined, leaving the retry logic (checks of attempts >= MAX_ATTEMPTS, increments
of attempts) and the send_batch/SendResult handling untouched.
- Around line 132-147: The public constructor OutputWorkerPool::new currently
panics on bad input via assert!(max_workers >= 1); change its signature to
return Result<OutputWorkerPool, E> (e.g., Result<Self, anyhow::Error> or your
crate's error type), validate max_workers and return an Err describing the
invalid argument instead of aborting, and update the function body to construct
and return Ok(OutputWorkerPool { ... }) when valid; ensure callers of
OutputWorkerPool::new are updated to handle the Result.
- Around line 26-29: The module doc promises worker panic handling but currently
JoinSet::join_next() is only polled in drain(), so panics during steady-state
are not observed and no rejection AckItem is sent; update the worker-panic
handling by polling JoinSet::join_next() from the main loop (the same loop that
processes pipeline messages) — e.g., add a select! branch alongside message
handling to drain JoinSet::join_next() continuously and log the panic, or detect
worker channel closure and synthesize an AckItem rejection for the worker's
buffered/in-flight batch before removing its state; touch the code paths around
JoinSet::join_next, drain(), the worker channel handling, and AckItem creation
to ensure a rejection ack is emitted and logged when a worker task panics or
aborts.

---

Duplicate comments:
In `@crates/logfwd/src/worker_pool.rs`:
- Around line 233-239: The drain() implementation currently awaits
handle.tx.send(WorkerMsg::Shutdown).await which can block if a worker's bounded
channel is full and thus prevents starting the graceful_timeout; change the
shutdown signalling to a non-blocking send so Phase 1 cannot be stalled: in
drain() replace the awaited send with handle.tx.try_send(WorkerMsg::Shutdown)
and handle the Result (ignore or log Full/Closed errors appropriately) so the
loop completes immediately and the graceful_timeout can start; reference
symbols: drain, WorkerMsg::Shutdown, handle.tx.
- Around line 191-208: In submit(), currently failures in spawn_worker(), the
ignored try_send() result, and a potential Err from tx.send(msg).await can cause
WorkItem to vanish and its ticket never become an AckItem; update submit() to
handle all three failure paths: if spawn_worker() returns Err, if
handle.tx.try_send(msg) returns Err, or if tx.send(msg).await returns Err, then
convert the pending WorkItem ticket into an AckItem (or otherwise signal
rejection) and return appropriately so the pipeline can observe the failure.
Locate logic around submit(), spawn_worker(), the new-worker try_send call and
the async tx.send(msg).await, and ensure every send/fork failure path explicitly
transitions the ticket to an AckItem (or enqueues a rejection) instead of
silently dropping it. Ensure you clone/send/return without holding &mut self
across await as before.
- Around line 256-263: After the forced-drain path that calls
self.cancel.cancel() and self.join_set.shutdown().await, ensure we synthesize
failure acknowledgements for any work that would otherwise be dropped: implement
failure-on-drop for the active WorkItem (so dropping it invokes an AckItem {
success: false }) and, before exiting the drain task (or before shutdown/abort),
drain the worker rx channel and send AckItem { success: false } for every
buffered WorkerMsg::Work; update the logic around cancel.cancel(),
JoinSet::shutdown(), and the WorkItem/WorkerMsg handling to guarantee all
in-flight and queued tickets transition from Sending to a terminal failed ack
instead of being silently dropped.

🪄 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: 4b106a91-1e4b-40a0-bdbd-5322b1c26b72

📥 Commits

Reviewing files that changed from the base of the PR and between 7d65fb9 and 0b93d24.

📒 Files selected for processing (2)

• .coderabbit.yaml
• crates/logfwd/src/worker_pool.rs

[coderabbitai]

⚠️ Potential issue | 🔴 Critical

🧩 Analysis chain

🏁 Script executed:

# Check file structure and size
wc -l crates/logfwd/src/worker_pool.rs

Repository: strawgate/memagent

Length of output: 98

---

🏁 Script executed:

# Read module-level docs and the specific lines mentioned
sed -n '1,60p' crates/logfwd/src/worker_pool.rs | cat -n

Repository: strawgate/memagent

Length of output: 2889

---

🏁 Script executed:

# Check lines 243-250 mentioned in "Also applies to"
sed -n '240,260p' crates/logfwd/src/worker_pool.rs | cat -n

Repository: strawgate/memagent

Length of output: 961

---

🏁 Script executed:

# Examine worker_task and panic handling
sed -n '300,400p' crates/logfwd/src/worker_pool.rs | cat -n

Repository: strawgate/memagent

Length of output: 4528

---

🏁 Script executed:

# Check the drain() implementation mentioned
sed -n '200,250p' crates/logfwd/src/worker_pool.rs | cat -n

Repository: strawgate/memagent

Length of output: 2564

---

🏁 Script executed:

# Search for JoinSet usage and panic handling
rg -n "join_next|panic|panic!|\.abort" crates/logfwd/src/worker_pool.rs -A 2 -B 2

Repository: strawgate/memagent

Length of output: 1300

---

Steady-state panic recovery is unimplemented.

The module doc at lines 26-29 promises that worker panics are caught, logged, and rejection acks are sent. However, JoinSet::join_next() is only polled during drain() (line 244). During normal operation, a panicked worker's task completes silently—JoinSet holds the error internally, but no code polls it. The panicked worker exits without sending an AckItem for its queued or in-flight batch, so the pipeline never receives the rejection it needs to advance checkpoints with delivered=false. Buffered items are lost.

Move panic detection into steady-state by polling join_next in the main loop (e.g., via select! alongside pipeline messages), or synthesize failure acks when the worker channel closes from panic/abort.

🤖 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 26 - 29, The module doc
promises worker panic handling but currently JoinSet::join_next() is only polled
in drain(), so panics during steady-state are not observed and no rejection
AckItem is sent; update the worker-panic handling by polling
JoinSet::join_next() from the main loop (the same loop that processes pipeline
messages) — e.g., add a select! branch alongside message handling to drain
JoinSet::join_next() continuously and log the panic, or detect worker channel
closure and synthesize an AckItem rejection for the worker's buffered/in-flight
batch before removing its state; touch the code paths around JoinSet::join_next,
drain(), the worker channel handling, and AckItem creation to ensure a rejection
ack is emitted and logged when a worker task panics or aborts.

[coderabbitai]

⚠️ Potential issue | 🟠 Major

Return a validation error from new().

Line 133 turns an invalid max_workers config into a process abort. This constructor is public and takes caller-controlled input, so it should reject bad values with Result, not assert!.

Proposed fix

-    pub fn new(factory: Arc<dyn SinkFactory>, max_workers: usize, idle_timeout: Duration) -> Self {
-        assert!(max_workers >= 1, "max_workers must be at least 1");
+    pub fn new(
+        factory: Arc<dyn SinkFactory>,
+        max_workers: usize,
+        idle_timeout: Duration,
+    ) -> io::Result<Self> {
+        if max_workers == 0 {
+            return Err(io::Error::new(
+                io::ErrorKind::InvalidInput,
+                "output worker pool requires max_workers >= 1",
+            ));
+        }
         let (ack_tx, ack_rx) = mpsc::unbounded_channel();
-        OutputWorkerPool {
+        Ok(OutputWorkerPool {
             workers: VecDeque::with_capacity(max_workers),
             factory,
             ack_rx,
             ack_tx,
             cancel: CancellationToken::new(),
             join_set: JoinSet::new(),
             channel_capacity: 2,
             max_workers,
             idle_timeout,
             next_id: 0,
             max_retry_delay: Duration::from_secs(30),
-        }
+        })
     }

As per coding guidelines "Public API functions receiving user input must return Result/Option and never panic" and "Reject panic!(), assert!(), and unreachable!() in production paths; use debug_assert!() for internal invariants, never assert!() in production".

🤖 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 132 - 147, The public
constructor OutputWorkerPool::new currently panics on bad input via
assert!(max_workers >= 1); change its signature to return
Result<OutputWorkerPool, E> (e.g., Result<Self, anyhow::Error> or your crate's
error type), validate max_workers and return an Err describing the invalid
argument instead of aborting, and update the function body to construct and
return Ok(OutputWorkerPool { ... }) when valid; ensure callers of
OutputWorkerPool::new are updated to handle the Result.

[coderabbitai]

⚠️ Potential issue | 🟠 Major

MAX_ATTEMPTS is off by one.

Both retry branches compare attempts >= MAX_ATTEMPTS before incrementing, so MAX_ATTEMPTS = 4 currently means 1 initial send + 4 retries. That is an extra external write attempt under failure.

Proposed fix

-    let mut attempts = 0u32;
+    let mut attempts = 1u32;
@@
-                if attempts >= MAX_ATTEMPTS {
+                if attempts == MAX_ATTEMPTS {
                     eprintln!("worker_pool: worker {worker_id} RetryAfter exceeded max attempts");
                     return false;
                 }
@@
-                if attempts >= MAX_ATTEMPTS {
+                if attempts == MAX_ATTEMPTS {
                     eprintln!(
                         "worker_pool: worker {worker_id} gave up after {MAX_ATTEMPTS} attempts: {e}"
                     );
                     return false;
                 }

🤖 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 367 - 419, MAX_ATTEMPTS is
off-by-one: with the current checks (if attempts >= MAX_ATTEMPTS) we perform 1
initial send plus MAX_ATTEMPTS retries, so change MAX_ATTEMPTS from 4 to 3 and
update its comment to "1 initial + 3 retries"; adjust the const MAX_ATTEMPTS
declaration and its inline comment where defined, leaving the retry logic
(checks of attempts >= MAX_ATTEMPTS, increments of attempts) and the
send_batch/SendResult handling untouched.

[coderabbitai]

Actionable comments posted: 5

♻️ Duplicate comments (9)

crates/logfwd-output/src/lib.rs (1)

518-531: ⚠️ Potential issue | 🟠 Major

The new Loki knobs are still unreachable from YAML.

This factory call hardcodes tenant_id, static_labels, and label_columns to empty values, and crates/logfwd-config/src/lib.rs:142-156 does not expose those fields. Multi-tenant routing and label-based stream partitioning stay dead until those fields are added to OutputConfig and plumbed through here.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@crates/logfwd-output/src/lib.rs` around lines 518 - 531, The Loki branch
currently calls LokiSinkFactory::new with hardcoded tenant_id (None),
static_labels (Vec::new()), and label_columns (Vec::new()), and OutputConfig
(crates/logfwd-config::OutputConfig) does not expose those fields; add
tenant_id, static_labels, and label_columns to OutputConfig, parse them from
YAML, and then replace the hardcoded values in the OutputType::Loki branch
(where LokiSinkFactory::new is invoked) to pass the configured tenant_id,
static_labels, and label_columns through to the factory so multi-tenant routing
and label-based partitioning are reachable from config.

crates/logfwd-output/src/loki.rs (3)

93-98: ⚠️ Potential issue | 🟠 Major

Loki still shares one reqwest::Client across all workers.

The factory clones an Arc<reqwest::Client> into each sink, so dropping an idle worker does not drop its HTTP pool. That defeats the same idle-connection cleanup goal already handled in Elasticsearch; build the client in create() and store it directly on LokiAsyncSink.

In reqwest 0.12, does cloning or Arc-wrapping `reqwest::Client` share the same connection pool, and do pooled connections stay alive while any clone exists?

Also applies to: 350-413

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@crates/logfwd-output/src/loki.rs` around lines 93 - 98, LokiAsyncSink
currently holds client: Arc<reqwest::Client>, which causes all sinks to share
the same pool; instead construct the reqwest::Client inside the sink factory
create() and store it directly on LokiAsyncSink (e.g., change the field to
client: reqwest::Client), stop cloning or Arc-wrapping the client in the
factory, and update all usages and constructors that reference
LokiAsyncSink::client to use the owned Client; this ensures dropping a
worker/sink drops its connection pool.

---

131-168: ⚠️ Potential issue | 🟠 Major

Timestamp resolution is still using raw schema names and unchecked casts.

The combined position() does not actually prefer _timestamp, it misses normalized fields like @timestamp_str / _timestamp_int, and the later Int64 as u64 cast turns null or negative timestamps into bogus nanosecond values. Resolve the timestamp from build_col_infos() once and fall back to observed_time_ns for null/invalid rows before converting.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@crates/logfwd-output/src/loki.rs` around lines 131 - 168, The timestamp logic
in the loop (ts_col_idx and the DataType match) incorrectly relies on raw field
names and performs unchecked casts (e.g., Int64 -> u64) allowing null/negative
values to become bogus timestamps; instead, use the resolved timestamp column
info produced by build_col_infos() (the column index/type resolved earlier) to
determine the timestamp column, check for nulls or invalid/negative values
per-row before converting, and if a row is null/invalid fall back to
metadata.observed_time_ns; ensure conversions handle signed Int64 safely (reject
negative) and UInt64/null checks before constructing ts_ns so no unchecked
as_primitive casts produce invalid nanoseconds.

---

68-76: ⚠️ Potential issue | 🟠 Major

Preserve equal-timestamp order and surface u64::MAX overflow.

sort_unstable_by_key() can reorder equal-timestamp entries before dedup, so same-ns logs may be emitted reversed. On top of that, + 1 wraps or panics once a stream reaches u64::MAX. Use a stable sort here, and make this helper return an error when it can no longer advance the timestamp.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@crates/logfwd-output/src/loki.rs` around lines 68 - 76, The helper
sort_and_dedup_timestamps should preserve original order for equal timestamps
and surface overflow instead of wrapping; change it to use a stable sort
(replace sort_unstable_by_key with sort_by_key / stable equivalent) and make the
function return a Result (e.g., Result<(), SomeError>) instead of void so
callers can handle failure when a timestamp cannot be incremented because it is
u64::MAX; update the loop in sort_and_dedup_timestamps to check before doing ts
= prev + 1 and return an error when prev == u64::MAX, and update any callers of
sort_and_dedup_timestamps to propagate or handle the returned Result.

crates/logfwd/src/pipeline.rs (2)

243-248: ⚠️ Potential issue | 🟠 Major

Keep with_output() workers non-expiring.

This path still wraps the sink in OnceFactory, but the worker idles out after 30s. Once that sole worker exits, the factory is exhausted and every later batch is rejected. Match the from_config() path and disable idle expiry here too.

Suggested fix

-        self.pool = OutputWorkerPool::new(factory, 1, Duration::from_secs(30));
+        self.pool = OutputWorkerPool::new(factory, 1, Duration::MAX);

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@crates/logfwd/src/pipeline.rs` around lines 243 - 248, The with_output() path
currently constructs the OutputWorkerPool with a 30s idle timeout which lets the
sole worker exit and exhaust the OnceFactory; change the idle timeout passed to
OutputWorkerPool::new in with_output (where factory is created via
OnceFactory::new and self.pool is assigned) to disable expiry (e.g., use a
zero/none duration consistent with the from_config() path) so the single worker
never idles out.

---

604-640: ⚠️ Potential issue | 🟠 Major

Record output completion on the ack path, not submission.

record_batch() still runs before the pool has accepted or delivered anything, so slow or failed sends contribute near-zero output time and completed batch rows up front. AckItem needs at least submitted_at and row_count so apply_pool_ack() can finalize metrics when delivery actually resolves.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@crates/logfwd/src/pipeline.rs` around lines 604 - 640, The code currently
calls self.metrics.record_batch(...) at submission time; instead add
submitted_at: Instant (or nanos) and row_count: usize to AckItem in the
pool/WorkItem path, capture those when submitting in submit(WorkItem { ... })
and remove the pre-submit self.metrics.record_batch(...) call, then in
apply_pool_ack(&mut self, ack: AckItem) compute the true submission→completion
duration (using ack.submitted_at and now_nanos()/Instant::now()), use
ack.row_count for rows, and call self.metrics.record_batch(...) there to
finalize output timing; keep the existing failure path
(inc_dropped_batch/output_error) and still call
self.ack_all_tickets(ack.tickets, ack.success).

crates/logfwd/src/worker_pool.rs (2)

260-268: ⚠️ Potential issue | 🔴 Critical

Worker panics are still only observed during drain().

This is the only place join_set.join_next() is polled. A worker that panics during steady state can therefore lose its current or queued WorkItems without any AckItem; by the time a later try_send() notices Closed, those tickets are already orphaned. If panic recovery is part of the contract, the pool needs to reap join_set during normal operation and reject the outstanding work explicitly.

🤖 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 260 - 268, The pool currently
only polls join_set.join_next() inside drain(), so panicked workers are only
observed at shutdown and in-flight/queued WorkItem tickets can become orphaned;
update the runtime to reap join_set during normal operation (e.g., integrate
join_set.join_next() into the main select loop used by try_send/worker dispatch
or spawn a dedicated reaper task) so worker panics are detected immediately, and
when a panic is observed from join_set, explicitly reject outstanding WorkItem
tickets (produce corresponding AckItem failures or return an error from
try_send) and clean up any related state; refer to join_set, drain(),
try_send(), WorkItem and AckItem to locate where to add the join_set polling and
the rejection logic.

---

278-299: ⚠️ Potential issue | 🔴 Critical

Cancellation still drops buffered work instead of rejecting it.

Because the worker select! is biased, cancel.cancelled() wins over a ready rx.recv(). After drain() times out and fires the token, any queued WorkerMsg::Work still sitting in the channel is discarded when the task breaks, and no AckItem { success: false } is emitted for those tickets. Drain needs to flush buffered work into failure acks or keep the tickets outside the task before aborting.

Based on learnings, "Follow tokio async patterns including runtime, bounded channels, CancellationToken, and select! safety as documented in dev-docs/references/tokio-async-patterns.md".

Also applies to: 356-371

🤖 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 278 - 299, drain() currently
fires self.cancel.cancel() which wins in the worker's biased select! and causes
any buffered WorkerMsg::Work in the rx channel to be dropped without emitting
AckItem failures; change drain() to first flush the receiver(s) of any pending
WorkerMsg::Work (iteratively recv() until empty or timeout) and for each queued
Work produce/send an AckItem { success: false } (or move tickets back to the
caller-facing queue) before calling self.cancel.cancel(), or alternatively alter
the worker select! so it does not bias cancellation over rx.recv() (ensure
rx.recv() is polled first or use non-biased select semantics) so in-flight
queued work is either processed or explicitly failed prior to
self.join_set.shutdown().await.

crates/logfwd-output/src/sink.rs (1)

53-57: ⚠️ Potential issue | 🟠 Major

Benchmark the boxed send_batch future before locking it into the hot path.

This signature allocates once per batch on every send, and the module docs now make that overhead part of the core sink API. I’d want before/after perf data or a non-boxed fast path before merging this into the send hot path.

As per coding guidelines, "Do not introduce allocations in hot paths without benchmarking".

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@crates/logfwd-output/src/sink.rs` around lines 53 - 57, The new send_batch
signature boxes a Future per batch (allocating in the hot path); either
benchmark the before/after cost and attach results to the PR, or change the
trait to avoid per-call allocation by introducing an associated future type
(e.g., add type SendFuture<'a>: Future<Output = io::Result<SendResult>> + Send +
'a and change fn send_batch(...) -> Self::SendFuture<'a>), then update all
implementors to return that concrete future (or provide both a boxed default and
a non-boxed fast-path implementation). Reference send_batch, SendResult,
BatchMetadata, and RecordBatch when making the change so implementors and
reviewers can find the API points to update.

🤖 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-output/src/elasticsearch.rs`:
- Around line 540-543: The code currently clones the entire NDJSON payload with
self.batch_buf.clone() before each async send, causing an extra allocation and
memcpy; change the call site to move the buffer into the send path (e.g., use
std::mem::take(&mut self.batch_buf) or swap-out the Vec) and adjust do_send to
accept Vec<u8> by value so you can compute byte_len and call
self.do_send(body).await? without cloning; ensure you reset self.batch_buf to an
empty buffer after moving it out and preserve row_count computation from
batch.num_rows() as needed.

In `@crates/logfwd-output/src/fanout.rs`:
- Around line 61-231: The tests module (#[cfg(test)] mod tests) is placed before
the impl OutputSink for FanOut, violating the rule that unit tests must be at
the bottom; move the entire #[cfg(test)] mod tests { ... } block to the end of
this file (after the impl OutputSink for FanOut and any other impls) so that
impl OutputSink for FanOut and related code appear before the tests and the
tests module is the last item in the module.

In `@crates/logfwd-output/src/json_lines.rs`:
- Around line 121-286: The #[cfg(test)] mod tests is placed before the impl
OutputSink for JsonLinesSink, splitting production code; move the entire test
module block (the module that contains tests referencing JsonLinesSink, batch_*
helpers, and tests like serialize_batch... and passthrough_raw_only) to the end
of the file so that impl OutputSink for JsonLinesSink remains contiguous and all
unit tests live at the bottom of the module.

In `@crates/logfwd-output/src/sink.rs`:
- Around line 121-143: The SyncSinkAdapter currently calls
tokio::task::block_in_place in send_batch, flush, and shutdown, which makes a
runtime assumption; change these to use tokio::task::spawn_blocking and await
the JoinHandle instead. For the async closures returned by send_batch and
flush/shutdown, call spawn_blocking(move || self.inner.send_batch(batch,
metadata)) (and spawn_blocking(move || self.inner.flush()) for flush/shutdown),
await the handle, propagate/map the join error and inner Result into the
expected Future::Output (e.g. map inner Ok(()) to SendResult::Ok and map
io::Error or JoinError to an io::Result::Err), and keep name() delegating to
self.inner.name(); update references to SyncSinkAdapter, self.inner, send_batch,
flush, and shutdown accordingly.

In `@crates/logfwd/src/worker_pool.rs`:
- Around line 163-166: The submit function currently only uses debug_assert! to
prevent submissions after drain, so in release builds callers can still enqueue
WorkItem which may be dropped without an AckItem; change submit(&mut self, item:
WorkItem) to return a Result (or Option) and explicitly check
cancel.is_cancelled() at runtime, returning an Err (or None) if cancelled;
similarly update the code path that creates WorkerMsg::Work and the
spawn_worker/worker-launching logic (the block referenced around lines 191-197)
to either not spawn a worker when cancelled or to immediately send a failure
AckItem for the provided WorkItem before returning the error, and update
callers/tests to handle the new Result/Option return.

---

Duplicate comments:
In `@crates/logfwd-output/src/lib.rs`:
- Around line 518-531: The Loki branch currently calls LokiSinkFactory::new with
hardcoded tenant_id (None), static_labels (Vec::new()), and label_columns
(Vec::new()), and OutputConfig (crates/logfwd-config::OutputConfig) does not
expose those fields; add tenant_id, static_labels, and label_columns to
OutputConfig, parse them from YAML, and then replace the hardcoded values in the
OutputType::Loki branch (where LokiSinkFactory::new is invoked) to pass the
configured tenant_id, static_labels, and label_columns through to the factory so
multi-tenant routing and label-based partitioning are reachable from config.

In `@crates/logfwd-output/src/loki.rs`:
- Around line 93-98: LokiAsyncSink currently holds client: Arc<reqwest::Client>,
which causes all sinks to share the same pool; instead construct the
reqwest::Client inside the sink factory create() and store it directly on
LokiAsyncSink (e.g., change the field to client: reqwest::Client), stop cloning
or Arc-wrapping the client in the factory, and update all usages and
constructors that reference LokiAsyncSink::client to use the owned Client; this
ensures dropping a worker/sink drops its connection pool.
- Around line 131-168: The timestamp logic in the loop (ts_col_idx and the
DataType match) incorrectly relies on raw field names and performs unchecked
casts (e.g., Int64 -> u64) allowing null/negative values to become bogus
timestamps; instead, use the resolved timestamp column info produced by
build_col_infos() (the column index/type resolved earlier) to determine the
timestamp column, check for nulls or invalid/negative values per-row before
converting, and if a row is null/invalid fall back to metadata.observed_time_ns;
ensure conversions handle signed Int64 safely (reject negative) and UInt64/null
checks before constructing ts_ns so no unchecked as_primitive casts produce
invalid nanoseconds.
- Around line 68-76: The helper sort_and_dedup_timestamps should preserve
original order for equal timestamps and surface overflow instead of wrapping;
change it to use a stable sort (replace sort_unstable_by_key with sort_by_key /
stable equivalent) and make the function return a Result (e.g., Result<(),
SomeError>) instead of void so callers can handle failure when a timestamp
cannot be incremented because it is u64::MAX; update the loop in
sort_and_dedup_timestamps to check before doing ts = prev + 1 and return an
error when prev == u64::MAX, and update any callers of sort_and_dedup_timestamps
to propagate or handle the returned Result.

In `@crates/logfwd-output/src/sink.rs`:
- Around line 53-57: The new send_batch signature boxes a Future per batch
(allocating in the hot path); either benchmark the before/after cost and attach
results to the PR, or change the trait to avoid per-call allocation by
introducing an associated future type (e.g., add type SendFuture<'a>:
Future<Output = io::Result<SendResult>> + Send + 'a and change fn
send_batch(...) -> Self::SendFuture<'a>), then update all implementors to return
that concrete future (or provide both a boxed default and a non-boxed fast-path
implementation). Reference send_batch, SendResult, BatchMetadata, and
RecordBatch when making the change so implementors and reviewers can find the
API points to update.

In `@crates/logfwd/src/pipeline.rs`:
- Around line 243-248: The with_output() path currently constructs the
OutputWorkerPool with a 30s idle timeout which lets the sole worker exit and
exhaust the OnceFactory; change the idle timeout passed to OutputWorkerPool::new
in with_output (where factory is created via OnceFactory::new and self.pool is
assigned) to disable expiry (e.g., use a zero/none duration consistent with the
from_config() path) so the single worker never idles out.
- Around line 604-640: The code currently calls self.metrics.record_batch(...)
at submission time; instead add submitted_at: Instant (or nanos) and row_count:
usize to AckItem in the pool/WorkItem path, capture those when submitting in
submit(WorkItem { ... }) and remove the pre-submit
self.metrics.record_batch(...) call, then in apply_pool_ack(&mut self, ack:
AckItem) compute the true submission→completion duration (using ack.submitted_at
and now_nanos()/Instant::now()), use ack.row_count for rows, and call
self.metrics.record_batch(...) there to finalize output timing; keep the
existing failure path (inc_dropped_batch/output_error) and still call
self.ack_all_tickets(ack.tickets, ack.success).

In `@crates/logfwd/src/worker_pool.rs`:
- Around line 260-268: The pool currently only polls join_set.join_next() inside
drain(), so panicked workers are only observed at shutdown and in-flight/queued
WorkItem tickets can become orphaned; update the runtime to reap join_set during
normal operation (e.g., integrate join_set.join_next() into the main select loop
used by try_send/worker dispatch or spawn a dedicated reaper task) so worker
panics are detected immediately, and when a panic is observed from join_set,
explicitly reject outstanding WorkItem tickets (produce corresponding AckItem
failures or return an error from try_send) and clean up any related state; refer
to join_set, drain(), try_send(), WorkItem and AckItem to locate where to add
the join_set polling and the rejection logic.
- Around line 278-299: drain() currently fires self.cancel.cancel() which wins
in the worker's biased select! and causes any buffered WorkerMsg::Work in the rx
channel to be dropped without emitting AckItem failures; change drain() to first
flush the receiver(s) of any pending WorkerMsg::Work (iteratively recv() until
empty or timeout) and for each queued Work produce/send an AckItem { success:
false } (or move tickets back to the caller-facing queue) before calling
self.cancel.cancel(), or alternatively alter the worker select! so it does not
bias cancellation over rx.recv() (ensure rx.recv() is polled first or use
non-biased select semantics) so in-flight queued work is either processed or
explicitly failed prior to self.join_set.shutdown().await.

🪄 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: a558b32e-632f-4c21-9245-582ea84b32cf

📥 Commits

Reviewing files that changed from the base of the PR and between 0b93d24 and 51cea83.

📒 Files selected for processing (8)

• crates/logfwd-output/src/elasticsearch.rs
• crates/logfwd-output/src/fanout.rs
• crates/logfwd-output/src/json_lines.rs
• crates/logfwd-output/src/lib.rs
• crates/logfwd-output/src/loki.rs
• crates/logfwd-output/src/sink.rs
• crates/logfwd/src/pipeline.rs
• crates/logfwd/src/worker_pool.rs

[coderabbitai]

⚠️ Potential issue | 🟠 Major

Avoid cloning the whole bulk body before every async send.

self.batch_buf.clone() duplicates the entire NDJSON payload right before every request, so large batches now pay one extra allocation and memcpy on the steady-state path. Move the buffer into do_send() or add benchmark justification for the extra copy. As per coding guidelines, "Do not introduce allocations in hot paths without benchmarking" and "Reject any allocation growing proportionally to number of lines per batch without benchmark justification".

🤖 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 540 - 543, The code
currently clones the entire NDJSON payload with self.batch_buf.clone() before
each async send, causing an extra allocation and memcpy; change the call site to
move the buffer into the send path (e.g., use std::mem::take(&mut
self.batch_buf) or swap-out the Vec) and adjust do_send to accept Vec<u8> by
value so you can compute byte_len and call self.do_send(body).await? without
cloning; ensure you reset self.batch_buf to an empty buffer after moving it out
and preserve row_count computation from batch.num_rows() as needed.

[coderabbitai]

⚠️ Potential issue | 🟡 Minor

Move this #[cfg(test)] module to the end of the file.

impl OutputSink for FanOut still appears after the tests, so the module layout now breaks the repo's test-placement rule. As per coding guidelines, "Unit tests live in #[cfg(test)] mod tests {} at bottom of module file".

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@crates/logfwd-output/src/fanout.rs` around lines 61 - 231, The tests module
(#[cfg(test)] mod tests) is placed before the impl OutputSink for FanOut,
violating the rule that unit tests must be at the bottom; move the entire
#[cfg(test)] mod tests { ... } block to the end of this file (after the impl
OutputSink for FanOut and any other impls) so that impl OutputSink for FanOut
and related code appear before the tests and the tests module is the last item
in the module.

[coderabbitai]

⚠️ Potential issue | 🟡 Minor

Move this #[cfg(test)] module to the end of the file.

The trailing impl OutputSink for JsonLinesSink means the tests now split production code in the middle of the module. As per coding guidelines, "Unit tests live in #[cfg(test)] mod tests {} at bottom of module file".

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@crates/logfwd-output/src/json_lines.rs` around lines 121 - 286, The
#[cfg(test)] mod tests is placed before the impl OutputSink for JsonLinesSink,
splitting production code; move the entire test module block (the module that
contains tests referencing JsonLinesSink, batch_* helpers, and tests like
serialize_batch... and passthrough_raw_only) to the end of the file so that impl
OutputSink for JsonLinesSink remains contiguous and all unit tests live at the
bottom of the module.

[coderabbitai]

⚠️ Potential issue | 🟠 Major

🧩 Analysis chain

🌐 Web query:

Does Tokio 1.x tokio::task::block_in_placepanic on the current-thread runtime, and can a future wrapped intokio::time::timeoutor task abort be preempted while theblock_in_place closure is still running?

💡 Result:

Yes, tokio::task::block_in_place panics if called on the current-thread runtime. No, a future wrapped in tokio::time::timeout or aborted via task abort cannot be preempted while the block_in_place closure is running; code inside block_in_place cannot be cancelled and runs to completion on the current thread.

Citations:

• 1: https://docs.rs/tokio/latest/tokio/task/fn.block_in_place.html
• 2: https://docs.rs/tokio/latest/src/tokio/task/blocking.rs.html
• 3: block_in_place + block_on can hang on runtime shutdown / runtime drop tokio-rs/tokio#6463

---

🏁 Script executed:

# Check the actual test configuration in worker_pool.rs
sed -n '884,992p' crates/logfwd/src/worker_pool.rs | head -50

Repository: strawgate/memagent

Length of output: 1905

---

🏁 Script executed:

# Check if tokio-async-patterns.md exists and what it says about block_in_place
find . -name "*tokio*async*patterns*" -o -name "tokio-async-patterns.md" | head -5

Repository: strawgate/memagent

Length of output: 107

---

🏁 Script executed:

# Search for uses of SyncSinkAdapter in the codebase, particularly in tests
rg "SyncSinkAdapter" -B 3 -A 3 --type rs

Repository: strawgate/memagent

Length of output: 88

---

🏁 Script executed:

# Check if there are any tests that instantiate or use the adapter
rg "#\[tokio::test\]" crates/logfwd-output/src/ -A 10 --type rs | head -50

Repository: strawgate/memagent

Length of output: 88

---

🏁 Script executed:

# Read the tokio-async-patterns.md file
cat dev-docs/references/tokio-async-patterns.md

Repository: strawgate/memagent

Length of output: 16166

---

🏁 Script executed:

# Check the full sink.rs file to understand SyncSinkAdapter context
head -150 crates/logfwd-output/src/sink.rs

Repository: strawgate/memagent

Length of output: 5435

---

🏁 Script executed:

# Search for SyncSinkAdapter usage (without type flag)
rg "SyncSinkAdapter" -B 2 -A 2

Repository: strawgate/memagent

Length of output: 2586

---

SyncSinkAdapter should use spawn_blocking instead of block_in_place to remain runtime-agnostic.

block_in_place() with an undocumented assumption that logfwd uses a multi-threaded runtime is fragile. The API is public and exported, so callers may invoke it from current-thread tests or other contexts where it would panic. Per dev-docs/references/tokio-async-patterns.md (section 6), wrap blocking sync calls with spawn_blocking instead — it's the documented pattern for safe sync-to-async bridging and avoids baking in runtime assumptions.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@crates/logfwd-output/src/sink.rs` around lines 121 - 143, The SyncSinkAdapter
currently calls tokio::task::block_in_place in send_batch, flush, and shutdown,
which makes a runtime assumption; change these to use
tokio::task::spawn_blocking and await the JoinHandle instead. For the async
closures returned by send_batch and flush/shutdown, call spawn_blocking(move ||
self.inner.send_batch(batch, metadata)) (and spawn_blocking(move ||
self.inner.flush()) for flush/shutdown), await the handle, propagate/map the
join error and inner Result into the expected Future::Output (e.g. map inner
Ok(()) to SendResult::Ok and map io::Error or JoinError to an io::Result::Err),
and keep name() delegating to self.inner.name(); update references to
SyncSinkAdapter, self.inner, send_batch, flush, and shutdown accordingly.

[coderabbitai]

⚠️ Potential issue | 🔴 Critical

Reject submit() after drain in release builds too.

debug_assert! is the only guard here. Once cancel is set, this path can still spawn a worker with an already-cancelled token; that task exits before reading its channel, so the enqueued WorkItem is dropped without an AckItem. Make post-drain submission an explicit error or failure-ack path instead of a debug-only check.

As per coding guidelines, "Public API functions receiving user input must return Result/Option and never panic".

Also applies to: 191-197

🤖 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 163 - 166, The submit function
currently only uses debug_assert! to prevent submissions after drain, so in
release builds callers can still enqueue WorkItem which may be dropped without
an AckItem; change submit(&mut self, item: WorkItem) to return a Result (or
Option) and explicitly check cancel.is_cancelled() at runtime, returning an Err
(or None) if cancelled; similarly update the code path that creates
WorkerMsg::Work and the spawn_worker/worker-launching logic (the block
referenced around lines 191-197) to either not spawn a worker when cancelled or
to immediately send a failure AckItem for the provided WorkItem before returning
the error, and update callers/tests to handle the new Result/Option return.