feat(connectors): Add Iggy source connector

[jiengup]

Which issue does this PR address?

Closes #3869
Relates to #3764

Rationale

Cross-cluster topic replication had no first-class path: the connectors subsystem's sinks and sources target external systems (Postgres, Elasticsearch, HTTP, etc.), and the runtime itself connects to a single Iggy cluster. Cross-cluster sync (disaster recovery, data migration, federated setups) therefore required external tooling. This PR adds iggy_source: a source connector that replicates a topic from an upstream Iggy cluster into the cluster the connectors runtime is connected to.

What changed?

Previously there was no way to sync two Iggy clusters through connectors. Now core/connectors/sources/iggy_source connects to the upstream cluster and discovers all partitions in open(), polls each partition with an explicit poll_messages(offset) call per cycle, and passes payloads and user headers through as Schema::Raw while preserving upstream message IDs for downstream deduplication. Sync progress (per-partition confirmed offsets, total synced, error count) is persisted in connector state, and restarts resume from saved_offset + 1 without replaying history.

Design tradeoffs

• Low-level poll_messages API instead of the high-level IggyConsumer: crash recovery requires per-partition starting offsets, but IggyConsumer accepts only one global polling_strategy at construction time, which cannot express per-partition offsets. The low-level API allows passing PollingStrategy::offset(saved + 1) per partition explicitly.
• No consumer group: server-side group offset storage is untrusted (consume ack precedes downstream produce, so a crash loses messages). For single-instance replication the group provides only fencing, no real benefit, while introducing join/rebalance failure modes. JoinConsumerGroupResponse is empty (no member id returned), so assignment-driven multi-instance scaling is not expressible through the low-level API. Revisit once the server exposes member ids.
• Direct offset jump vs replay-and-filter: recovery uses offset(saved + 1) directly rather than first() plus in-memory filtering, keeping restart cost O(batch) instead of O(topic history).
• Schema::Raw byte passthrough: no decode/encode; payloads, user headers, and message IDs are preserved verbatim, avoiding re-serialization cost and format corruption.
• InvalidOffset reset: when upstream retention expires a saved offset, the partition is warned and reset to initial_offset (earliest / latest / numeric) without blocking other partitions.
• Exponential backoff: connection-level failures reuse the SDK's exponential_backoff + jitter (starting at retry_interval, capped at max_retry_interval); the failure counter is an AtomicU64 and resets on any successful cycle.
• Missing upstream stream/topic: warn and auto-create (topics with 1 partition).

Persisted state design

#[derive(Debug, Serialize, Deserialize)]
struct State {
    offsets: HashMap<u32, u64>,  // partition_id -> offset of the last message confirmed written downstream
    messages_synced: u64,        // cumulative synced message count
    errors_count: u64,           // cumulative errors (connection failures, conversion failures, offset resets)
}

• Encoding and durability: serialized with MessagePack (rmp_serde) via ConnectorState::serialize/deserialize; the runtime persists it with the existing FileStateProvider to {state_path}/source_{key}.state, inheriting its atomic-rename + fsync protocol and 0o600 permissions. The state save path is untouched.
• Confirmation semantics (the key invariant): the connector advances offsets in state only for messages handed to the runtime, and the runtime saves state only after a successful downstream send. The on-disk state is therefore always the "confirmed delivered downstream" watermark and the single authoritative source for crash recovery.
• Bounded size: state is one u64 per partition plus two counters, O(partition count), so rewriting the whole file every batch is cheap.
• Failure tolerance: deserialization failures (corruption, version mismatch) log a warning and start from a fresh state (non-fatal). initial_offset applies only to partitions with no saved entry (first run or newly added partitions); existing entries always win.
• Error-count persistence: errors_count increments and offset resets ride the same state channel, and state is returned even on empty-message cycles (e.g., connection-failure cycles), so error accounting and offset resets survive restarts.

Crash recovery analysis

