downstream: add coroutine-based event dispatch

[edsiper]

Purpose

Add coroutine-based event dispatch to the downstream core and use Forward as
the first end-to-end consumer for both TLS and plain stream connections.

This PR is based on master (cb7256c28), replaces #11547, and fixes #11551.
Instead of bounding TLS retry loops in plugin code, it fixes the downstream
execution model: callbacks can suspend on read/write readiness and resume when
the connection has data, reaches EOF, fails, times out, or is cancelled.

Design

• Downstream owns a persistent callback coroutine per registered connection.
• Engine, threaded-input, and downstream-worker loops dispatch through the same
  core abstraction.
• Plugins remain unaware of WANT_READ, WANT_WRITE, and retry sentinels.
• Shared async I/O switches the descriptor to the required readiness mask and
  restores its original type, mask, priority, and handler after completion.
• Timeout, pause, shutdown, and connection release wake suspended callbacks and
  unwind the wrapper, event, TLS session, and coroutine safely.
• Reentrant release from inside an active callback is deferred until the
  callback returns, preventing use-after-free during backpressure pause.
• Registration is limited to stream transports; datagram semantics are
  intentionally outside this interface.

There is no retry cap when net.io_timeout is zero. The coroutine is dormant
until descriptor readiness or cancellation, so indefinite waits do not spin.
Configured idle timeouts wake and terminate the pending operation.

Forward integration

Forward TLS and plain TCP now use the downstream dispatch interface. Unrelated
server inputs remain unchanged and can migrate separately after this core path
is proven.

The plain-TCP event ownership changes internally, but its protocol behavior and
configuration remain compatible.

Coroutine stack safety

The default coroutine stack now has a 64 KiB minimum while preserving explicit
FLB_CORO_STACK_SIZE overrides and larger platform defaults.

The former glibc-derived default could be only 24 KiB. That is insufficient
headroom for the full parser/processor ingestion path now reachable from a
downstream callback; historical yyjson parser frames alone have approached
38 KiB. The same safe default applies consistently to upstream, downstream,
input, output, and scheduler coroutines.

On systems that previously selected 24 KiB, the cost is approximately 40 KiB
of additional virtual stack reservation per coroutine.

Integration coverage

The focused Forward matrix contains 24 downstream-coroutine scenarios, each
run normally and under strict Valgrind:

• TLS and plain TCP
• engine, threaded-input, and four-worker dispatch
• partial data and listener responsiveness while a callback is suspended
• completion and reuse of the same suspended connection
• EOF, Forward protocol errors, and TLS handshake/protocol errors
• idle timeout with connection-drop proof before shutdown
• pause/backpressure unwind with connection-drop proof before shutdown
• shutdown with a suspended coroutine

The timeout and pause tests assert that the connection wrapper is dropped
before process shutdown, so they cannot pass by leaking until teardown.

tests/integration/.venv/bin/python -m pytest \
  tests/integration/scenarios/in_forward/tests/test_in_forward_001.py \
  -q -k 'downstream_coro'

Result across the final matrix: 24 passed.

VALGRIND=1 VALGRIND_STRICT=1 \
  tests/integration/.venv/bin/python -m pytest \
  tests/integration/scenarios/in_forward/tests/test_in_forward_001.py \
  -q -k 'downstream_coro'

Result across the final matrix: 24 passed with strict Valgrind.

Core validation

cmake -S . -B build -DFLB_TESTS_RUNTIME=On -DFLB_TESTS_INTERNAL=On
cmake --build build -j8
ctest --test-dir build \
  -R '^(flb-rt-in_forward|flb-it-network|flb-it-downstream_worker|flb-it-upstream_tls)$' \
  --output-on-failure

flb-it-network, flb-it-downstream_worker, and flb-it-upstream_tls
passed. flb-rt-in_forward could not bind port 24224 because the host
fluentd.service owns it; that service was not disrupted.

The full PR range passes commit-prefix lint, DCO, and git diff --check.
Windows x86, x64, and ARM64 builds are covered by PR CI because a local MSVC
toolchain is unavailable.

Compatibility

• No configuration properties or defaults change.
• net.io_timeout: 0 retains indefinite, readiness-driven waiting.
• Forward plain and TLS protocol behavior is preserved.
• Existing downstream plugins are not migrated implicitly.
• Explicit coroutine stack-size overrides remain authoritative.

Related

• Replaces tls: Fix infinite retry loops in flb_tls_net_read/write #11547
• Fixes Stuck TLS Connection Hangs Fluent-bit and Consumes 100% CPU #11551

Summary by CodeRabbit

• Bug Fixes
  • Improved resilience of Forward input connections during partial records, EOFs, TLS errors, pauses, shutdowns, and worker timeouts.
  • Improved cleanup and recovery of stalled or interrupted network connections, including safer async event restoration and more consistent teardown behavior.
  • Reduced the risk of listener unresponsiveness during connection recovery.
• New Features
  • Added support for connection-level event callbacks and per-connection flag management to enable async event-driven handling.
