Connector Runtime

Modifying the Apache Iggy Connectors Runtime

Runtime is the host. Loads .so/.dylib/.dll plugins, drives their lifecycle, ferries messages between Apache Iggy and plugins, exposes /stats, /metrics, control HTTP API. Bugs here affect every plugin.

Universal connector rules (benchmark, SecretString, verbose, drop accounting, exemplar patterns) live in connectors-overview. This skill covers runtime internals only.

Contents

• STOP and ask the user before
• File map
• Architecture invariants
• State storage (state.rs)
• Source forwarding loop
• Sink consumption loop
• Logging format
• The verbose flag
• Config provider rules
• Error categorization
• Metrics (metrics.rs)
• Drop accounting (wired sites)
• Hard rules
• Common pitfalls
• Before declaring done

STOP and ask the user before

• Changing FFI #[repr(C)] layouts (SourceApi, SinkApi, RawMessages, headers) - existing plugin .sos misalign.
• Modifying state.rs save protocol (atomic tmp + rename + fsync ordering) - corruption risk.
• Renaming the default consumer group iggy-connect-sink-{key} - operators hold offsets there.
• Hot-reloading or unloading a .so - mid-FFI tasks would segfault. Architectural change required.
• Changing the SerDe split (postcard for FFI payloads, serde_json for configs, rmp_serde for state) - downstream incompat.

File map

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
├── stream.rs                Stream + consumer/producer setup
├── transform.rs             Loads transforms from config, applies them in chain
├── state.rs                 FileStateProvider, atomic ConnectorState file I/O
├── context.rs               Per-instance context (Iggy client, metrics, config)
├── error.rs                 RuntimeError with as_code() for HTTP responses
├── log.rs                   init_logging (text/JSON x telemetry-on/off matrix) + LOG_CALLBACK
├── benchmark.rs             emit_sink_event / emit_source_event + as_micros
├── metrics.rs               Prometheus families + gauges + stage histograms + label caches
├── stats.rs                 /stats endpoint payload assembly
├── manager/{mod,sink,source}.rs   SinkManager / SourceManager, status, restart_guard
├── configs/
│   ├── runtime.rs           RuntimeConfig + LoggingConfig + LogFormat
│   ├── connectors.rs        ConnectorConfig, SinkConfig, SourceConfig (verbose + benchmark)
│   └── connectors/{local,http}_provider.rs
└── api/                     HTTP control + observability endpoints

Architecture invariants

Load-bearing assumptions. Breaking them silently corrupts data or hangs.

Plugin instance identity

• Each instance gets a globally unique plugin_id: u32 from PLUGIN_ID: AtomicU32 in main.rs. Never reused.
• Same .so may load many instances. each gets a fresh plugin_id but shares the Container (dlopen handle) via Arc (manager/{sink,source}.rs).
• All FFI calls keyed by plugin_id. Mixing IDs = data to the wrong plugin.

FFI pointer lifetimes

• Config and state are (*const u8, usize). Valid for the call duration only. Plugin must copy if it needs the bytes past the call. SDK macros do this correctly. don't bypass.
• LogCallback lifetime spans the whole process - kept static in log.rs. Don't make it dynamic.

Container ownership

• dlopen2::wrapper::Container<{Sink,Source}Api> (defined in main.rs) is shared via Arc across all instances of the same .so.
• Arc must outlive every spawned task that calls into the plugin. Manager holds the Arc in {Sink,Source}Details for the connector lifetime.
• Unloading a .so while a task is mid-FFI-call would segfault. Runtime does not unload. Hot reload needs a rearchitect.

Plugin manager state

• SinkManager / SourceManager use DashMap<String, Arc<Mutex<Details>>>. Key = TOML key field.
• Status transitions (Starting → Running → Stopped/Error) update running counters only on transitions to/from Running. See SinkManager::update_status.
• restart_guard: Arc<Mutex<()>> prevents concurrent restart of the same connector. restart_connector() uses try_lock() - returns OK without restart if busy. Intentional anti-thundering-herd.

Serialization split

• FFI message payloads (TopicMetadata, MessagesMetadata, RawMessages, headers): postcard (compact, stable wire).
• Connector configs passed to plugin: serde_json (TOML in, JSON over FFI - human-editable).
• ConnectorState bytes inside plugins: rmp_serde (MessagePack - compact, opaque to runtime).

Don't mix.

State storage (state.rs)

