mlevkov opened a new pull request, #3795: URL: https://github.com/apache/iggy/pull/3795
## Summary The channel between a source plugin's send callback and the runtime's forwarding loop was `flume::unbounded()`, so a slow or hung Iggy meant batches accumulated in memory without bound instead of propagating backpressure into the plugin's polling loop. This is the prerequisite runtime fix requested in the HTTP source discussion (#3039), and it applies to every source connector, including the source PRs currently in flight. ## What changed - The forwarding channel is now a bounded crossfire channel (`crossfire::mpsc::bounded_blocking_async`), the same shape `shard` and `server-ng` use. flume is no longer a runtime dependency. - Capacity comes from a new optional `SourceConfig` field, `channel_capacity`, counted in batches (a single batch can be megabytes), defaulting to 1024 and clamped to [1, 65536] since crossfire eagerly allocates the ring and asserts capacity < 2^31. The existing `ConfigEnv` derive provides `IGGY_CONNECTORS_SOURCE_<KEY>_CHANNEL_CAPACITY`; configs without the field behave as before apart from the bound. - The FFI send callback applies backpressure with a `try_send` fast path and a `send_timeout(10ms)` retry loop that re-reads a per-instance shutdown flag between waits. The manager sets that flag before `iggy_source_close` so a hung Iggy cannot deadlock the close. Process shutdown sets every instance's flag (`signal_shutdown_all`) before the sequential stops, because instances loaded from one plugin library share a single tokio runtime and a wedged sibling would otherwise hold a worker an earlier close needs. - A full channel logs one `warn!` per backpressure episode (latched, cleared on genuine recovery). A batch that still cannot be enqueued after the stop signal is dropped and counted in `iggy_connector_errors_total`. - Unit tests pin the behavior shutdown relies on: buffered batches drain after the senders drop (crossfire's docs do not promise this), the retry loop unblocks when the flag flips mid-backoff, and shutdown drops are counted without being enqueued. - Docs updated across the runtime README, sources README, and the connector skills, which still described flume and the pre-#3321 shutdown order. ## One correction to the discussion notes @hubcio the spec assumed crossfire's blocking sender has no `send_timeout`. It does: `blocking_tx.rs:288` on `Tx`, reachable from `MTx` via `Deref`. The loop is built on it instead of `try_send` plus sleep, so the sender wakes as soon as capacity frees while shutdown latency stays bounded by the retry interval. ## Known limitation Stopping a single connector via the runtime API while enough same-library sibling instances are saturated can delay that close until the siblings drain, because the callback parks a worker of the shared plugin runtime. The code comment and the connector skill document this. The complete fix is an SDK-side worker handoff (`tokio::task::block_in_place` around the callback invocation); happy to file it as a follow-up issue. ## Test plan - `cargo clippy -p iggy-connectors --all-targets -- -D warnings` clean - `cargo test -p iggy-connectors`: 128 passed, including the new channel and shutdown tests - `cargo build -p iggy_connector_stdout_sink -p iggy_connector_random_source` - `cargo test -p integration -- connectors::runtime::` could not run on this machine (`hwlocality-sys` needs `pkg-config`); relying on CI for the integration suite -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
