apache / apache/iggy

SDK: hand off the plugin-runtime worker during blocking source send callbacks

Open
#3,796 0 comments 0 reactions 0 assignees View on GitHub
bug connectors rust
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.