Refactor to unify OutputSink and Sink traits by removing sync outputs

[Copilot]

• Explore codebase and understand OutputSink/Sink architecture
• Deprecate OutputSink trait with #[deprecated] attribute in lib.rs
• Add #[allow(deprecated)] to all existing impl OutputSink blocks (production sinks, test helpers, FanOut, etc.)
• Add #[allow(deprecated)] to all references to OutputSink in type positions (OnceFactory, build_output_sink, FanOut::new, etc.)
• Add OnceAsyncFactory — wraps a single Box<dyn Sink> into a SinkFactory without going through sync adapter
• Add tests for OnceAsyncFactory
• Update sink.rs docs to mark Sink as canonical
• Convert logfwd-test-utils/sinks.rs to implement async Sink trait (DevNullSink, SlowSink, FrozenSink, FailingSink, CountingSink)
• Add Pipeline::with_sink method accepting Box<dyn Sink> directly
• Deprecate Pipeline::with_output
• Update pipeline.rs test call sites to use with_sink instead of with_output
• Update compliance.rs CaptureSink to implement async Sink
• Update bench pipeline.rs NullSink with #[allow(deprecated)]
• Fix pre-existing unused imports in pipeline.rs tests
• cargo clippy -- -D warnings passes
• cargo fmt --check passes
• cargo check --workspace --tests passes
• Full cargo test not yet confirmed (was running when session ended)

[strawgate]

@coderabbitai review

[coderabbitai]

✅ Actions performed

Review triggered.

Note: CodeRabbit is an incremental review system and does not re-review already reviewed commits. This command is applicable only when automatic reviews are paused.

[coderabbitai]

Important

Review skipped

Bot user detected.

To trigger a single review, invoke the @coderabbitai review command.

⚙️ Run configuration

Configuration used: Repository YAML (base), Organization UI (inherited)

Review profile: ASSERTIVE

Plan: Pro

Run ID: 257c6d3f-b456-40c1-b813-57e46dcefe03

You can disable this status message by setting the reviews.review_status to false in the CodeRabbit configuration file.

Use the checkbox below for a quick retry:

• 🔍 Trigger review

Walkthrough

This PR transitions the logfwd output subsystem from a synchronous OutputSink trait to an asynchronous Sink trait. A new OnceAsyncFactory type is introduced to manage single-use async sink instances. The deprecated OutputSink trait is annotated across all existing sync sink implementations with #[allow(deprecated)] to suppress warnings. Test utilities are migrated to implement the async Sink trait with tokio-based async operations. The Pipeline API gains a new with_sink method accepting async sinks and deprecates the legacy with_output method. The migration maintains backward compatibility through SyncSinkAdapter bridging.

Possibly related PRs

