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

[mlevkov]

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<Runtime>). 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:

// 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).