fix(connectors): bound source forwarding channel with backpressure - #3795
fix(connectors): bound source forwarding channel with backpressure#3795mlevkov wants to merge 2 commits into
Conversation
|
Thanks for the PR. It is labeled Slash commands (own line, regular comment) move it around the queue:
See CONTRIBUTING.md for details. |
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## master #3795 +/- ##
=============================================
- Coverage 82.84% 20.17% -62.68%
+ Complexity 1299 1291 -8
=============================================
Files 1199 1198 -1
Lines 161885 135243 -26642
Branches 131360 104844 -26516
=============================================
- Hits 134120 27289 -106831
- Misses 24225 107329 +83104
+ Partials 3540 625 -2915
🚀 New features to boost your workflow:
|
e5ecd55 to
f03fe1f
Compare
aa2a33f to
d7580d5
Compare
|
/request-review @hubcio |
Iggy has no way to receive a webhook. Every provider that pushes events over HTTP needs something in front of it, and today that means running a separate service whose only job is to accept a POST and republish it. This connector removes that hop: it runs an embedded HTTP server, accepts authenticated POST bodies, and produces them to the instance's stream and topic as raw bytes. One plugin .so is loaded once no matter how many source entries reference it, so the listener cannot live on any single instance. It lives in a process-global registry keyed by listen address: the first open binds the public and admin ports, later opens validate their body limit, admin address, management token and instance name against the running listener before joining, and the last close releases both ports. Mismatches fail that instance's open rather than silently handing it a listener its configuration does not describe. A single port can therefore serve many providers, each routed to its own topic. Requests resolve against an ArcSwap route table that is rebuilt whole on every control-plane change, so one atomic load yields both the endpoint's auth rules and the destination bridge. Secret paths carry 128 bits in the URL itself, on the model of a Slack webhook, with optional bearer or HMAC on top; HMAC is verified over the raw body in constant time. Revoked endpoints answer 404 alongside paths that never existed, so a leaked URL cannot be used to confirm it was once live. Endpoints can be registered, re-keyed and revoked at runtime through a token-guarded API on the admin listener, because revoking a compromised endpoint is time-critical and provisioning one per tenant is inherently programmatic. Those endpoints ride the SDK's ConnectorState, and state is attached only to an empty batch: the runtime saves state solely on the success branch of the Iggy send, and an empty send always succeeds, so a mutation cannot be lost to an unrelated send failure. Revocation writes a tombstone that outranks TOML on restore, so a stale config file cannot resurrect an endpoint an operator revoked. Delivery is best-effort in both directions and the README says so first, before anything else: HTTP 200 means accepted into an in-memory buffer, and both the loss and duplicate windows are enumerated with what mitigates each. A full bridge answers 429 with Retry-After rather than blocking, since holding the connection open would turn a slow Iggy into a retry storm. Gateway metrics on the admin listener cover accept-to-200 latency, which the runtime's own stage histograms begin too late to see. Part of the webhook gateway design accepted in apache#3039. The backpressure chain is only complete once the bounded runtime forwarding channel from apache#3795 lands; until then a full bridge signals an arrival burst rather than a slow Iggy, which the README documents. Co-authored-by: Claude <noreply@anthropic.com>
d7580d5 to
bd19da2
Compare
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 without bound instead of propagating backpressure into the plugin's polling loop. Swap it for a bounded crossfire channel (the shard and server-ng standard), sized by an optional SourceConfig channel_capacity counted in batches, defaulting to 1024. The FFI callback retries with send_timeout while re-reading a shutdown flag, set by the manager before iggy_source_close and for every instance ahead of the sequential process-shutdown stops, since same-library instances share one plugin runtime and a wedged sibling would otherwise hold a worker an earlier close needs. A unit test pins that buffered batches drain after the senders drop, which shutdown relies on and crossfire's docs do not promise. This drops flume from the runtime. Requested in the HTTP source discussion (apache#3039). Co-authored-by: Claude <noreply@anthropic.com>
bd19da2 to
399e46c
Compare
Iggy has no way to receive a webhook. Every provider that pushes events over HTTP needs something in front of it, and today that means running a separate service whose only job is to accept a POST and republish it. This connector removes that hop: it runs an embedded HTTP server, accepts authenticated POST bodies, and produces them to the instance's stream and topic as raw bytes. One plugin .so is loaded once no matter how many source entries reference it, so the listener cannot live on any single instance. It lives in a process-global registry keyed by listen address: the first open binds the public and admin ports, later opens validate their body limit, admin address, management token and instance name against the running listener before joining, and the last close releases both ports. Mismatches fail that instance's open rather than silently handing it a listener its configuration does not describe. A single port can therefore serve many providers, each routed to its own topic. Requests resolve against an ArcSwap route table that is rebuilt whole on every control-plane change, so one atomic load yields both the endpoint's auth rules and the destination bridge. Secret paths carry 128 bits in the URL itself, on the model of a Slack webhook, with optional bearer or HMAC on top; HMAC is verified over the raw body in constant time. Revoked endpoints answer 404 alongside paths that never existed, so a leaked URL cannot be used to confirm it was once live. Endpoints can be registered, re-keyed and revoked at runtime through a token-guarded API on the admin listener, because revoking a compromised endpoint is time-critical and provisioning one per tenant is inherently programmatic. Those endpoints ride the SDK's ConnectorState, and state is attached only to an empty batch: the runtime saves state solely on the success branch of the Iggy send, and an empty send always succeeds, so a mutation cannot be lost to an unrelated send failure. Revocation writes a tombstone that outranks TOML on restore, so a stale config file cannot resurrect an endpoint an operator revoked. Delivery is best-effort in both directions and the README says so first, before anything else: HTTP 200 means accepted into an in-memory buffer, and both the loss and duplicate windows are enumerated with what mitigates each. A full bridge answers 429 with Retry-After rather than blocking, since holding the connection open would turn a slow Iggy into a retry storm. Gateway metrics on the admin listener cover accept-to-200 latency, which the runtime's own stage histograms begin too late to see. Part of the webhook gateway design accepted in apache#3039. The backpressure chain is only complete once the bounded runtime forwarding channel from apache#3795 lands; until then a full bridge signals an arrival burst rather than a slow Iggy, which the README documents. Co-authored-by: Claude <noreply@anthropic.com>
hubcio
left a comment
hubcio
left a comment
There was a problem hiding this comment.
a few things that don't fit a diff line:
manager/source.rs:164- the 5s task-await is per handle (there are two), fixed, andchannel_capacitynow scales the post-cleanup_senderdrain against it (buffered batches still drain after the sender drops, up to capacity send+save iterations). abort mid-drain is a clean tail truncation since the state save is atomic, so it costs lost position, not corruption - but the interaction deserves a line in the readme: raising capacity lengthens shutdown.source.rs:627-iggy_source_handle's i32 is discarded too, same family as the close-code comment; only reachable in a start-then-stop race today, so hygiene.- elasticsearch_source with
[state] enabled = truepersists its own cursor at close and overrides the runtime state at open, so a runtime-side latch can't cover it. opt-in and off by default; follow-up issue. - the pre-existing
producer.send()err path has the same cursor-supersession shape (skip save, continue, next batch persists) - the latch here doesn't close that one; separate follow-up. - worth a regression test once the latch lands: saturate the channel, stop the connector, restart, assert no row gap -
state_persists_across_connector_restartin the postgres integration suite is a natural template. spawn_source_handleris at 11 params; passing&SourceConfiginstead needs the resolved-path/version split onSourceConnectorPluginfirst, so follow-up sized.
| messages: ProducedMessages, | ||
| ) { | ||
| let mut messages = match sender.try_send(messages) { | ||
| Ok(()) => { |
There was a problem hiding this 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.
| return; | ||
| } | ||
| }; | ||
| if shutdown.load(Ordering::Acquire) { |
There was a problem hiding this 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.
| } | ||
| } | ||
|
|
||
| fn drop_during_shutdown(plugin_id: u32, message_count: usize, error_counter: &Counter) { |
There was a problem hiding this 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.
| } | ||
| } | ||
|
|
||
| // Parks a worker thread of the plugin library's shared tokio runtime - a |
There was a problem hiding this 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.
|
|
||
| #[test] | ||
| fn given_backoff_in_progress_when_shutdown_signaled_should_unblock_and_drop() { | ||
| let (sender, _receiver) = bounded_channel(1); |
There was a problem hiding this 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.
| } | ||
|
|
||
| #[test] | ||
| fn given_registered_entry_when_signal_shutdown_called_should_set_flag() { |
There was a problem hiding this 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.
| 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); |
There was a problem hiding this 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.
|
|
||
| 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. |
There was a problem hiding this 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.
| channel_capacity = 1024 | ||
| ``` | ||
|
|
||
| Environment override: `IGGY_CONNECTORS_SOURCE_<KEY>_CHANNEL_CAPACITY`. |
There was a problem hiding this 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".
| @@ -72,6 +73,7 @@ path = "libiggy_connector_random_source" # Path to the source connector | |||
| config_format = "toml" | |||
There was a problem hiding this 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.
github-actions
Bot
added
S-waiting-on-author
2 participants
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
crossfire::mpsc::bounded_blocking_async), the same shapeshardandserver-nguse. flume is no longer a runtime dependency.SourceConfigfield,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 existingConfigEnvderive providesIGGY_CONNECTORS_SOURCE_<KEY>_CHANNEL_CAPACITY; configs without the field behave as before apart from the bound.try_sendfast path and asend_timeout(10ms)retry loop that re-reads a per-instance shutdown flag between waits. The manager sets that flag beforeiggy_source_closeso 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.warn!per backpressure episode (latched, cleared on genuine recovery). A batch that still cannot be enqueued after the stop signal is dropped and counted iniggy_connector_errors_total.One correction to the discussion notes
@hubcio the spec assumed crossfire's blocking sender has no
send_timeout. It does:blocking_tx.rs:288onTx, reachable fromMTxviaDeref. The loop is built on it instead oftry_sendplus 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_placearound the callback invocation); happy to file it as a follow-up issue.Test plan
cargo clippy -p iggy-connectors --all-targets -- -D warningscleancargo test -p iggy-connectors: 128 passed, including the new channel and shutdown testscargo build -p iggy_connector_stdout_sink -p iggy_connector_random_sourcecargo test -p integration -- connectors::runtime::could not run on this machine (hwlocality-sysneedspkg-config); relying on CI for the integration suite