• Tests
  • Added/expanded integration coverage for threaded, worker-based, paused, timed-out, and TLS-enabled Forward scenarios.

[coderabbitai]

No actionable comments were generated in the recent review. 🎉

ℹ️ Recent review info

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro Plus

Run ID: 6fa8b9af-5ed3-40bd-9ea4-cf180209d30d

📥 Commits

Reviewing files that changed from the base of the PR and between 8078710 and ede95a2.

📒 Files selected for processing (21)

• include/fluent-bit/flb_connection.h
• include/fluent-bit/flb_coro.h
• plugins/in_forward/fw.c
• plugins/in_forward/fw_conn.c
• src/flb_connection.c
• src/flb_coro.c
• src/flb_downstream.c
• src/flb_downstream_worker.c
• src/flb_engine.c
• src/flb_input_thread.c
• src/flb_io.c
• src/tls/flb_tls.c
• tests/integration/scenarios/in_forward/config/in_forward_pause.yaml
• tests/integration/scenarios/in_forward/config/in_forward_threaded.yaml
• tests/integration/scenarios/in_forward/config/in_forward_threaded_pause.yaml
• tests/integration/scenarios/in_forward/config/in_forward_tls_pause.yaml
• tests/integration/scenarios/in_forward/config/in_forward_tls_threaded.yaml
• tests/integration/scenarios/in_forward/config/in_forward_tls_threaded_pause.yaml
• tests/integration/scenarios/in_forward/config/in_forward_tls_workers_timeout.yaml
• tests/integration/scenarios/in_forward/config/in_forward_workers_timeout.yaml
• tests/integration/scenarios/in_forward/tests/test_in_forward_001.py

🚧 Files skipped from review as they are similar to previous changes (20)

• tests/integration/scenarios/in_forward/config/in_forward_threaded_pause.yaml
• tests/integration/scenarios/in_forward/config/in_forward_threaded.yaml
• tests/integration/scenarios/in_forward/config/in_forward_workers_timeout.yaml
• tests/integration/scenarios/in_forward/config/in_forward_tls_threaded_pause.yaml
• tests/integration/scenarios/in_forward/config/in_forward_tls_pause.yaml
• tests/integration/scenarios/in_forward/config/in_forward_tls_workers_timeout.yaml
• tests/integration/scenarios/in_forward/config/in_forward_pause.yaml
• include/fluent-bit/flb_coro.h
• src/flb_engine.c
• tests/integration/scenarios/in_forward/config/in_forward_tls_threaded.yaml
• plugins/in_forward/fw.c
• src/flb_connection.c
• include/fluent-bit/flb_connection.h
• src/flb_input_thread.c
• src/flb_downstream_worker.c
• src/flb_io.c
• plugins/in_forward/fw_conn.c
• src/flb_downstream.c
• src/tls/flb_tls.c
• tests/integration/scenarios/in_forward/tests/test_in_forward_001.py

---

📝 Walkthrough

Walkthrough

Adds downstream-owned event coroutines and deferred connection teardown for Forward TLS connections. Engine and worker dispatch resume these coroutines, asynchronous I/O propagates network and restoration errors, and integration tests cover partial records, shutdown, protocol errors, pause handling, and worker timeouts.

Changes

TLS event coroutine lifecycle