• File path: {state_path}/source_{connector_key}.state. Mode 0o600 on Unix.
• save() is crash-atomic. rename(2) is atomic in the namespace (no observer sees a half-renamed file). the preceding sync_data on the tmp file plus the post-rename parent-dir sync_all are what make a crash leave either the old or new content - never truncated. The parent-dir fsync failure is propagated as CannotWriteStateFile (not swallowed) so a lost rename surfaces instead of silently restarting the source from scratch. A Mutex<()> serializes concurrent saves on the same provider.
• Save after every successful Apache Iggy send. Save failure logs + continues. next batch retries.
• load() returns Ok(None) for missing or empty files. After a NotFound read, re-stats the parent directory: missing parent -> Err(CannotOpenStateFile) so a broken state path fails at init rather than masquerading as "fresh start".
• Sinks have no state - only sources.

Source forwarding loop

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<DashMap<u32, SourceSenderEntry>> (pub(crate)). SourceSenderEntry wraps the sender + a pre-extracted owned Counter (the errors series, Arc<AtomicU64> inside). The FFI callback bumps errors on deserialize or channel-closed failure with one relaxed atomic - no Family lookup, no Arc<Metrics> handle.
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.

Gotchas:

• SOURCE_SENDERS must be cleaned up on connector close or memory leaks (channel + task).
• 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.spawns 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.

Sink consumption loop

Per [[streams]] entry per sink:

1. Build / ensure consumer group iggy-connect-sink-{key} (connect matches the docker image name apache/iggy-connect - see runtime/src/sink.rs::default_consumer_group). Overridable via the stream config's consumer_group.
2. Spawn one task per topic (spawn_consume_tasks).
3. Poll Iggy → batch messages (default batch_length = 1000, poll_interval = 5ms). A consumer.next() Err (transport/decode at the Iggy client boundary) bumps errors and continues - offset already auto-committed via AutoCommit::When(AutoCommitWhen::PollingMessages), message effectively dropped.
4. Decode via Schema::decoder(). Decode failure logs + bumps errors. messages_filtered is reserved for the transform filter contract (Ok(None)).
5. Apply transforms (chain). A transform Err is logged, bumps errors, and drops only that message (drop-and-continue, mirroring the source) - one bad payload never kills the batch. Ok(None) filters the message - bumps messages_filtered{connector_type="sink"}.
6. Postcard-encode batch as RawMessages. Headers as a separate postcard blob. Per-message failures (missing field, payload conversion, header serialization) bump errors + skip. Batch-level failures (postcard serialization of metadata / RawMessages) propagate Err -> spawn_consume_tasks wrapper bumps errors + set_error. The total histogram and benchmark emit still fire for the failed batch before the Err propagates.
7. Call iggy_sink_consume(id, topic_meta, messages_meta, messages).
8. Non-zero return → log + errors increment.

Per-message counters (decode errors, transform errors, filters, field/serialize drops) are accumulated and flushed once per batch via inc_errors_by_with_labels / inc_messages_filtered_with_labels, not one Family lookup per message.

Stages (both labels snake_case): decode (schema decode loop), prepare (transform + serialize = total - decode - ffi), ffi (plugin consume call), total. Same decode / prepare meaning as the source side.

Gotchas:

• Batch flushes on current_offset != message_offset (gap) OR batch full - higher latency for low-volume topics.
• Auto-commit happens on poll - idempotency at the sink protects against re-delivery on crash before plugin write.

Logging format

LoggingConfig in configs/runtime.rs exposes format: LogFormat (Text default, Json), env-addressable via IGGY_CONNECTORS_LOGGING_FORMAT=json. log::init_logging matches on (telemetry.enabled, format) and installs the right fmt::layer()/fmt::layer().json() + OpenTelemetry layer. OTel pipeline untouched by format - only the stdout layer switches.

fmt::layer() and fmt::layer().json() are different concrete types, so the stdout layer is built once as a Box<dyn Layer<_>> (via .boxed()) by matching on format, then a single 2-arm branch on telemetry.enabled attaches the OTel layers. When extending, keep the boxed-layer + 2-arm shape rather than re-expanding into a 4-way match.

The verbose flag

SinkConfig and SourceConfig in configs/connectors.rs carry verbose: bool (default false). Runtime threads it to spawn_consume_tasks / source equivalent to gate per-batch log promotion:

if verbose {
    info!("Processing {messages_count} messages for sink connector with ID: {plugin_id}");
} else {
    debug!("Processing {messages_count} messages for sink connector with ID: {plugin_id}");
}

New per-batch logging follows the same pattern. Default to debug!, upgrade to info! when verbose. No third level.

Config provider rules

Local (configs/connectors/local_provider.rs)

