fix(connectors): hand off SDK worker during blocking send callbacks

[mlevkov]

Summary

Closes #3796. The send callback runs synchronously inside the SDK's polling task (handle_messages in core/connectors/sdk/src/source.rs), on the tokio runtime shared by every connector instance loaded from the same plugin library (static RUNTIME: OnceLock<Runtime> in lib.rs). A callback that blocks pins one worker for the duration. With #3795's bounded forwarding channel, a callback legitimately blocks for backpressure while the channel is full, so enough saturated instances of one library can occupy all workers, and iggy_source_close for a sibling instance then waits behind them: the close blocks on the sibling's polling task getting scheduled to observe its shutdown signal.

Change

One call site: wrap the callback invocation in tokio::task::block_in_place, so the worker is handed off before the callback runs and the runtime keeps scheduling sibling tasks regardless of how long the callback blocks.

Notes from the analysis in #3796:

• The wrap has to live in the SDK, not the runtime: block_in_place consults the calling thread's tokio context, which is only set on the plugin runtime's own worker threads.
• The SDK runtime is Runtime::new(), i.e. multi-threaded, which block_in_place requires.
• No FFI or ABI change. Existing plugin binaries keep working and pick the fix up when rebuilt against the updated SDK.

This composes with #3795 (bounded channel + shutdown signaling) but does not depend on it: any long-blocking callback benefits.

Test plan

• cargo clippy -p iggy_connector_sdk --all-targets --all-features -- -D warnings clean
• cargo test -p iggy_connector_sdk --all-features passes
• cargo build -p iggy_connector_stdout_sink -p iggy_connector_random_source -p iggy-connectors (macro consumers + runtime rebuild cleanly)
• Behavior under saturation is exercised end to end by the integration suite in CI

[github-actions]

Thanks for the PR. It is labeled S-waiting-on-review and queued for review.

Slash commands (own line, regular comment) move it around the queue:

• /ready - back to S-waiting-on-review after addressing feedback
• /author - flip to S-waiting-on-author while you finish changes
• /request-review @user-or-team - request a reviewer

See CONTRIBUTING.md for details.

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 68.19%. Comparing base (be5012e) to head (50ef34d).

Additional details and impacted files

@@              Coverage Diff              @@
##             master    #3797       +/-   ##
=============================================
- Coverage     82.99%   68.19%   -14.80%     
  Complexity     1339     1339               
=============================================
  Files          1218     1218               
  Lines        165627   139885    -25742     
  Branches     133940   108198    -25742     
=============================================
- Hits         137459    95398    -42061     
- Misses        24509    40788    +16279     
- Partials       3659     3699       +40     

Components	Coverage Δ
Rust Core	64.19% <100.00%> (-19.32%)	⬇️
Java SDK	66.55% <ø> (ø)
C# SDK	76.43% <ø> (ø)
Python SDK	90.00% <ø> (ø)
PHP SDK	84.48% <ø> (ø)
Node SDK	95.68% <ø> (ø)
Go SDK	69.04% <ø> (ø)

Files with missing lines	Coverage Δ
core/connectors/sdk/src/source.rs	82.55% <100.00%> (-1.13%)	⬇️

... and 336 files with indirect coverage changes

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[github-actions]

/request-review: could not request mlevkov - a handle is not a repo collaborator, or the team is unknown.

[mlevkov]

/request-review @hubcio

[github-actions]

This pull request has been automatically marked as stale because it has not had recent activity. It will be closed in 7 days if no further activity occurs.

If you need a review, please ensure CI is green and the PR is rebased on the latest master. Don't hesitate to ping the maintainers - either @core on Discord or by mentioning them directly here on the PR.

Thank you for your contribution!

[mlevkov]

Holding this pending a decision on #3795, because af9ce9548 (#3855) removed its justification.

The comment this PR adds says the callback can block "while its bounded forwarding channel is full". On master c0c74931b that channel is flume::unbounded() (runtime/src/source.rs:625), so the send never parks. handle_messages also awaits the batch-result oneshot immediately after the callback returns (sdk/src/source.rs:278), which yields the worker at that point anyway, so block_in_place buys nothing on top of it.

The backpressure this protects against exists only if #3795's bounded channel lands, and I have argued over there that the bound is no longer needed. The SendCallback signature also gained batch_id, which is the conflict showing here.

Separately, the discipline this PR was arguing for has already landed in the direction that genuinely blocks: the runtime invokes batch_result_callback inside tokio::task::spawn_blocking (runtime/src/source.rs:559) because the plugin runs its async on_batch_result to completion inside that FFI call.

So my read is that this closes along with #3795 unless you want the bound, and I am happy to reopen if the bounded channel comes back. Should #3796 close too, or be reframed around the 30s timeout case I described in #3795?

[mlevkov]

Closing this alongside #3795, per your call on Discord.

Without the bounded channel there is no backpressure for block_in_place to hand off: master's flume::unbounded() send never parks, and handle_messages awaits the batch-result oneshot immediately after the callback returns, which yields the worker at that point anyway. The discipline this PR was arguing for also already landed in the direction that genuinely blocks, since the runtime invokes batch_result_callback inside tokio::task::spawn_blocking because the plugin runs its async on_batch_result to completion inside that FFI call.

#3796 is still open. Happy to close it too, or reframe it around the 30s timeout case, whichever you prefer.