hubcio commented on code in PR #3795:
URL: https://github.com/apache/iggy/pull/3795#discussion_r3774052533
##########
core/connectors/runtime/src/source.rs:
##########
@@ -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::<ProducedMessages>(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(()) => {
Review Comment:
the `Ok(())` arm never reads `shutdown`, which makes the shutdown drop below
non-terminal: drop batch N, the forwarding loop frees a slot, batch N+1
enqueues, ships, and persists its state - the saved cursor now covers N.
sources advance the cursor at poll time and snapshot it into every batch
(postgres `tracking_offsets`, same shape in the other sources), so a restart
resumes past the hole. silent mid-stream loss, not tail truncation.
the same supersession already exists on master via the `producer.send()` err
path (skips the save, continues), but that one flips the connector to Error
status; this one only logs + counts, and fires on routine paths - SIGTERM
(`signal_shutdown_all` arms every source for the whole sequential-stop window)
and connector restart via the api.
fix is a per-instance `dropped` latch: after the first drop, drop every
later batch too, so the persisted cursor can never pass the gap (ring is fifo,
buffered older batches still flush). use the same flag to latch the `error!` in
`drop_during_shutdown`, otherwise after latching it fires once per poll; keep
the counter bumping per batch.
##########
core/connectors/runtime/src/source.rs:
##########
@@ -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::<ProducedMessages>(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) {
Review Comment:
zero grace: the first `Full` after the flag drops immediately, while the
forwarding loop keeps draining until `cleanup_sender`. one bounded
`send_timeout` round before the first drop (skipped once latched) lets most
in-flight batches land - the parked sender is woken on drain, so the wait is
the drain period, not the full timeout. anything larger only pays off after
`iggy_source_close` moves to `spawn_blocking`, since it currently blocks a host
worker for the duration.
##########
core/connectors/runtime/src/source.rs:
##########
@@ -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::<ProducedMessages>(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) {
Review Comment:
the log reports a message count but the counter moves by 1 per batch, into
the same `iggy_connector_errors_total` series as decode/send/save failures.
this is the only signal for a permanently lost batch - a dedicated counter
(with `inc_by(message_count)` where the count exists) keeps loss
distinguishable from ordinary errors.
##########
core/connectors/runtime/src/source.rs:
##########
@@ -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;
Review Comment:
1024 batches is a weak bound in bytes - postgres defaults to 1000 rows per
batch, so this admits about a million messages plus payloads before
backpressure engages. something like 64 would still be generous. sequencing
note: land any smaller default after the shutdown-drop fix, since a smaller
channel makes `Full` (and drops) more frequent.
##########
core/connectors/sources/README.md:
##########
@@ -72,6 +73,7 @@ path = "libiggy_connector_random_source" # Path to the source
connector
config_format = "toml"
Review Comment:
pre-existing, two lines from your change: the example key is `config_format`
but the real field is `plugin_config_format` - no serde alias and no
`deny_unknown_fields`, so the documented key is silently ignored.
##########
core/connectors/runtime/src/source.rs:
##########
@@ -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.
Review Comment:
crossfire does document drain-before-disconnect for `try_recv` and
`recv_timeout` ("...and the channel is empty") - it's `recv()` whose doc omits
the qualifier. reword to name `recv()` so the comment survives someone checking
the docs; the test is worth keeping regardless.
##########
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_<KEY>_CHANNEL_CAPACITY`.
Review Comment:
only the local config provider wires env overrides - with `config_type =
"http"` this var is silently ignored, and the unknown-var warning is suppressed
for the `IGGY_CONNECTORS_SOURCE_` prefix, so nothing surfaces it. qualify with
"local config provider only".
##########
core/connectors/runtime/src/source.rs:
##########
@@ -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::<ProducedMessages>(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
Review Comment:
this covers the close-delay case but not steady state: the park removes a
worker from the plugin library's shared runtime (one runtime per .so, workers =
available_parallelism), so saturated instances degrade every sibling from the
same .so, and a 1-2 vcpu container can wedge on one instance. worth a sentence
here and in the skill doc. also, a runtime-side `tokio::task::block_in_place`
would be a silent no-op on these threads (they belong to the plugin's tokio,
not ours), so the handoff fix really does have to be sdk-side.
##########
core/connectors/runtime/src/source.rs:
##########
@@ -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);
Review Comment:
this test and
`given_full_channel_when_receiver_frees_capacity_should_deliver_batch` are the
only ones that enter the backoff loop, and both run capacity 1, which crossfire
routes to a different queue impl (`OneMpsc`) and a different backoff regime
(`large = capacity >= 10` gates yield-vs-spin). prod default 1024 is
`ArrayMpsc` with `large = true`, so the park/wake path being pinned isn't the
shipped one. capacity >= 10 here fixes both. the capacity-4 drain test is fine
- `try_send`/`recv` never enter the backoff.
##########
core/connectors/runtime/src/source.rs:
##########
@@ -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);
Review Comment:
future-tuning note: lowering this buys no latency - a parked sender is woken
by the receiver the moment a slot frees, the interval only bounds shutdown-flag
observation. smaller values just multiply yield/park churn.
##########
core/connectors/runtime/src/source.rs:
##########
@@ -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<Array<ProducedMessages>>;
Review Comment:
`DEFAULT_CHANNEL_CAPACITY`, `BatchSender` and `BatchReceiver` have no uses
outside this file and can all drop `pub(crate)` - the private-interfaces lint
checks the resolved type (public crossfire types), not the alias, so private
aliases in `pub(crate)` signatures compile clean.
##########
core/connectors/runtime/src/source.rs:
##########
@@ -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)) =
Review Comment:
only `from_raw_parts` needs the unsafe block - `let messages = unsafe {
std::slice::from_raw_parts(messages_ptr, messages_len) };` and everything else
moves out. the wide block predates this pr, but it grew here.
##########
core/connectors/runtime/src/source.rs:
##########
@@ -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() {
Review Comment:
this can pass even if `signal_shutdown` were a no-op: the sibling test's
`signal_shutdown_all()` sets every entry's flag, including this one, and id
partitioning doesn't help against a whole-map op. one merged test with two
entries - target set, sibling not set, then `signal_shutdown_all` sets both -
is hermetic. only plain `cargo test` is affected (nextest is process-per-test),
but that's the documented local flow.
##########
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);
Review Comment:
return code dropped, and the new ordering's safety argument leans on close
actually stopping callbacks - the sdk returns -1 when it can't join the polling
task, and then callbacks can outlive the close and land on the removed-entry
branch that drops with no metric. `init` checks the same call (`if close_result
!= 0 { warn! }`) - mirror it here.
##########
core/connectors/runtime/src/source.rs:
##########
@@ -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| {
Review Comment:
simpler shape for this: `DashMap<u32, Arc<SourceSenderEntry>>`, clone one
arc, pass `&SourceSenderEntry` down (`send_with_backpressure` goes 6 params ->
3), and `shutdown`/`backpressure_active` become plain `AtomicBool`. disconnect
timing is unchanged - the arc keeps the `MTx` alive exactly like today's clone.
if you take it, write `Arc::clone(&entry)` explicitly; dashmap's `Ref` has no
`Clone`, so a bare `.clone()` only works via deref and reads as a guard clone.
##########
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.
Review Comment:
this undersells the failure: the drop isn't just an error count - a later
batch can still enqueue and persist its state, moving the saved position past
the dropped batch, so the data is silently gone (see the send-callback
comment). it also happens on connector restart via the api, not only at
SIGTERM. once the fix lands this should state the guarantee: replay from the
last delivered batch's state - duplicates for offset-mode sources, permanent
loss for `delete_after_read` / `processed_column` ones. same wording lives in
the connector-runtime skill doc.
--
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]