• Discovery walk reads *.toml from config_dir. Skips hidden files (.) and Cargo.toml.
• Filename: {key}_{type}[_v{N}].toml (e.g., postgres_sink_v2.toml).
• One TOML deserialized into SinkConfig or SourceConfig. Grouped by key. Highest version wins unless .active_versions.toml overrides.
• .active_versions.toml is NOT read via the discovery walk (hidden). Read via direct path through LocalConnectorsConfigProvider::active_versions_file_path.

HTTP (configs/connectors/http_provider.rs)

• reqwest_middleware + RetryTransientMiddleware. Retries 5xx + connection errors. Never 4xx (PermanentHttpError convention).
• Default URL templates documented in runtime/README.md. Configurable via TOML.

Env-var overrides via ConfigEnv derive

SinkConfig, SourceConfig, and inner structs derive ConfigEnv (configs_derive::ConfigEnv). Generates env-var addressability as IGGY_CONNECTORS_<TYPE>_<KEY>_<FIELD> 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)].

Versioning

• Configs carry version: u64. Local provider auto-increments if absent. HTTP provider trusts the server.
• Restart on version change uses restart_guard.

Error categorization

RuntimeError (error.rs) carries as_code() for HTTP API responses.

Fatal errors propagate to main and exit. Per-connector / per-message errors are isolated.

Metrics (metrics.rs)

All families labeled by connector_key + connector_type (histogram adds stage):

• Counters: iggy_connector_messages_{produced,sent,consumed,processed,filtered}_total and iggy_connector_errors_total. These are the rendered names; each is registered without the _total, which the OpenMetrics encoder appends.
  • messages_filtered_total - intentional drops via transform Ok(None).
  • errors_total - unexpected drops (decode/encode/build failure, missing field, ...) + batch-level failures.
• Histograms: iggy_connector_stage_duration_seconds{stage} (snake_case stage labels - prepare, ffi, decode, iggy_send, state_save, total). Buckets STAGE_BUCKETS_SECONDS. Always populated regardless of any flag. Scraped at /metrics when [http.metrics] enabled = true.
• Gauges: iggy_connectors_{sources,sinks}_{total,running}.

/stats JSON surface mirrors counters per-connector via ConnectorStats (sdk/api.rs): messages_filtered, errors, kind-specific counters.

When adding a metric:

• Add family to Metrics struct + init, register with name + help text. Never end a Counter family's registered name in _total: the encoder appends it and the series renders _total_total. Gauges get no suffix, so a gauge name may end in _total literally.
• New label sets define EncodeLabelSet struct + label enum (hand-impl EncodeLabelValue for snake_case values - the derive emits PascalCase).
• Histograms: pass fn() -> Histogram to Family::new_with_constructor.
• Add unit tests under mod tests with given_*_when_*_should_* BDD names.

Drop accounting (wired sites)

Per-message drops in the batch loops are counted into a local u64 and flushed once after the loop via inc_errors_by_with_labels / inc_messages_filtered_with_labels - one Family lookup per batch, not per message. One-shot drops outside a loop (e.g. consumer.next() Err) still call inc_errors_with_labels directly. Hot path uses pre-built SinkLabels/SourceLabels. &str-based wrappers are #[cfg(test)]-only.

• sink.rs::consume_messages - consumer.next() Err
• sink.rs::process_messages - decode, transform Err (drop-and-continue), missing required fields, payload conversion, header serialization (accumulated, flushed once per batch)
• 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

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

Common pitfalls

1. Mutating INSTANCES (in SDK macro) from runtime side - impossible by design. only the macro touches it.
2. Calling iggy_*_open with a duplicate ID - returns -1. Means runtime forgot to iggy_*_close first.
3. Holding a DashMap shard guard across .await - lockups. Pattern: .get().clone(), drop guard, await.
4. Adding fields to #[repr(C)] types without an SDK version bump - existing plugins misalign on postcard::from_bytes.
5. Changing the default consumer group name without coordinating with operators - they have offsets stored there.

Before declaring done

cargo fmt --all
cargo sort --no-format --workspace
cargo clippy -p iggy-connectors --all-targets -- -D warnings
cargo test -p iggy-connectors

# Sanity-build in-tree plugins:
cargo build -p iggy_connector_stdout_sink -p iggy_connector_random_source

# Runtime-focused integration tests:
cargo test -p integration -- connectors::runtime::

# Smoke-test with example config:
IGGY_CONNECTORS_CONFIG_PATH=core/connectors/runtime/example_config/config.toml \
  cargo run --bin iggy-connectors

Update runtime/README.md if endpoints, env vars, or config schema change.

Discussion / help: see AGENTS.md.