Layer / File(s)	Summary
Connection event contracts include/fluent-bit/flb_connection.h, include/fluent-bit/flb_downstream.h, include/fluent-bit/flb_coro.h, src/flb_connection.c, src/flb_coro.c	Adds event callback state, flag helpers, release-status constants, event registration APIs, and minimum coroutine stack sizing.
Downstream coroutine lifecycle src/flb_downstream.c	Adds callback coroutines and coordinates event wakeups, deferred release, timeout cancellation, forced teardown, and pending destruction.
Event dispatch and Forward integration plugins/in_forward/fw.c, plugins/in_forward/fw_conn.c, src/flb_downstream_worker.c, src/flb_engine.c, src/flb_input_thread.c	Routes Forward connections through downstream event registration, updates pause cleanup, and resumes event coroutines from engine and worker events.
Asynchronous network error handling src/flb_io.c, src/tls/flb_tls.c	Stops asynchronous I/O retries after network errors, checks event switching and restoration failures, and restores coroutine state before returning failure.
Forward lifecycle integration coverage tests/integration/scenarios/in_forward/config/*, tests/integration/scenarios/in_forward/tests/test_in_forward_001.py	Adds scenarios and tests for partial records, EOF, protocol errors, shutdown, pause unwinding, and worker timeouts.

Estimated code review effort: 4 (Complex) | ~60 minutes

Sequence Diagram(s)

sequenceDiagram
  participant TLSClient
  participant fw_conn_add
  participant Downstream
  participant TLSIO
  participant Output
  TLSClient->>fw_conn_add: establish TLS connection
  fw_conn_add->>Downstream: register event callback
  Downstream->>TLSIO: resume asynchronous read/write
  TLSIO-->>Downstream: deliver data or net_error
  Downstream->>Output: forward decoded records
  Downstream-->>TLSClient: close or wake timed-out connection

Loading

Possibly related PRs

• fluent/fluent-bit#11114: Modifies Forward connection lifetime handling during event and pause cleanup.
• fluent/fluent-bit#11775: Adds downstream worker event-dispatch infrastructure used to resume connection coroutines.
• fluent/fluent-bit#12044: Modifies downstream connection destruction and deferred cleanup around queued events.

Suggested labels: bug

Suggested reviewers: cosmo0920

🚥 Pre-merge checks | ✅ 4 | ❌ 1

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 11.76% which is insufficient. The required threshold is 80.00%.	Write docstrings for the functions missing them to satisfy the coverage threshold.

✅ Passed checks (4 passed)

Check name	Status	Explanation
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The title is concise and accurately summarizes the main change: coroutine-based event dispatch in downstream.
Linked Issues check	✅ Passed	The changes add coroutine-based downstream event handling and Forward coverage that directly address the stuck TLS/CPU hang in #11551.
Out of Scope Changes check	✅ Passed	The added core, API, and test changes all support coroutine-based event dispatch and the Forward TLS fix, with no clear unrelated scope creep.

✨ Finishing Touches

📝 Generate docstrings

• Create stacked PR
• Commit on current branch

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch tls-downstream-coro-11551

---

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.}

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: c08d3a6339

ℹ️ About Codex in GitHub

Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".

[chatgpt-codex-connector]

Avoid resuming worker coroutines from the engine thread

When a timed-out downstream belongs to a threaded input or downstream worker, this branch resumes its connection coroutine directly from whichever thread called flb_downstream_conn_timeouts_stream(). That function is also reached by the main engine timer via flb_downstream_conn_timeouts(&ctx->downstreams), and worker/threaded downstreams are inserted into that global list during setup, so a Forward TLS worker timeout can resume a coroutine and touch its worker event-loop state from the engine thread instead of the owning worker/input thread. Please only do the direct resume when the caller is the stream's owner, or wake the owner event loop instead.

Useful? React with 👍 / 👎.

[coderabbitai]

🧹 Nitpick comments (1)

plugins/in_forward/fw_conn.c (1)

253-258: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value

Use the required Allman brace style in edited control blocks.

• plugins/in_forward/fw_conn.c#L253-L258: move if/else opening braces to their own lines.
• plugins/in_forward/fw_conn.c#L360-L362: move the if opening brace to its own line.
• src/flb_downstream_worker.c#L213-L225: update the if/else blocks.
• src/flb_engine.c#L1295-L1300: update the if/else blocks.
• src/flb_input_thread.c#L462-L467: update the if/else blocks.
• src/tls/flb_tls.c#L414-L417, #L429-L434, #L533-L538, #L549-L554, #L579-L584: update each error branch.

Proposed pattern

-    if (condition) {
+    if (condition)
+    {

As per coding guidelines, C control-block opening braces must be on the next line.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@plugins/in_forward/fw_conn.c` around lines 253 - 258, Apply Allman brace
formatting to the edited control blocks: place every if and else opening brace
on its own line at plugins/in_forward/fw_conn.c lines 253-258 and 360-362;
src/flb_downstream_worker.c lines 213-225; src/flb_engine.c lines 1295-1300;
src/flb_input_thread.c lines 462-467; and each listed error branch in
src/tls/flb_tls.c lines 414-417, 429-434, 533-538, 549-554, and 579-584.
Preserve all existing control flow and behavior.

Source: Coding guidelines

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Nitpick comments:
In `@plugins/in_forward/fw_conn.c`:
- Around line 253-258: Apply Allman brace formatting to the edited control
blocks: place every if and else opening brace on its own line at
plugins/in_forward/fw_conn.c lines 253-258 and 360-362;
src/flb_downstream_worker.c lines 213-225; src/flb_engine.c lines 1295-1300;
src/flb_input_thread.c lines 462-467; and each listed error branch in
src/tls/flb_tls.c lines 414-417, 429-434, 533-538, 549-554, and 579-584.
Preserve all existing control flow and behavior.

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro Plus

Run ID: 5bf89f17-3a58-413a-abb0-45e6851bc91b

📥 Commits

Reviewing files that changed from the base of the PR and between cb7256c and c08d3a6.

📒 Files selected for processing (13)

• include/fluent-bit/flb_connection.h
• include/fluent-bit/flb_downstream.h
• plugins/in_forward/fw_conn.c
• src/flb_connection.c
• src/flb_downstream.c
• src/flb_downstream_worker.c
• src/flb_engine.c
• src/flb_input_thread.c
• src/tls/flb_tls.c
• tests/integration/scenarios/in_forward/config/in_forward_tls_pause.yaml
• tests/integration/scenarios/in_forward/config/in_forward_tls_threaded.yaml
• tests/integration/scenarios/in_forward/config/in_forward_tls_workers_timeout.yaml
• tests/integration/scenarios/in_forward/tests/test_in_forward_001.py