connector-sdk

# Extending the Apache Iggy Connectors SDK The SDK (`core/connectors/sdk/`) is the **stable contract** between the runtime and every plugin. Changes ripple to every sink/source in-tree and every third-party plugin. Treat the public surface as a versioned API. > **Universal connector rules** (benchmark, SecretString, drop accounting, exemplar patterns) live in > [connectors-overview](../connectors-overview/SKILL.md). This skill covers SDK-level changes only. ## Contents - [STOP and ask the user before](#stop-and-ask-the-user-before) - [File map](#file-map) - [Cardinal rules](#cardinal-rules) - [Adding a new `Schema` variant](#adding-a-new-schema-variant) - [Adding a `StreamDecoder` / `StreamEncoder`](#adding-a-streamdecoder--streamencoder) - [Adding a new `Error` variant](#adding-a-new-error-variant) - [Modifying `Sink` / `Source` / `Transform` traits](#modifying-sink--source--transform-traits) - [FFI macros (`sink_connector!`, `source_connector!`)](#ffi-macros-sink_connector-source_connector) - [`Payload::try_to_bytes(&self)` is non-negotiable for JSON](#payloadtry_to_bytesself-is-non-negotiable-for-json) - [Retry helpers (`retry.rs`)](#retry-helpers-retryrs) - [`ConnectorState`](#connectorstate) - [Tests for SDK changes](#tests-for-sdk-changes) - [Before declaring done](#before-declaring-done) ## STOP and ask the user before - Bumping the SDK MAJOR version or changing any `#[repr(C)]` layout - misaligns every pre-built plugin `.so`. - Changing `Sink` / `Source` / `Transform` / `StreamDecoder` / `StreamEncoder` trait signatures - cascades through every plugin. - Removing or renaming a `Schema` variant or `Error` variant - decoders/encoders pinned to it. downstream pattern matches break. - Changing wire serialization (postcard / rmp_serde / serde_json) for any FFI / state / config payload. - Changing FFI macro return codes or function names (`iggy_*_open` / `consume` / `handle` / `close`) - runtime + plugins both need lockstep update. ## File map ```text sdk/src/ ├── lib.rs Traits (Sink, Source, StreamDecoder, StreamEncoder), │ Payload, Schema, ConnectorState, Error enum, FFI message structs. ├── sink.rs SinkContainer + sink_connector! macro (FFI plumbing). ├── source.rs SourceContainer + source_connector! macro (FFI plumbing). ├── api.rs ConnectorStatus, ConnectorStats (feature = "api"). ├── convert.rs owned_value_into_serde_json / owned_value_to_serde_json (simd_json ⇄ serde_json bridge). ├── log.rs CallbackLayer for tracing across FFI. ├── retry.rs retry_async + RetryPolicy, CircuitBreaker, HttpRetryMiddleware. ├── decoders/ One per schema: json, raw, text, proto, flatbuffer, avro. ├── encoders/ Mirror of decoders. └── transforms/ add_fields, delete_fields, update_fields, filter_fields, unwrap_envelope, proto_convert, flatbuffer_convert, avro_convert. ``` ## Cardinal rules 1. **Apache 2.0 license header** on every new file. 2. **`Send + Sync`** on every public trait (FFI runs across thread pools). 3. **`#[repr(C)]`** on every type that crosses FFI (`Schema`, `TopicMetadata`, `MessagesMetadata`, `RawMessage`, `ProducedMessages`, `ConsumedMessage`, `DecodedMessage`, etc. in `lib.rs`). Adding fields requires bumping the SDK version - existing plugins built against the old layout will misalign on `postcard::from_bytes`. 4. **postcard** for FFI message serialization (handled by `SinkContainer::consume`, `SourceContainer`'s `handle_messages`). **MessagePack (`rmp_serde`)** for `ConnectorState`. **JSON** (`serde_json`) only for human-editable config that crosses FFI. Don't mix. 5. **`simd_json::OwnedValue`** for JSON payloads, not `serde_json::Value`. Use `convert::owned_value_into_serde_json` as a bridge when you own the value (the common case - it moves strings instead of copying them); fall back to the borrowing `convert::owned_value_to_serde_json` only when you still need the `OwnedValue` afterward. 6. **`BTreeMap`** for headers - deterministic ordering. Never `HashMap` on the wire. 7. **No breaking changes** to `Sink`/`Source`/`StreamDecoder`/`StreamEncoder`/`Transform` trait signatures without coordinating with all in-tree plugins in the same PR. 8. **A `Schema` variant that changes meaning is as breaking as a new variant.** The discriminant crosses FFI unchanged, so a plugin built against the old SDK keeps decoding it the old way and misreads every batch with no error anywhere. Same process as a trait change: bump the SDK minor version, update every in-tree plugin in the same PR, and state in the PR description that plugins built against the previous SDK must be rebuilt. ## Adding a new `Schema` variant The `Schema` enum (in `lib.rs`) is `#[repr(C)]` - it crosses FFI. Touch points to update **in one PR**: 1. `Schema` enum: add variant with `#[strum(to_string = "...")]` matching the `serde(rename_all="snake_case")` form. 2. `Payload` enum: add a matching variant carrying the deserialized form. 3. `Payload::try_into_vec` - consuming bytes-out path. 4. `Payload::try_to_bytes` - **borrowing** bytes-out path. For non-trivial payloads (parsed trees), implement a no-clone serialization, not a `clone() + serialize`. See the `Payload::Json` arm for the canonical optimization. 5. `Payload::schema()` - variant → `Schema`. The runtime tags every sink batch from this, so a missing arm will not compile but a wrong one silently mislabels the batch. 6. `Payload::Display`. 7. `Schema::try_into_payload` - bytes → `Payload`, for a tag naming a wire format (the source path). 8. `Payload::try_from_schema` - bytes → `Payload`, for a tag that came from `Payload::schema()` (the sink path). The exact inverse of item 5. 9. `Schema::decoder()` - factory returning `Arc<dyn StreamDecoder>`. 10. `Schema::encoder()` - factory returning `Arc<dyn StreamEncoder>`. 11. New `decoders/<name>.rs` and `encoders/<name>.rs`. 12. Update the schema list in `sdk/README.md`. 13. Tests: round-trip encode/decode, error paths. `Payload::schema()` must round-trip through `Payload::try_from_schema`, with no exceptions: every variant survives that trip and the test is table-driven over all of them. Miss any of these → silent failures at runtime (typically `Error::InvalidPayloadType` or surprising decoder behavior in plugin code). ## Adding a `StreamDecoder` / `StreamEncoder` Pattern from `decoders/json.rs::JsonStreamDecoder`: ```rust pub struct MyStreamDecoder; // unit struct if stateless impl StreamDecoder for MyStreamDecoder { fn schema(&self) -> Schema { Schema::MyFormat } fn decode(&self, payload: Vec<u8>) -> Result<Payload, Error> { // Decode payload bytes into Payload::MyFormat(...). // Errors: Error::CannotDecode(Schema::MyFormat), Error::InvalidJsonPayload, etc. } } ``` For **stateful** decoders (proto/avro/flatbuffer) holding schema descriptors: - Store the schema state directly in the struct (`config`, `message_descriptor`, `schema`). The instance is built once and wrapped in `Arc<dyn>` for sharing - it must be `Send + Sync` but does not need additional locking if construction-time loading is final. - Provide **two constructors** when schema loading can fail: lenient `new(config) -> Self` (logs + degrades) AND strict `try_new(config) -> Result<Self, Error>` (fail-fast). Plus a `Default` impl for the no-schema case. Canonical example: `sdk/src/decoders/avro.rs` exports all three (`AvroStreamDecoder::new`, `::try_new`, `Default`). - If you genuinely need to mutate the schema after construction (e.g., a `update_config` method), use `&mut self` plus `std::mem::replace(&mut self.config, new_config)` to swap without cloning the old value. Live pattern: `sdk/src/decoders/avro.rs::AvroStreamDecoder::update_config` and `sdk/src/encoders/avro.rs::AvroStreamEncoder::update_config`. - A fresh decoder instance is created via `Schema::decoder()` on each call (it returns `Arc<dyn StreamDecoder>`), so per-decoder caching of shared mutable state is the wrong abstraction. ## Adding a new `Error` variant The `Error` enum (in `lib.rs`) is `Clone + PartialEq + Eq + Hash`. New variants must preserve these. Guidelines: - **Use existing variants first.** Only add a new one if the failure mode is distinct in handling, not just description. `InvalidConfigValue(String)` covers most "bad config" cases. - **Carry context as `String`** - don't add a struct unless multiple discrete fields are needed by callers programmatically. - **Document retry semantics in the docstring.** Existing variants (`PermanentHttpError`, `CatalogCommitError`, `TransactionApplyError`) document retry behavior - mirror that style. - **Place near related variants** to keep the enum readable top-to-bottom. ## Modifying `Sink` / `Source` / `Transform` traits **Breaking changes.** Required process: 1. Document migration in the SDK changelog (and `connectors/sdk/README.md`). 2. Update every in-tree plugin in the same PR. CI builds them all. 3. Bump SDK minor version. 4. Note in PR description: dynamic plugins compiled against the old SDK version will fail to load with mismatched FFI symbols. Non-breaking additions (new default method, new struct field with `#[serde(default)]`, new variant on a non-exhaustive enum) are preferable. ## FFI macros (`sink_connector!`, `source_connector!`) The macros (in `sink.rs` and `source.rs`) generate: - `iggy_<kind>_open(id, config_ptr, config_len, [state_ptr, state_len,] log_callback) -> i32` - `iggy_sink_consume(id, ...)` / `iggy_source_handle(id, callback) -> i32` - `iggy_<kind>_close(id) -> i32` - `iggy_<kind>_version() -> *const c_char` (static lifetime, from `env!("CARGO_PKG_VERSION")`) Invariants: - **Duplicate-ID guard** returns `-1` if the caller reopens an ID without closing first. Don't remove this - it prevents silent buffered-data loss. - `INSTANCES: Lazy<DashMap<u32, SinkContainer<$type>>>` (and `SourceContainer<$type>` mirror) is the only mutable global the macro introduces - keep it that way. - Return codes: `0` success, `-1` invalid call, `1` open failure. Don't repurpose. Changes to the FFI signature must update `runtime/src/main.rs::{SourceApi, SinkApi}` in the same PR. ## `Payload::try_to_bytes(&self)` is non-negotiable for JSON Plugin authors call this on every consumed message. The implementation in `lib.rs::Payload::try_to_bytes` documents *why* it skips the deep `OwnedValue` clone - replacing `O(n) clone + O(n) serialize` with `O(n) serialize`. If you add a new variant requiring a parsed tree, do the same: serialize in place, don't clone. ## Retry helpers (`retry.rs`) - `CircuitBreaker`: threshold + cooldown, `try_lock()` on the success path to avoid hot-path contention. - `retry_async(policy, context, should_retry, op)`: the retry loop for anything failing as `Err`. Owns attempt counting, backoff and the per-retry log; returns `RetryFailure { error, attempts, exhausted }` so callers write their own terminal log. - `retry_backoff(base, retry, max)`: backoff only, for a loop that computes its own delay. `retry` is 1-based. `exponential_backoff` is the 0-based primitive underneath and applies no jitter, so call it directly only when the delay must be exact (`source.rs::nack_retry_delay`, whose tests assert exact values). - `HttpRetryMiddleware`: integrates with `reqwest-middleware`. Retries 429 + 5xx + network errors. Honors `Retry-After` on 429 and 5xx as a floor over the computed backoff, capped at a one-hour `MAX_RETRY_AFTER`, not `max_delay`. - `max_retries` = **total attempts** including the first try, not extra retries. Document if you change this convention. `meilisearch_sink` is the standing exception: its `max_retries` / `max_open_retries` count retries *after* the first, as its README states. - New helpers must take `Duration` (not `u64 millis`) on the public API. Internal computation uses `humantime` parsing of `String`. ## `ConnectorState` - Bytes are opaque - the wrapping type does not impose schema. - Serialization is MessagePack via `rmp_serde`. Compact, deterministic, well-supported in serde. - `serialize`/`deserialize` helpers return `Option<T>` and log on failure. **Failures are non-fatal** - design downstream code to tolerate fresh state. - Don't change the underlying format without coordinating with all sources. ## Tests for SDK changes - Unit tests for any new pure function in the changed module. - Round-trip tests for new schemas/transforms (encode → decode == identity for JSON-equivalent payloads). - `sdk/tests/` integration tests when the change crosses module boundaries. ## Before declaring done ```text cargo fmt --all cargo sort --no-format --workspace cargo clippy -p iggy_connector_sdk --all-targets --all-features -- -D warnings cargo test -p iggy_connector_sdk --all-features # Rebuild all plugins to catch breaking-change leaks: cargo build -p iggy_connector_stdout_sink -p iggy_connector_random_source # If FFI signatures changed, also build the runtime: cargo build -p iggy-connectors ``` --- Discussion / help: see [AGENTS.md](../../../AGENTS.md#discussion-and-support).