connector-source

# Writing an Apache Iggy Connector Source A **source** is a Rust `cdylib` that implements `iggy_connector_sdk::Source` and exposes FFI symbols via the `source_connector!` macro. The runtime calls `poll()` in a loop, applies transforms, encodes via the configured `Schema`, sends to Apache Iggy, and persists the returned `ConnectorState` after every successful send. > **Universal connector rules** (SecretString, benchmark, verbose flag, drop accounting, filter contract, exemplar patterns) live in > [connectors-overview](../connectors-overview/SKILL.md). This skill > covers only what's source-specific. ## Contents - [STOP and ask the user before](#stop-and-ask-the-user-before) - [Quick reference](#quick-reference) - [Hard rules](#hard-rules) - [Common pitfalls](#common-pitfalls) - [Tests](#tests) - [Before declaring done](#before-declaring-done) ## STOP and ask the user before - Changing the SDK trait surface (`Source::open` / `poll` / `on_batch_result` / `close`) - that's an SDK change. - Adding a long-running side task in the plugin - the runtime owns lifecycle, and orphans survive `close()`. Sanctioned only where the source is itself a server the runtime cannot drive, as in `http_source`'s listener, and then only with an explicit shutdown in the last `close()` that awaits its tasks before returning. - Persisting unbounded state - `State` is rewritten every batch. - Adding a source that requires authoritative offsets external to Apache Iggy without coordinating retention. ## Quick reference - Skeleton: [TEMPLATE.md](TEMPLATE.md) (load on demand). - Exemplars: `random_source` (minimal + canonical state tests), `postgres_source` (cursor / delete-after-read / processed-column modes, restart-survives-state tests), `elasticsearch_source` (scroll cursor), `influxdb_source` (time-series scan). ## Hard rules ### `poll()` signature is `&self` The macro shares the source as `Arc<T>` across the FFI callback and forwarding loop. Signature: `async fn poll(&self) -> ...` - any mutable state behind `tokio::sync::Mutex`. **Single most common new-contributor mistake.** ### Lock discipline Never hold the state `Mutex` across upstream I/O. Build a candidate from committed state, then stage it until the runtime reports the batch result: ```rust let mut candidate = self.state.lock().await.clone(); let rows = client.query(&sql, &[&candidate.cursor]).await?; candidate.cursor = Some(new_cursor); let persisted = ConnectorState::serialize(&candidate, CONNECTOR_NAME, self.id) .ok_or_else(|| Error::Serialization("failed to serialize source state".into()))?; *self.pending.lock().await = Some(candidate); ``` ### Delivery acknowledgment `on_batch_result` (added by #3855) is how a source learns what happened to the batch it just returned. The SDK keeps exactly one batch in flight: it will not call `poll()` again until this returns, and it stops the source after `MAX_CONSECUTIVE_NACKS` (5) consecutive NACKs, roughly 1.5s of backoff, without calling `close()`. - `Ack` means the runtime sent the batch **and** persisted its state. `Nack` means it could not confirm both, which is **not** the same as neither happening: a batch that reached the topic but whose state save failed is NACKed, and the SDK NACKs on its own result timeout while the send may still have landed. A source that replays on `Nack` is at-least-once, not exactly-once. - The trait has a **default no-op**, which suits only a source with no staged cursor and no destructive work. If `poll()` advances a cursor, deletes rows, or drains an in-memory buffer, omitting this loses data silently and nothing will tell you. `random_source` and `http_source` implement it; the other shipped sources take the default and skip rows on a NACK. - Stage in `poll()`, apply on `Ack`, discard or replay on `Nack`. A source whose input is pushed to it, rather than re-readable upstream, has to hold the batch itself: see `http_source`'s staging. - Returning `Err` from it stops the source immediately, so it is not a retry signal. ### State persistence - `ConnectorState` is `Vec<u8>` via MessagePack (`rmp_serde`). Use `ConnectorState::serialize(&state, NAME, id)` and `ConnectorState::deserialize::<State>(NAME, id)`. - `poll()` must not commit cursors or destructive work. Return messages with candidate state and keep the corresponding work staged. - The runtime sends the batch, saves its candidate state to `{state_path}/source_{key}.state`, then calls `on_batch_result(Ack)`. Commit staged in-memory state and external delete or mark operations only on ACK. A NACK discards the candidate so the same data can be polled again. - A crash between `poll()` returning and state persistence leaves the prior cursor for the next poll, so downstream must tolerate at-least-once delivery. - Cursor sources that stage state until `on_batch_result` can attach state to the corresponding message batch. Control-plane state that is independent of a publish can ride an empty batch instead. - Return `state: None` for an empty poll when no watermark changed. If an empty poll advances a watermark, stage and return the new state through the same ACK handshake. - The runtime can still NACK an empty batch when state storage is latched or a pending checkpoint cannot resolve. Never treat hand-off as durable; `on_batch_result` reports the outcome. - Treat candidate-state serialization failure as a poll error. Do not send messages without the state needed to resume them safely. - Keep `State` small - rewritten every batch. No unbounded vecs. The SDK allows one in-flight batch. Five consecutive NACKs stop the source and require a manual restart. Returning `Err` from `on_batch_result` is fatal, so retry transient backend failures inside the callback before returning an error. The runtime must report ACK or NACK within the SDK's 30-second batch-result window. Once the result is received, the SDK waits for `on_batch_result` to finish, so the callback must bound its own connection acquisition and retry budget rather than relying on the SDK deadline. ### Sleep first `poll()` must `sleep(self.poll_interval).await` before any work. Without it, an empty source spins a CPU. ### Schema selection Match `ProducedMessages.schema` to the bytes in `messages[i].payload`: - JSON-serialized rows → `Schema::Json` - Already-protobuf bytes → `Schema::Proto` - Already-avro bytes → `Schema::Avro` - Opaque → `Schema::Raw` ### IDs and timestamps - `ProducedMessage.id: Option<u128>` - set when a natural ID exists (DB PK, document id). It rides the wire as the message id, which is what a **consumer** can dedupe an at-least-once duplicate on. - **Iggy itself does not dedupe on that id.** The server's only read of the field is `core/server/src/http/wire.rs`, which mints a fresh uuid when the incoming id is 0 and otherwise passes it through. Its dedup path is a per-client request-id watermark (`dedup_clients_max`) that never looks at this field. - `origin_timestamp: Option<u64>` - source-system event time in nanoseconds. Lets downstream sinks reason about lag. - `timestamp` and `checksum` are Iggy-side - leave `None`. ### Concurrency - Runtime spawns ONE `poll()` task per source. No concurrent `poll()`. - Don't spawn your own long-running Tokio tasks: the runtime owns lifecycle. The exception is a source that listens rather than polls, which has to own its listener; `http_source` is the worked example, and it shuts its tasks down in the last `close()` rather than leaving them to outlive the connector. ### Errors | Scenario | Variant | | ------------------------------------------- | ------------------------------------------------- | | Bad config in `new()`/`open()` | `Error::InitError` | | Cannot reach external system at startup | `Error::InitError` or `Error::Connection` | | Transient fetch failure (retry-worthy) | `Error::Connection` or `Error::HttpRequestFailed` | | Permanent fetch failure (auth, schema gone) | `Error::PermanentHttpError` | | Row failed to serialize | `Error::Serialization(...)` | | State serialization failed | `Error::Serialization(...)` | Returning `Err` from `poll()` is only logged by the SDK's FFI bridge (`sdk/src/source.rs::handle_messages`) - the loop continues, the next `poll()` runs. Connector status does NOT flip to `Error` from a poll failure. Status `Error` is set by the runtime only on transform/encode failure, Iggy send failure, or state save failure (`runtime/src/source.rs::source_forwarding_loop` calls to `context.sources.set_error`). To surface a poll failure as unhealth, raise it through the metric counter or escalate to `Error::InitError` from `open()`. ### Logging ```rust info!("Opened <connector> connector ID: {}, endpoint: {}", self.id, ...); info!("Restored state for <connector> ID: {id}, cursor: {:?}", ...); debug!("Polled {} rows for <connector> ID: {}", rows.len(), self.id); warn!("Transient fetch failure for <connector> ID: {}, will retry: {error}", self.id); error!("Failed to <op> for <connector> ID: {}, error: {error}", self.id); info!("Closed <connector> connector ID: {}, total produced: {}", self.id, ...); ``` Iggy consumer-loop labels use literal API names (`offset=`, `current_offset=`). ## Common pitfalls 1. `async fn poll(&mut self)` - won't compile. Use `&self` + `Mutex<State>`. 2. Holding `state.lock()` across the fetch I/O - blocks `close()`, causes shutdown timeouts. 3. Forgetting to sleep - 100% CPU on idle source. 4. Committing a cursor or deleting source data in `poll()` - stage it and wait for ACK. 5. Unbounded data in `State` - rewritten every batch. keep O(constant). 6. `std::sync::Mutex` - blocks the executor. Use `tokio::sync::Mutex`. 7. Not setting `ProducedMessage.id` when a stable ID exists - leaves a consumer nothing to dedupe a replayed duplicate on. It does not make the write idempotent server-side, because nothing there reads it. 8. Spawning side tasks - the runtime owns the scheduler. The one exception is a source that listens rather than polls and must own its listener (see [Concurrency](#concurrency) and the STOP list); it owes an explicit shutdown in the last `close()` that awaits its tasks. ## Tests Mandatory four canonical source state tests (see [connector-testing](../connector-testing/SKILL.md) for the full pattern). Copy from `sources/random_source/src/lib.rs::tests`. Plus config defaults, payload building, schema selection. Integration tests under `core/integration/tests/connectors/<backend>/` for any source backed by external infra. Use `#[iggy_harness]` + a `TestFixture` backed by `testcontainers-modules`. Reference: `core/integration/tests/connectors/postgres/postgres_source.rs` (multi-mode tests) + `restart.rs` (state survives restart). Exercise both ACK and NACK paths when the source stages cursors or destructive work. ## Before declaring done ```bash cargo fmt --all cargo sort --no-format --workspace cargo clippy -p iggy_connector_<name>_source --all-targets -- -D warnings cargo test -p iggy_connector_<name>_source # Integration tests: cargo test -p integration -- connectors::<backend>::<test_name> ``` Update `core/connectors/sources/README.md` and add a sample TOML under `core/connectors/runtime/example_config/connectors/`. --- Discussion / help: see [AGENTS.md](../../../AGENTS.md#discussion-and-support).