Failure point	Behavior
Connector process crash (upstream and downstream healthy)	The runtime persists state only after a successful downstream send; offsets not yet persisted are re-polled → at-least-once, with the preserved upstream message ID enabling downstream dedup
Upstream cluster outage	Poll errors → errors_count incremented, offsets not advanced, exponential backoff with jitter; the SDK client reconnects automatically
Downstream cluster outage	Runtime producer.send fails → state not saved; the connector's in-memory offsets have advanced, so that batch is dropped for the lifetime of the process, this can be fixed when #3855 is merged; a connector restart replays from the stale state file → at-least-once
Upstream retention expires data	InvalidOffset → partition reset to initial_offset; expired messages are unrecoverable (inherent to offset-based replication)

Sync semantics

• At-least-once: state records only offsets confirmed written downstream; recovery always resumes from the confirmation point + 1, so no message is lost, and duplicates within the crash window are deduplicated downstream via the preserved upstream message IDs.
• The state mutex is acquired exactly twice per poll cycle (read offsets / write offsets) and never held across I/O; empty polls do not count as errors.

Local Execution

• Passed: cargo fmt --all, cargo sort --no-format --workspace, cargo clippy -p iggy_connector_iggy_source --all-targets --all-features -- -D warnings, cargo test -p iggy_connector_iggy_source (11 unit tests: state restore, serialization round-trip, config defaults, initial_offset parsing, next_strategy jump, connection-string redaction), taplo, markdownlint, license-headers, cargo machete
• Passed: cargo test -p integration -- connectors::iggy_source
• Pre-commit hooks ran / not ran: not ran

End-to-end verification (two real server-ng instances + runtime)

• Upstream/downstream iggy-server (TCP 8090/8091) + iggy-connectors; a CLI producer continuously emitted messages with headers
• All 542 messages synced, message counts identical on both clusters; sampled payloads, user headers (producer:string, seq:uint64), and message IDs are byte-identical
• Crash recovery: killed the connector, produced 3 more messages, restarted → Restored state ... Offsets: {0: 4}, messages synced: 5 → only the 3 missing messages were synced, the first 5 with zero duplicates
• The topic-size difference between the two clusters was traced to the server's 256-byte per-save metadata blocks (CLI single-message sends vs runtime batched sends), not to any data difference

AI Usage

1. Deepseek V4 pro (opencode)
2. Architecture and trade-offs were decided by the human; implementation was AI-generated with human review
3. E2E test was verified by both the agent and human

[codecov]

Codecov Report

❌ Patch coverage is 83.23171% with 55 lines in your changes missing coverage. Please review.
✅ Project coverage is 82.92%. Comparing base (7c0fd68) to head (37116c6).

Files with missing lines	Patch %	Lines
core/connectors/sources/iggy_source/src/lib.rs	83.23%	52 Missing and 3 partials ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master    #3886      +/-   ##
============================================
- Coverage     82.96%   82.92%   -0.05%     
  Complexity     1339     1339              
============================================
  Files          1218     1219       +1     
  Lines        165595   165925     +330     
  Branches     133910   134365     +455     
============================================
+ Hits         137394   137592     +198     
+ Misses        24540    24529      -11     
- Partials       3661     3804     +143     

Components	Coverage Δ
Rust Core	83.53% <83.23%> (+0.04%)	⬆️
Java SDK	66.55% <ø> (ø)
C# SDK	74.72% <ø> (-1.71%)	⬇️
Python SDK	90.00% <ø> (ø)
PHP SDK	84.48% <ø> (ø)
Node SDK	95.68% <ø> (ø)
Go SDK	69.04% <ø> (ø)

Files with missing lines	Coverage Δ
core/connectors/sources/iggy_source/src/lib.rs	83.23% <83.23%> (ø)

... and 79 files with indirect coverage changes

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[jiengup]

/request-review @kparisa @slbotbm