SDK: hand off the plugin-runtime worker during blocking source send callbacks
- Dominant language
- Rust
- Stars
- 4.9k
- Forks
- 432
- Avg merge
- 2d 10h
- Merged PRs (30d)
- 173
Description
## Context
#3795 bounded the source forwarding channel. When the channel is full, the runtime's send callback (`handle_produced_messages` -> `send_with_backpressure` in `core/connectors/runtime/src/source.rs`) parks in bounded `send_timeout` waits until capacity frees, the connector shuts down, or the channel disconnects. That park is the backpressure mechanism and is intentional.
The callback runs synchronously inside the SDK's polling loop (`core/connectors/sdk/src/source.rs`, the `callback(plugin_id, messages.as_ptr(), messages.len())` call in `handle_messages`), which is spawned on the plugin library's process-global tokio runtime (`core/connectors/sdk/src/lib.rs`, `static RUNTIME: OnceLock`). Every instance loaded from the same `.so` shares that runtime's workers.
## Problem
A parked callback holds one worker for the duration of its backpressure episode. With enough saturated instances of one plugin library, all workers can be occupied, and `iggy_source_close` for a sibling instance is delayed until a channel drains: the close blocks on the sibling's polling task, which needs a free worker to observe its shutdown signal. `signal_shutdown_all()` covers process shutdown (added in #3795), but the runtime API path that stops a single connector has no bound when its same-library siblings are saturated. The limitation is documented in the `send_with_backpressure` comment and in the connector-runtime skill.
## Proposed fix
Hand the worker off in the SDK before invoking the callback:
```rust
// core/connectors/sdk/src/source.rs, inside handle_messages
tokio::task::block_in_place(|| callback(plugin_id, messages.as_ptr(), messages.len()));
```
Notes:
- The wrap must 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 multi-thread (`Runtime::new()`), 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.
## Alternatives considered (in #3795 review)
- Signaling all same-library siblings during a single-connector stop: unblocks the close but drops sibling batches that are merely backpressured, punishing healthy instances.
- A wall-clock deadline on the retry loop: bounds the park but converts sustained backpressure into data loss, which defeats the purpose of the bounded channel.
Refs: #3795, discussion #3039 (bounded-channel spec).
Contributor guide
Research direction
Start in core/connectors/sdk/src/source.rs at handle_messages and read core/connectors/sdk/src/lib.rs to understand the process-global Tokio runtime. Trace the callback path into core/connectors/runtime/src/source.rs, then verify that a saturated instance no longer prevents a sibling connector's close from being observed while backpressure remains intact.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- rust
- Domain
- backend, distributed-systems
- Issue type
- Bug
- Difficulty
- 3/5
- Estimated time
- 1-2 days
- Activity status
- Quiet
- Clarity
- Clearly specified
- Newbie friendliness
- 72/100