diff --git a/.claude/skills/connector-runtime/SKILL.md b/.claude/skills/connector-runtime/SKILL.md index ce194c08e6..9b52c04163 100644 --- a/.claude/skills/connector-runtime/SKILL.md +++ b/.claude/skills/connector-runtime/SKILL.md @@ -44,7 +44,7 @@ lifecycle, ferries messages between Apache Iggy and plugins, exposes runtime/src/ ├── main.rs Entry, plugin path resolution, SourceApi/SinkApi FFI structs ├── sink.rs Sink lifecycle, Iggy consumer wiring, FFI consume calls -├── source.rs Source lifecycle, flume forwarding, state save loop +├── source.rs Source lifecycle, bounded crossfire forwarding, state save loop ├── stream.rs Stream + consumer/producer setup ├── transform.rs Loads transforms from config, applies them in chain ├── state.rs FileStateProvider, atomic ConnectorState file I/O @@ -57,7 +57,7 @@ runtime/src/ ├── manager/{mod,sink,source}.rs SinkManager / SourceManager, status, restart_guard ├── configs/ │ ├── runtime.rs RuntimeConfig + LoggingConfig + LogFormat -│ ├── connectors.rs ConnectorConfig, SinkConfig, SourceConfig (verbose + benchmark) +│ ├── connectors.rs ConnectorConfig, SinkConfig, SourceConfig (verbose + benchmark + channel_capacity) │ └── connectors/{local,http}_provider.rs └── api/ HTTP control + observability endpoints ``` @@ -109,19 +109,24 @@ Don't mix. 1. `iggy_source_handle(id, send_callback)` - plugin registers itself. 2. Plugin polls + invokes `send_callback(plugin_id, ptr, len)`. -3. Callback runs in the SDK macro's spawned async task. Pushes postcard `ProducedMessages` into a `flume` channel keyed by `plugin_id` in `SOURCE_SENDERS: Lazy>` (`pub(crate)`). `SourceSenderEntry` wraps the sender + a pre-extracted owned `Counter` (the `errors` series, `Arc` inside). The FFI callback bumps errors on deserialize or channel-closed failure with one relaxed atomic - no `Family` lookup, no `Arc` handle. +3. Callback runs in the SDK macro's spawned async task. Pushes postcard `ProducedMessages` into a **bounded crossfire channel** (`crossfire::mpsc::bounded_blocking_async`, capacity = `SourceConfig::channel_capacity` in batches, default 1024, clamped to [1, 65536]) keyed by `plugin_id` in `SOURCE_SENDERS: Lazy>` (`pub(crate)`). + - `SourceSenderEntry` wraps the sender + a pre-extracted owned `Counter` (the `errors` series, `Arc` inside) + two `Arc`s (`shutdown`, `backpressure_active` - the latter latches the channel-full `warn!` to one per backpressure episode). + - The callback clones the fields out and drops the DashMap guard before sending. Holding the shard guard through a stall would block `cleanup_sender`. + - A full channel makes `send_with_backpressure` retry via `send_timeout(SEND_RETRY_INTERVAL)` - the stall IS the backpressure into the plugin's polling loop. The batch is dropped (+1 `errors`) only on disconnect or when the shutdown flag is observed while full. 4. `source_forwarding_loop` pulls from the channel, deserializes, applies transforms, encodes via `StreamEncoder`, sends to Iggy producer. 5. On success, save returned `ConnectorState` via `FileStateProvider`. **Shutdown ordering (`manager/source.rs::stop_connector`):** -1. Call `iggy_source_close` FIRST. It blocks until the plugin's polling task stops, so no new send callbacks fire after it returns. -2. `cleanup_sender(plugin_id)` NEXT - dropping the channel sender makes the forwarding task's `recv_async()` resolve with `Disconnected` and exit cleanly, instead of blocking until the abort timeout. -3. Finally await spawned handlers with `tokio::time::timeout`. On timeout, `handle.abort()` + drain - prevents leaked tasks colliding with the next `start_connector` (a late `file.save()` could otherwise race the new instance). The silent-drop branch in `handle_produced_messages` only covers the window between close and cleanup. +1. `signal_shutdown(plugin_id)` FIRST - sets the entry's shutdown flag so a send callback parked in the full-channel retry loop unblocks; `iggy_source_close` waits on the polling task that loop runs in, so skipping this can deadlock the close when Iggy is hung. (Process shutdown in `main.rs` calls `signal_shutdown_all()` before the sequential per-connector stops, because same-`.so` instances share one plugin runtime.) +2. Call `iggy_source_close` NEXT. It blocks until the plugin's polling task stops, so no new send callbacks fire after it returns. +3. `cleanup_sender(plugin_id)` NEXT - dropping the channel sender makes the forwarding task's `recv()` resolve with `Disconnected` and exit cleanly, instead of blocking until the abort timeout. +4. Finally await spawned handlers with `tokio::time::timeout`. On timeout, `handle.abort()` + drain - prevents leaked tasks colliding with the next `start_connector` (a late `file.save()` could otherwise race the new instance). The silent-drop branch in `handle_produced_messages` only covers the window between close and cleanup. Gotchas: - `SOURCE_SENDERS` must be cleaned up on connector close or memory leaks (channel + task). +- `send_with_backpressure` parks a worker of the plugin library's shared tokio runtime while the channel is full (bounded per park by `SEND_RETRY_INTERVAL`). All instances of one `.so` share that runtime, so saturated siblings can delay another instance's close; the SDK-side worker handoff (`block_in_place`) is the known follow-up. - `spawn_source_handler` wraps the outer `iggy_source_handle(id, callback)` FFI call in `tokio::task::spawn_blocking()`. That call returns quickly - the SDK macro internally `runtime.spawn`s an async `handle_messages` task and returns. **`send_callback` invocations come from that async task, not from `spawn_blocking`.** A long-running synchronous poll inside the plugin would block one Tokio worker. async polls don't. - No timeout on the registration call - a plugin whose `iggy_source_handle` never returns stalls one blocking worker for the process lifetime. @@ -183,7 +188,9 @@ New per-batch logging follows the same pattern. Default to `debug!`, upgrade to ### Env-var overrides via `ConfigEnv` derive -`SinkConfig`, `SourceConfig`, and inner structs derive `ConfigEnv` (`configs_derive::ConfigEnv`). Generates env-var addressability as `IGGY_CONNECTORS___` for primitive fields. Used heavily by integration tests to inject testcontainer ports - see `core/integration/tests/connectors/fixtures/postgres/container.rs` for env-var constants. Mark new compound fields `#[config_env(skip)]`, leaf primitives `#[config_env(leaf)]`. +`SinkConfig`, `SourceConfig`, and inner structs derive `ConfigEnv` (`configs_derive::ConfigEnv`). Generates env-var addressability as `IGGY_CONNECTORS___` for primitive fields. Used heavily by integration tests to inject testcontainer ports - see `core/integration/tests/connectors/fixtures/postgres/container.rs` for env-var constants. + +Mark new compound fields `#[config_env(skip)]`; non-primitive leaves (enums, `PathBuf`) need `#[config_env(leaf)]`, while plain primitives and `Option`-of-primitive (e.g. `channel_capacity: Option`) are auto-detected. ### Versioning @@ -235,17 +242,17 @@ Per-message drops in the batch loops are counted into a local `u64` and flushed - `sink.rs::spawn_consume_tasks` task wrapper - bumps once on `consume_messages` Err - `source.rs::source_forwarding_loop` - payload decode, prepare (transform/encode) failure, Iggy send Err, state save Err - `source.rs::process_messages` - transform Err (logs + bumps `errors` + continue. does NOT propagate, so one bad payload doesn't flip the connector to permanent ERROR), transform encode failure, `build_iggy_message` failure -- `source.rs::handle_produced_messages` - postcard deserialize failure, `sender.send` channel-closed +- `source.rs::handle_produced_messages` / `send_with_backpressure` - postcard deserialize failure, channel disconnected, channel-still-full-at-shutdown drop (all one `inc()` per batch) Filter case bumps `messages_filtered` via `inc_messages_filtered_with_labels`. Adding a new drop path: mirror this pattern. ## Hard rules -1. **Never block the executor.** All I/O async. `spawn_blocking` only for FFI registration (already in place). +1. **Never block the executor.** All I/O async. `spawn_blocking` only for FFI registration (already in place). One documented exception: `send_with_backpressure` parks a plugin-runtime worker in bounded `send_timeout` waits - that stall is the source backpressure mechanism. 2. **Plugin ID counter is monotonic.** No reset, no reuse. 3. **FFI return codes:** `0` success, non-zero failure. 4. **Don't add static mutable state** beyond `LOG_CALLBACK`, `PLUGIN_ID`, `SOURCE_SENDERS`. -5. **Pair `cleanup_sender(id)` with shutdown** for sources (avoid flume leak). Order: close FFI -> cleanup sender -> drain/abort tasks. +5. **Pair `cleanup_sender(id)` with shutdown** for sources (avoid channel leak). Order: signal shutdown -> close FFI -> cleanup sender -> drain/abort tasks. 6. **Restart uses `restart_guard.try_lock()`** - no thundering-herd regression. 7. **No timeouts on plugin FFI calls without a kill-task strategy.** A timeout that returns from the runtime but leaves the plugin running has the worst of both worlds. diff --git a/.claude/skills/connectors-overview/SKILL.md b/.claude/skills/connectors-overview/SKILL.md index 91ff86ec7c..95bfae52c5 100644 --- a/.claude/skills/connectors-overview/SKILL.md +++ b/.claude/skills/connectors-overview/SKILL.md @@ -190,7 +190,7 @@ Each implemented in at least one in-tree plugin or runtime path. | Duplicate-ID FFI guard | `sdk/src/sink.rs::sink_connector!`, `sdk/src/source.rs::source_connector!` | Prevents silent data loss on reopen-without-close | | `restart_guard.try_lock()` | `runtime/src/manager/{sink,source}.rs::restart_connector` | No thundering-herd restarts | | `tokio::time::timeout(..., handle).await` + `handle.abort()` on timeout | `runtime/src/manager/source.rs::stop_connector` | Bounded shutdown + leak prevention | -| `flume::unbounded()` channel | `runtime/src/source.rs::spawn_source_handler` / `source_forwarding_loop` | MPSC handoff from SDK async task to runtime loop | +| `crossfire::mpsc::bounded_blocking_async` channel | `runtime/src/source.rs::spawn_source_handler` / `source_forwarding_loop` | Bounded MPSC handoff with backpressure to plugin | | `tokio::sync::watch::channel(())` | `sdk/src/{sink,source}.rs`, `runtime/src/sink.rs`, `runtime/src/manager/*` | One-shot shutdown broadcast | | `dashmap::DashMap` | `runtime/src/manager/sink.rs`, `source.rs::SOURCE_SENDERS`, SDK `INSTANCES` | Lock-free concurrent keyed access | | `secrecy::SecretString` + `iggy_common::serde_secret::serialize_secret` | `sinks/postgres_sink::PostgresSinkConfig::connection_string` | Auto-redact on Debug/Display + serialization | diff --git a/Cargo.lock b/Cargo.lock index 0cf2039bd2..c6d381c5c3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6756,6 +6756,7 @@ dependencies = [ "clap", "configs", "configs_derive", + "crossfire", "dashmap", "derive_more", "dirs", @@ -6763,7 +6764,6 @@ dependencies = [ "dotenvy", "figlet-rs", "figment", - "flume", "futures", "iggy", "iggy_common", diff --git a/core/connectors/runtime/Cargo.toml b/core/connectors/runtime/Cargo.toml index 9533516678..dbb99c52cb 100644 --- a/core/connectors/runtime/Cargo.toml +++ b/core/connectors/runtime/Cargo.toml @@ -39,6 +39,7 @@ axum-server = { workspace = true } clap = { workspace = true } configs = { workspace = true } configs_derive = { workspace = true } +crossfire = { workspace = true } dashmap = { workspace = true } derive_more = { workspace = true } dirs = { workspace = true } @@ -46,7 +47,6 @@ dlopen2 = { workspace = true } dotenvy = { workspace = true } figlet-rs = { workspace = true } figment = { workspace = true } -flume = { workspace = true } futures = { workspace = true } iggy = { workspace = true } iggy_common = { workspace = true } diff --git a/core/connectors/runtime/README.md b/core/connectors/runtime/README.md index 1c1339f49c..70f7218ba2 100644 --- a/core/connectors/runtime/README.md +++ b/core/connectors/runtime/README.md @@ -219,6 +219,21 @@ Emitted fields: Filter the stream via `RUST_LOG=iggy_connectors::benchmark=info`. The corresponding stage durations are also recorded in the `iggy_connector_stage_duration_seconds` histogram regardless of this flag, so Prometheus dashboards remain available without enabling text events. +## Source Channel Capacity + +Each source configuration accepts an optional `channel_capacity` setting that bounds the channel between the plugin's send callback and the runtime's forwarding loop. Capacity is counted in batches (one `poll()` result each, potentially megabytes), not messages or bytes. The default is 1024 batches. + +When the channel is full (Iggy accepts messages more slowly than the plugin produces them), the send callback backs off and retries instead of buffering without bound, so backpressure propagates into the plugin's polling loop. During shutdown, a batch that still cannot be enqueued after the stop signal is dropped and counted in `iggy_connector_errors_total`, so a saturated source may report errors at SIGTERM. Values outside `[1, 65536]` are clamped with a warning. + +```toml +type = "source" +key = "postgres" +# ... other fields ... +channel_capacity = 1024 +``` + +Environment override: `IGGY_CONNECTORS_SOURCE__CHANNEL_CAPACITY`. + ## Metrics The runtime exposes Prometheus-compatible metrics via the `/metrics` endpoint when enabled. The following metrics are available: diff --git a/core/connectors/runtime/src/configs/connectors.rs b/core/connectors/runtime/src/configs/connectors.rs index 24647e8703..ca82cc9db6 100644 --- a/core/connectors/runtime/src/configs/connectors.rs +++ b/core/connectors/runtime/src/configs/connectors.rs @@ -137,6 +137,8 @@ pub struct CreateSourceConfig { pub verbose: bool, #[serde(default)] pub benchmark: bool, + /// Forwarding channel capacity in batches; defaults to 1024. + pub channel_capacity: Option, } impl CreateSourceConfig { @@ -153,6 +155,7 @@ impl CreateSourceConfig { plugin_config: self.plugin_config.clone(), verbose: self.verbose, benchmark: self.benchmark, + channel_capacity: self.channel_capacity, } } } @@ -175,6 +178,10 @@ pub struct SourceConfig { pub verbose: bool, #[serde(default)] pub benchmark: bool, + /// Capacity of the plugin -> runtime forwarding channel, counted in + /// batches (a single batch can be megabytes), not messages or bytes. + /// Defaults to 1024; values are clamped to [1, 65536]. + pub channel_capacity: Option, } #[derive(Debug, Clone, Default, Serialize, Deserialize)] @@ -333,7 +340,7 @@ impl std::fmt::Display for SourceConfig { fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { write!( f, - "{{ enabled: {}, name: {}, path: {}, transforms: {:?}, streams: [{}], plugin_config_format: {:?}, verbose: {}, benchmark: {} }}", + "{{ enabled: {}, name: {}, path: {}, transforms: {:?}, streams: [{}], plugin_config_format: {:?}, verbose: {}, benchmark: {}, channel_capacity: {:?} }}", self.enabled, self.name, self.path, @@ -346,6 +353,7 @@ impl std::fmt::Display for SourceConfig { self.plugin_config_format, self.verbose, self.benchmark, + self.channel_capacity, ) } } diff --git a/core/connectors/runtime/src/main.rs b/core/connectors/runtime/src/main.rs index 5c5ebe7774..db384fdb57 100644 --- a/core/connectors/runtime/src/main.rs +++ b/core/connectors/runtime/src/main.rs @@ -265,6 +265,11 @@ async fn main() -> Result<(), RuntimeError> { } } + // Unwedge every send callback before the sequential closes: instances of + // one plugin library share a tokio runtime, and a backpressured instance + // late in the list would otherwise hold a worker an earlier close needs. + source::signal_shutdown_all(); + let source_keys: Vec = context .sources .get_all() @@ -452,6 +457,7 @@ struct SourceConnectorPlugin { error: Option, verbose: bool, benchmark: bool, + channel_capacity: Option, } struct SourceConnectorProducer { diff --git a/core/connectors/runtime/src/manager/source.rs b/core/connectors/runtime/src/manager/source.rs index b011472fe7..da01d756d8 100644 --- a/core/connectors/runtime/src/manager/source.rs +++ b/core/connectors/runtime/src/manager/source.rs @@ -147,8 +147,11 @@ impl SourceManager { ) }; - // Order: close FFI (stops callbacks) -> drop sender (unblocks - // recv_async) -> await tasks. Reversing risks an abort mid-save. + // Order: signal shutdown (unwedges a callback stuck in its + // full-channel backoff, which iggy_source_close waits on) -> close + // FFI (stops callbacks) -> drop sender (unblocks recv) -> await + // tasks. Reversing risks a deadlocked close or an abort mid-save. + source::signal_shutdown(plugin_id); if let Some(container) = &container { info!("Closing source connector with ID: {plugin_id} for plugin: {key}"); (container.iggy_source_close)(plugin_id); @@ -229,6 +232,7 @@ impl SourceManager { key, config.verbose, config.benchmark, + config.channel_capacity, producer, encoder, transforms, diff --git a/core/connectors/runtime/src/source.rs b/core/connectors/runtime/src/source.rs index e259e212cc..8f176157e4 100644 --- a/core/connectors/runtime/src/source.rs +++ b/core/connectors/runtime/src/source.rs @@ -15,9 +15,9 @@ // specific language governing permissions and limitations // under the License. +use crossfire::{AsyncRx, MTx, SendTimeoutError, TrySendError, mpsc::Array}; use dashmap::DashMap; use dlopen2::wrapper::Container; -use flume::{Receiver, Sender}; use iggy::prelude::{ DirectConfig, HeaderKey, HeaderValue, IggyClient, IggyDuration, IggyError, IggyMessage, IggyProducer, @@ -30,8 +30,11 @@ use iggy_connector_sdk::{ use std::{ collections::{BTreeMap, HashMap}, str::FromStr, - sync::{Arc, LazyLock, atomic::Ordering}, - time::Instant, + sync::{ + Arc, LazyLock, + atomic::{AtomicBool, Ordering}, + }, + time::{Duration, Instant}, }; use tracing::{debug, error, info, trace, warn}; @@ -50,11 +53,28 @@ use iggy_connector_sdk::api::ConnectorStatus; use prometheus_client::metrics::counter::Counter; use tokio::task::JoinHandle; +pub(crate) const DEFAULT_CHANNEL_CAPACITY: usize = 1024; + +// crossfire eagerly allocates the whole ring and asserts capacity < 2^31; +// the cap keeps a config typo from panicking the process or committing +// gigabytes up front. +const MAX_CHANNEL_CAPACITY: usize = 65_536; + +const SEND_RETRY_INTERVAL: Duration = Duration::from_millis(10); + +pub(crate) type BatchSender = MTx>; +pub(crate) type BatchReceiver = AsyncRx>; + pub(crate) struct SourceSenderEntry { - pub(crate) sender: Sender, + pub(crate) sender: BatchSender, // Owned errors counter (Arc inside) so the FFI callback bumps // it with one relaxed atomic - no Family RwLock + HashMap lookup per call. pub(crate) error_counter: Counter, + pub(crate) shutdown: Arc, + // Latched across callback invocations so sustained overload logs one + // warn per backpressure episode, not one per batch; cleared only by an + // uncontended fast-path send (i.e. genuine recovery). + pub(crate) backpressure_active: Arc, } pub(crate) static SOURCE_SENDERS: LazyLock> = @@ -64,6 +84,26 @@ pub(crate) fn cleanup_sender(plugin_id: u32) { SOURCE_SENDERS.remove(&plugin_id); } +/// Unwedges a send callback stuck in its full-channel backoff so +/// `iggy_source_close` (which blocks until the plugin's polling task exits) +/// cannot deadlock while the forwarding loop is itself blocked on a slow Iggy. +pub(crate) fn signal_shutdown(plugin_id: u32) { + if let Some(entry) = SOURCE_SENDERS.get(&plugin_id) { + entry.shutdown.store(true, Ordering::Release); + } +} + +/// Sets every live instance's shutdown flag. Process shutdown stops +/// connectors sequentially, and instances loaded from the same plugin +/// library share one tokio runtime - without this, an instance wedged in +/// backpressure later in the list keeps holding a runtime worker that an +/// earlier instance's close needs to observe its own shutdown. +pub(crate) fn signal_shutdown_all() { + for entry in SOURCE_SENDERS.iter() { + entry.shutdown.store(true, Ordering::Release); + } +} + /// Initializes all enabled source connectors. /// /// Per-connector failures (path resolution, dlopen, state load, plugin init, @@ -191,6 +231,7 @@ pub async fn init( error: init_error.clone(), verbose: config.verbose, benchmark: config.benchmark, + channel_capacity: config.channel_capacity, }); if let Some(error) = init_error { @@ -364,7 +405,7 @@ pub(crate) async fn source_forwarding_loop( encoder: Arc, transforms: Vec>, state_storage: StateStorage, - receiver: Receiver, + receiver: BatchReceiver, context: Arc, labels: Arc, ) { @@ -390,7 +431,7 @@ pub(crate) async fn source_forwarding_loop( topic: producer.topic().to_string(), }; - while let Ok(produced_messages) = receiver.recv_async().await { + while let Ok(produced_messages) = receiver.recv().await { let total_start = Instant::now(); let count = produced_messages.messages.len(); context @@ -554,6 +595,7 @@ pub(crate) fn spawn_source_handler( plugin_key: &str, verbose: bool, benchmark: bool, + channel_capacity: Option, producer: IggyProducer, encoder: Arc, transforms: Vec>, @@ -561,7 +603,15 @@ pub(crate) fn spawn_source_handler( callback: HandleCallback, context: Arc, ) -> Vec> { - let (sender, receiver) = flume::unbounded(); + let configured = channel_capacity.unwrap_or(DEFAULT_CHANNEL_CAPACITY); + let capacity = configured.clamp(1, MAX_CHANNEL_CAPACITY); + if capacity != configured { + warn!( + "Source connector with ID: {plugin_id} channel_capacity: {configured} clamped to {capacity}" + ); + } + let (sender, receiver) = crossfire::mpsc::bounded_blocking_async(capacity); + info!("Source connector with ID: {plugin_id} forwarding channel capacity: {capacity} batches"); let plugin_key = plugin_key.to_string(); let labels = Arc::new(SourceLabels::new(&plugin_key)); SOURCE_SENDERS.insert( @@ -569,6 +619,8 @@ pub(crate) fn spawn_source_handler( SourceSenderEntry { sender, error_counter: context.metrics.error_counter(&labels.counter), + shutdown: Arc::new(AtomicBool::new(false)), + backpressure_active: Arc::new(AtomicBool::new(false)), }, ); @@ -623,6 +675,7 @@ pub fn handle( &plugin_key, plugin.verbose, plugin.benchmark, + plugin.channel_capacity, producer_wrapper.producer, producer_wrapper.encoder, plugin.transforms, @@ -719,7 +772,18 @@ pub(crate) extern "C" fn handle_produced_messages( unsafe { // Entry missing = SOURCE_SENDERS cleaned up at shutdown; benign race // expected on stop/restart. No metric (would conflate with real failures). - let Some(entry) = SOURCE_SENDERS.get(&plugin_id) else { + // Clone out and drop the guard: the backoff loop below may run for a + // while, and holding the shard guard would block cleanup_sender. + let Some((sender, error_counter, shutdown, backpressure_active)) = + SOURCE_SENDERS.get(&plugin_id).map(|entry| { + ( + entry.sender.clone(), + entry.error_counter.clone(), + entry.shutdown.clone(), + entry.backpressure_active.clone(), + ) + }) + else { tracing::trace!( plugin_id, "dropping produced batch: sender already cleaned up" @@ -729,23 +793,102 @@ pub(crate) extern "C" fn handle_produced_messages( let messages = std::slice::from_raw_parts(messages_ptr, messages_len); match postcard::from_bytes::(messages) { Ok(messages) => { - if let Err(send_error) = entry.sender.send(messages) { - error!( - "Failed to send messages for source connector with ID: {plugin_id}. Channel closed: {send_error}" - ); - entry.error_counter.inc(); - } + send_with_backpressure( + plugin_id, + &sender, + &shutdown, + &backpressure_active, + &error_counter, + messages, + ); } Err(err) => { error!( "Failed to deserialize produced messages for source connector with ID: {plugin_id}. {err}" ); - entry.error_counter.inc(); + error_counter.inc(); } } } } +// Parks a worker thread of the plugin library's shared tokio runtime - a +// deliberate exception to the never-block-the-executor rule: the park IS +// the backpressure, propagating a full channel into the plugin's polling +// loop instead of buffering without bound. Every park is bounded by +// SEND_RETRY_INTERVAL with the shutdown flag re-read in between, so +// iggy_source_close (which waits on the polling task this runs in) cannot +// deadlock on a hung Iggy. Instances loaded from the same .so share that +// runtime, so a saturated sibling can still delay another instance's +// close; signal_shutdown_all covers the process-shutdown path, and the +// complete fix (handing the worker off via block_in_place) belongs in the +// SDK. +fn send_with_backpressure( + plugin_id: u32, + sender: &BatchSender, + shutdown: &AtomicBool, + backpressure_active: &AtomicBool, + error_counter: &Counter, + messages: ProducedMessages, +) { + let mut messages = match sender.try_send(messages) { + Ok(()) => { + // An uncontended send is the recovery signal; a send that only + // succeeded after stalling below is not. + if backpressure_active.swap(false, Ordering::Relaxed) { + info!( + "Forwarding channel for source connector with ID: {plugin_id} recovered from backpressure" + ); + } + return; + } + Err(TrySendError::Full(returned)) => returned, + Err(TrySendError::Disconnected(returned)) => { + log_channel_closed(plugin_id, returned.messages.len(), error_counter); + return; + } + }; + if shutdown.load(Ordering::Acquire) { + drop_during_shutdown(plugin_id, messages.messages.len(), error_counter); + return; + } + if !backpressure_active.swap(true, Ordering::Relaxed) { + warn!( + "Forwarding channel for source connector with ID: {plugin_id} is full. Backpressuring the plugin's polling task." + ); + } + loop { + match sender.send_timeout(messages, SEND_RETRY_INTERVAL) { + Ok(()) => return, + Err(SendTimeoutError::Timeout(returned)) => { + if shutdown.load(Ordering::Acquire) { + drop_during_shutdown(plugin_id, returned.messages.len(), error_counter); + return; + } + messages = returned; + } + Err(SendTimeoutError::Disconnected(returned)) => { + log_channel_closed(plugin_id, returned.messages.len(), error_counter); + return; + } + } + } +} + +fn drop_during_shutdown(plugin_id: u32, message_count: usize, error_counter: &Counter) { + error!( + "Dropping {message_count} produced messages for source connector with ID: {plugin_id}. Channel still full during shutdown." + ); + error_counter.inc(); +} + +fn log_channel_closed(plugin_id: u32, message_count: usize, error_counter: &Counter) { + error!( + "Failed to send {message_count} produced messages for source connector with ID: {plugin_id}. Channel closed." + ); + error_counter.inc(); +} + fn build_iggy_message( payload: Vec, id: Option, @@ -768,3 +911,314 @@ fn build_iggy_message( (None, None) => IggyMessage::builder().payload(payload.into()).build(), } } + +#[cfg(test)] +mod tests { + use super::*; + use iggy_connector_sdk::ProducedMessage; + + fn batch() -> ProducedMessages { + ProducedMessages { + schema: Schema::Raw, + messages: vec![ProducedMessage { + id: None, + checksum: None, + timestamp: None, + origin_timestamp: None, + headers: None, + payload: vec![0u8], + }], + state: None, + } + } + + fn bounded_channel(capacity: usize) -> (BatchSender, BatchReceiver) { + crossfire::mpsc::bounded_blocking_async(capacity) + } + + // Shutdown relies on buffered batches surviving sender drop; crossfire's + // docs don't promise it, so this pins the behavior against upgrades. + #[tokio::test] + async fn given_buffered_batches_when_senders_drop_should_drain_before_disconnect() { + let (sender, receiver) = bounded_channel(4); + for _ in 0..3 { + sender.try_send(batch()).expect("channel has capacity"); + } + drop(sender); + for _ in 0..3 { + assert!( + receiver.recv().await.is_ok(), + "buffered batch must drain after sender drop" + ); + } + assert!( + receiver.recv().await.is_err(), + "drained channel with no senders must disconnect" + ); + } + + #[test] + fn given_full_channel_when_shutdown_signaled_should_drop_batch_and_count_error() { + let (sender, receiver) = bounded_channel(1); + sender.try_send(batch()).expect("fills the channel"); + let shutdown = AtomicBool::new(true); + let backpressure_active = AtomicBool::new(false); + let error_counter = Counter::default(); + send_with_backpressure( + 0, + &sender, + &shutdown, + &backpressure_active, + &error_counter, + batch(), + ); + assert_eq!( + error_counter.get(), + 1, + "batch dropped during shutdown must count as an error" + ); + assert!( + receiver.try_recv().is_ok(), + "pre-existing batch must still be queued" + ); + assert!( + receiver.try_recv().is_err(), + "shutdown-dropped batch must not have been enqueued" + ); + } + + #[test] + fn given_backoff_in_progress_when_shutdown_signaled_should_unblock_and_drop() { + let (sender, _receiver) = bounded_channel(1); + sender.try_send(batch()).expect("fills the channel"); + let shutdown = Arc::new(AtomicBool::new(false)); + let error_counter = Counter::default(); + let sender_in_loop = sender.clone(); + let shutdown_in_loop = shutdown.clone(); + let counter_in_loop = error_counter.clone(); + let blocked = std::thread::spawn(move || { + let backpressure_active = AtomicBool::new(false); + send_with_backpressure( + 0, + &sender_in_loop, + &shutdown_in_loop, + &backpressure_active, + &counter_in_loop, + batch(), + ); + }); + // Let the loop enter backoff before flipping the flag, so a refactor + // that reads the flag once up front cannot pass this test. + std::thread::sleep(Duration::from_millis(30)); + shutdown.store(true, Ordering::Release); + let deadline = std::time::Instant::now() + Duration::from_secs(5); + while !blocked.is_finished() { + assert!( + std::time::Instant::now() < deadline, + "send_with_backpressure must unblock after the shutdown signal" + ); + std::thread::sleep(Duration::from_millis(5)); + } + blocked.join().expect("blocked thread panicked"); + assert_eq!( + error_counter.get(), + 1, + "batch dropped after mid-backoff shutdown must count as an error" + ); + } + + #[test] + fn given_registered_entry_when_signal_shutdown_called_should_set_flag() { + // SOURCE_SENDERS is process-global; an id far above anything the + // runtime allocates keeps this isolated from sibling tests. + let plugin_id = u32::MAX; + let (sender, _receiver) = bounded_channel(1); + SOURCE_SENDERS.insert( + plugin_id, + SourceSenderEntry { + sender, + error_counter: Counter::default(), + shutdown: Arc::new(AtomicBool::new(false)), + backpressure_active: Arc::new(AtomicBool::new(false)), + }, + ); + signal_shutdown(plugin_id); + let flag = SOURCE_SENDERS + .get(&plugin_id) + .expect("entry inserted above") + .shutdown + .load(Ordering::Acquire); + cleanup_sender(plugin_id); + assert!(flag, "signal_shutdown must set the registered entry's flag"); + } + + #[test] + fn given_multiple_registered_entries_when_signal_shutdown_all_called_should_set_every_flag() { + let plugin_ids = [u32::MAX - 1, u32::MAX - 2]; + for plugin_id in plugin_ids { + let (sender, _receiver) = bounded_channel(1); + SOURCE_SENDERS.insert( + plugin_id, + SourceSenderEntry { + sender, + error_counter: Counter::default(), + shutdown: Arc::new(AtomicBool::new(false)), + backpressure_active: Arc::new(AtomicBool::new(false)), + }, + ); + } + signal_shutdown_all(); + let flags: Vec = plugin_ids + .iter() + .map(|plugin_id| { + SOURCE_SENDERS + .get(plugin_id) + .expect("entry inserted above") + .shutdown + .load(Ordering::Acquire) + }) + .collect(); + for plugin_id in plugin_ids { + cleanup_sender(plugin_id); + } + assert!( + flags.iter().all(|flag_set| *flag_set), + "signal_shutdown_all must set every registered entry's flag" + ); + } + + #[test] + fn given_registered_entry_when_callback_invoked_should_deliver_batch_to_channel() { + let plugin_id = u32::MAX - 3; + let (sender, receiver) = bounded_channel(2); + SOURCE_SENDERS.insert( + plugin_id, + SourceSenderEntry { + sender, + error_counter: Counter::default(), + shutdown: Arc::new(AtomicBool::new(false)), + backpressure_active: Arc::new(AtomicBool::new(false)), + }, + ); + let bytes = postcard::to_allocvec(&batch()).expect("batch serializes"); + handle_produced_messages(plugin_id, bytes.as_ptr(), bytes.len()); + cleanup_sender(plugin_id); + let delivered = receiver + .try_recv() + .expect("callback must enqueue the deserialized batch"); + assert_eq!(delivered.messages.len(), 1); + } + + #[test] + fn given_invalid_payload_when_callback_invoked_should_count_error() { + let plugin_id = u32::MAX - 4; + let (sender, _receiver) = bounded_channel(1); + let error_counter = Counter::default(); + SOURCE_SENDERS.insert( + plugin_id, + SourceSenderEntry { + sender, + error_counter: error_counter.clone(), + shutdown: Arc::new(AtomicBool::new(false)), + backpressure_active: Arc::new(AtomicBool::new(false)), + }, + ); + let garbage = [0xFFu8; 3]; + handle_produced_messages(plugin_id, garbage.as_ptr(), garbage.len()); + cleanup_sender(plugin_id); + assert_eq!( + error_counter.get(), + 1, + "deserialize failure must count as an error" + ); + } + + #[test] + fn given_unregistered_plugin_when_callback_invoked_should_drop_silently() { + let bytes = postcard::to_allocvec(&batch()).expect("batch serializes"); + handle_produced_messages(u32::MAX - 5, bytes.as_ptr(), bytes.len()); + } + + #[test] + fn given_backpressure_latched_when_uncontended_send_succeeds_should_clear_latch() { + let (sender, receiver) = bounded_channel(2); + let shutdown = AtomicBool::new(false); + let backpressure_active = AtomicBool::new(true); + let error_counter = Counter::default(); + send_with_backpressure( + 0, + &sender, + &shutdown, + &backpressure_active, + &error_counter, + batch(), + ); + assert!( + !backpressure_active.load(Ordering::Relaxed), + "uncontended send must clear the backpressure latch" + ); + assert!(receiver.try_recv().is_ok(), "batch must be delivered"); + assert_eq!(error_counter.get(), 0); + } + + #[test] + fn given_disconnected_channel_when_sending_should_count_error() { + let (sender, receiver) = bounded_channel(1); + drop(receiver); + let shutdown = AtomicBool::new(false); + let backpressure_active = AtomicBool::new(false); + let error_counter = Counter::default(); + send_with_backpressure( + 0, + &sender, + &shutdown, + &backpressure_active, + &error_counter, + batch(), + ); + assert_eq!( + error_counter.get(), + 1, + "batch lost to a closed channel must count as an error" + ); + } + + #[test] + fn given_full_channel_when_receiver_frees_capacity_should_deliver_batch() { + let (sender, receiver) = bounded_channel(1); + sender.try_send(batch()).expect("fills the channel"); + let delivered = Arc::new(AtomicBool::new(false)); + let delivered_signal = delivered.clone(); + // Keeps the receiver alive until the send lands, so the backoff loop + // can only exit through the success path. + let drainer = std::thread::spawn(move || { + std::thread::sleep(Duration::from_millis(30)); + let drained = receiver.try_recv(); + while !delivered_signal.load(Ordering::Acquire) { + std::thread::sleep(Duration::from_millis(5)); + } + drained + }); + let shutdown = AtomicBool::new(false); + let backpressure_active = AtomicBool::new(false); + let error_counter = Counter::default(); + send_with_backpressure( + 0, + &sender, + &shutdown, + &backpressure_active, + &error_counter, + batch(), + ); + delivered.store(true, Ordering::Release); + assert!( + drainer.join().expect("drainer thread panicked").is_ok(), + "drainer must have freed a slot from the full channel" + ); + assert_eq!( + error_counter.get(), + 0, + "backpressured send must succeed once capacity frees up" + ); + } +} diff --git a/core/connectors/sources/README.md b/core/connectors/sources/README.md index 34989aef00..d25d98eb08 100644 --- a/core/connectors/sources/README.md +++ b/core/connectors/sources/README.md @@ -46,6 +46,7 @@ pub struct SourceConfig { pub plugin_config: Option, pub verbose: bool, // Log message processing at info level instead of debug (default: false) pub benchmark: bool, // Emit per-batch timing events on the `iggy_connectors::benchmark` target (default: false) + pub channel_capacity: Option, // Plugin -> runtime forwarding channel capacity in batches (default: 1024) } ``` @@ -72,6 +73,7 @@ path = "libiggy_connector_random_source" # Path to the source connector config_format = "toml" verbose = false # Log message processing at info level instead of debug benchmark = false # Emit per-batch timing events on `iggy_connectors::benchmark` target +channel_capacity = 1024 # Plugin -> runtime forwarding channel capacity in batches # Collection of the streams to which the produced messages are sent [[streams]]