• feat: async output worker pool (MRU dispatch) + ES/Loki async sinks #701 — Directly related as it migrates from legacy OutputSink to async Sink/SinkFactory abstraction, introducing the same SyncSinkAdapter, OnceFactory, and OnceAsyncFactory types.
• fix: delete sync ElasticsearchSink, port bench + tests to async factory #793 — Performs parallel sync-to-async sink conversion with factory model updates, touching the same sink/factory/pipeline code paths and deprecation patterns.
• feat: wire FileCheckpointStore into pipeline — persist and restore file offsets (#588) #784 — Modifies pipeline wiring in Pipeline::from_config and output factory integration, overlapping with the sink injection mechanism changes.

---

Caution

Pre-merge checks failed

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

• Ignore

❌ Failed checks (2 errors, 2 warnings)

Check name	Status	Explanation	Resolution
High-Quality Rust Practices	❌ Error	PR violates High-Quality Rust Practices: missing doc comment on new public constructor, improper timeout configuration causing premature batch rejection, and non-workspace tokio dependency.	Add doc comment to OnceAsyncFactory::new(), replace Duration::from_secs(30) with Duration::MAX in Pipeline::with_sink(), move tokio to workspace dependencies.
Crate Boundary And Dependency Integrity	❌ Error	PR violates workspace dependency policy by adding tokio as crate-local dependency instead of centralizing in [workspace.dependencies], and logfwd-test-utils lacks documentation in CRATE_RULES.md.	Centralize tokio in [workspace.dependencies], replace crate-local declarations with workspace references, document logfwd-test-utils in CRATE_RULES.md, and verify logfwd-output dependencies align with crate rules.
Documentation Thoroughly Updated	⚠️ Warning	Missing doc comment on OnceAsyncFactory::new() despite review feedback, ARCHITECTURE.md outdated with OutputSink still marked canonical, no ADR in DESIGN.md for OutputSink deprecation, PHASES.md task status unclear.	Add doc comment to OnceAsyncFactory::new(), update ARCHITECTURE.md section 7 to show Sink as canonical, add ADR to DESIGN.md explaining OutputSink deprecation, mark PHASES.md task 10b complete.
Maintainer Fitness	⚠️ Warning	PR has three unresolved review comments: tokio workspace dependency violation in logfwd-test-utils/Cargo.toml, missing doc comment on OnceAsyncFactory::new, and incorrect Duration::from_secs(30) timeout causing idle-expiration bugs in Pipeline::with_sink.	Fix workspace dependency violation by adding tokio to root [workspace.dependencies] with workspace = true; add /// doc comment to OnceAsyncFactory::new; change Duration::from_secs(30) to Duration::MAX in Pipeline::with_sink for single-use sinks.

✅ Passed checks (1 passed)

Check name	Status	Explanation
Formal Verification Coverage	✅ Passed	The formal verification check references a non-existent 'logfwd-core' crate and missing guidance documents; no Kani proof infrastructure exists in this repository.

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

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

[coderabbitai]

Actionable comments posted: 3

🤖 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/sink.rs`:
- Around line 172-173: Add a /// doc comment for the public constructor
OnceAsyncFactory::new describing what the factory does, its parameters (name:
String and sink: Box<dyn Sink>), and the returned type/behavior (creates a
OnceAsyncFactory that will forward to the provided sink once asynchronously).
Follow the project's CONFIG_REFERENCE.md doc style used elsewhere in this module
(brief description line, param annotations for `name` and `sink`, and any
panics/notes if applicable) and place the comment directly above the impl/new
signature to match the rest of the public output API.

In `@crates/logfwd-test-utils/Cargo.toml`:
- Line 13: The crate currently pins tokio locally (tokio = { version = "1",
features = ["rt", "time", "macros"] }); instead declare tokio in the root
[workspace.dependencies] with the explicit version (e.g., tokio = "1") and then
change this crate's dependency to reference the workspace (tokio = { workspace =
true, features = ["rt","time","macros"] }) so the workspace policy is followed;
update the workspace root Cargo.toml and replace the local tokio entry in this
crate's Cargo.toml accordingly.

In `@crates/logfwd/src/pipeline.rs`:
- Around line 287-290: The single-use async sink created in with_sink currently
uses a 30s idle timeout which lets the sole worker exit and causes subsequent
create() to hit the "already consumed" error; change the timeout passed to
OutputWorkerPool::new in with_sink to the same non-expiring value used for
single-use sinks in from_config (e.g., use Duration::MAX or the same sentinel
timeout constant) so the single worker will not idle-expire and the
OnceAsyncFactory won't be re-created and rejected.

🪄 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: 00adaf6b-f995-4ca8-b7c9-8ba398c6c09e

📥 Commits

Reviewing files that changed from the base of the PR and between 26cee6e and dbfb2a0.

⛔ Files ignored due to path filters (1)

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

📒 Files selected for processing (16)

• crates/logfwd-bench/benches/pipeline.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/null.rs
• crates/logfwd-output/src/otlp_sink.rs
• crates/logfwd-output/src/parquet.rs
• crates/logfwd-output/src/sink.rs
• crates/logfwd-output/src/stdout.rs
• crates/logfwd-output/src/tcp_sink.rs
• crates/logfwd-output/src/udp_sink.rs
• crates/logfwd-test-utils/Cargo.toml
• crates/logfwd-test-utils/src/sinks.rs
• crates/logfwd/src/pipeline.rs
• crates/logfwd/tests/compliance.rs

[coderabbitai]

⚠️ Potential issue | 🟡 Minor

Document OnceAsyncFactory::new.

Lines 172-173 add a new public constructor without a /// doc comment. Please document it like the rest of the public output API.

Suggested fix

 impl OnceAsyncFactory {
+    /// Create a single-use factory around one pre-built async sink.
     pub fn new(name: String, sink: Box<dyn Sink>) -> Self {
         OnceAsyncFactory {
             name,
             inner: Mutex::new(Some(sink)),
         }

As per coding guidelines, "All new public items must have /// doc comments matching CONFIG_REFERENCE.md where applicable".

📝 Committable suggestion

‼️ 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.

Suggested change

	impl OnceAsyncFactory {
	pub fn new(name: String, sink: Box<dyn Sink>) -> Self {
	impl OnceAsyncFactory {
	/// Create a single-use factory around one pre-built async sink.
	pub fn new(name: String, sink: Box<dyn Sink>) -> Self {

🤖 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 172 - 173, Add a /// doc
comment for the public constructor OnceAsyncFactory::new describing what the
factory does, its parameters (name: String and sink: Box<dyn Sink>), and the
returned type/behavior (creates a OnceAsyncFactory that will forward to the
provided sink once asynchronously). Follow the project's CONFIG_REFERENCE.md doc
style used elsewhere in this module (brief description line, param annotations
for `name` and `sink`, and any panics/notes if applicable) and place the comment
directly above the impl/new signature to match the rest of the public output
API.