feat(connectors): store source state on an HTTP state server

Cargo.toml

@@ -314,6 +314,7 @@ sqlx = { version = "0.9.0", features = [
 static-toml = "1.3.0"
 strum = { version = "0.28.0", features = ["derive"] }
 strum_macros = "0.28.0"
+subtle = "2.6.1"
 # Pinned to 2 because darling 0.23 still emits syn 2 types.
 syn = { version = "2", features = ["full", "extra-traits"] }
 sysinfo = "0.39.6"

core/connectors/runtime/Cargo.toml

@@ -33,6 +33,7 @@ publish = false
 async-trait = { workspace = true }
 axum = { workspace = true }
 axum-server = { workspace = true }
+bytes = { workspace = true }
 clap = { workspace = true }
 configs = { workspace = true }
 configs_derive = { workspace = true }
@@ -66,6 +67,7 @@ serde_json = { workspace = true }
 serde_with = { workspace = true }
 serde_yaml_ng = { workspace = true }
 strum = { workspace = true }
+subtle = { workspace = true }
 sysinfo = { workspace = true }
 system_stats = { workspace = true }
 thiserror = { workspace = true }
@@ -75,6 +77,8 @@ tower-http = { workspace = true }
 tracing = { workspace = true }
 tracing-opentelemetry = { workspace = true }
 tracing-subscriber = { workspace = true, features = ["json"] }
+uuid = { workspace = true }
 
 [dev-dependencies]
 tempfile = { workspace = true }
+wiremock = { workspace = true }

core/connectors/runtime/README.md

@@ -41,6 +41,60 @@ format = "text" # Options: "text" (default), "json"
 
 The path to the configuration can be overridden by `IGGY_CONNECTORS_CONFIG_PATH` environment variable. Each configuration section can be also additionally updated by using the following convention `IGGY_CONNECTORS_SECTION_NAME.KEY_NAME` e.g. `IGGY_CONNECTORS_IGGY_USERNAME` and so on.
 
+## State storage
+
+Source connectors checkpoint their progress (an opaque byte blob) through the runtime's state storage. The backend is selected via `state.storage`:
+
+- `file` (default): one file per source at `{state.path}/source_{key}.state`, written crash-atomically. Ties the cursor to the local disk.
+- `http`: one resource per source at `{state.http.url}/source_{key}` on any HTTP-speaking store (a sidecar in front of a database, an object-store gateway, a coordination service). Cursors survive node replacement and failover to another runtime instance.
+
+```toml
+[state]
+path = "local_state"      # used by storage = "file"
+storage = "http"          # "file" | "http"
+
+[state.http]
+url = "http://127.0.0.1:8080/connectors/state"  # base URL, no trailing slash
+timeout = "5s"
+
+# Static headers attached verbatim to every request. Carries authentication
+# (e.g. a bearer token) and any deployment-specific metadata the server
+# wants. The runtime never interprets them and never logs their values.
+[state.http.request_headers]
+authorization = "Bearer ..."
+
+[state.http.retry]
+enabled = true
+max_attempts = 4          # retries per logical operation, after the first try
+initial_backoff = "200ms"
+max_backoff = "2s"
+backoff_multiplier = 2
+```
+
+Everything except `request_headers` is also env-addressable: `IGGY_CONNECTORS_STATE_STORAGE`, `IGGY_CONNECTORS_STATE_HTTP_URL`, `IGGY_CONNECTORS_STATE_HTTP_TIMEOUT`, `IGGY_CONNECTORS_STATE_HTTP_RETRY_MAX_ATTEMPTS`, and so on.
+
+`storage = "http"` with a missing or invalid `url` is a fatal startup error. With the HTTP backend, any state-load failure other than a clean `404` while starting an enabled source also fails startup: the store is unhealthy, and treating the failure as "no state" would silently rewind the source.
+
+### HTTP backend semantics
+
+- Every read remembers the returned `ETag`; every write is conditional: `If-Match: <etag>` when a version is tracked, `If-None-Match: *` for the first-ever write. There is no unconditional overwrite path.
+- Every write carries an `Idempotency-Key` header, minted once per logical save and reused byte-identically across that save's retries, so a server that committed a write but lost the response can replay the original outcome instead of failing the retry with a spurious `412`.
+- `425`/`429`/`503`/`5xx`, timeouts and connect failures are retried with exponential backoff (honoring `Retry-After`, capped at `max_backoff`) and classified transient when exhausted: the batch is Nacked and the plugin re-polls.
+- `412`/`409` (version conflict), `401`/`403` (authorization lost) and protocol violations are permanent: the provider latches and every later save fails fast without touching the network, until the connector is restarted. A permanent error means another writer took over or this writer's authority was revoked - retrying cannot help and would mask the original error.
+- Durability is the server's durability. The runtime guarantees only that the checkpoint is not advanced (the batch is not Acked) unless the server confirmed the write.
+
+### Implementing a state server
+
+Any HTTP server can back the state endpoint if it meets this contract:
+
+1. `GET {url}/source_{key}` returns `200` + strong `ETag` + the stored bytes, or `404` if the key has never been written. If the value exists but cannot be served consistently yet, return `425` or `503` (optionally with `Retry-After`) - never `404`.
+2. `PUT {url}/source_{key}` must enforce `If-Match`/`If-None-Match` atomically against the stored version and return `412` on mismatch. Success responses (`200`/`201`/`204`) carry the new `ETag`. Requests without a conditional header may be rejected with `428 Precondition Required`.
+3. `Idempotency-Key`: remember each key's outcome for at least the client's retry horizon and replay it on repeats. Replays must not re-apply the write.
+4. ETags are strong and change on every committed write. The runtime treats them as opaque strings: stored verbatim, sent back verbatim.
+5. Authentication is the server's business, transported via the configured static headers; failures are `401`/`403`. Treat `403` as authoritative revocation, not a transient state.
+
+This contract lets a deployment back the endpoint with a replicated store and revoke a stale writer's authority mid-flight: the stale writer's next conditional PUT fails `403`/`412`, its provider latches, and its pipeline stops advancing state - no backend-specific knowledge in the runtime.
+
 ## Logging
 
 By default, the runtime emits human-readable text logs via the `tracing` crate. Switching `logging.format` to `json` produces structured JSON lines (one event per line, machine-parseable). When telemetry is enabled, the chosen format applies to the local stdout layer; OpenTelemetry export is unaffected.

core/connectors/runtime/config.toml

@@ -50,7 +50,26 @@ ca_file = "core/certs/iggy_ca_cert.pem"
 domain = "" # Optional domain for TLS connection
 
 [state]
-path = "local_state"
+path = "local_state" # Used by storage = "file"
+storage = "file" # "file" | "http"
+
+# HTTP state storage backend, used when storage = "http". Source state is
+# stored at {url}/source_{key} with optimistic concurrency (ETag/If-Match)
+# and idempotent retries (Idempotency-Key).
+[state.http]
+url = "" # Base URL of the state server, no trailing slash. Required for storage = "http".
+timeout = "5s"
+
+# Static headers attached verbatim to every request (e.g. authentication).
+# [state.http.request_headers]
+# authorization = "Bearer ..."
+
+[state.http.retry]
+enabled = true
+max_attempts = 4
+initial_backoff = "200ms"
+max_backoff = "2s"
+backoff_multiplier = 2
 
 [connectors]
 config_type = "local"

core/connectors/runtime/example_config/config.toml

@@ -45,7 +45,25 @@ enabled = false
 ca_file = "core/certs/iggy_ca_cert.pem"
 
 [state]
-path = "local_state"
+path = "local_state" # Used by storage = "file"
+storage = "file" # "file" | "http"
+
+# HTTP state storage backend, used when storage = "http". Source state is
+# stored at {url}/source_{key} with optimistic concurrency (ETag/If-Match)
+# and idempotent retries (Idempotency-Key).
+# [state.http]
+# url = "http://127.0.0.1:8080/connectors/state"
+# timeout = "5s"
+#
+# [state.http.request_headers]
+# authorization = "Bearer ..."
+#
+# [state.http.retry]
+# enabled = true
+# max_attempts = 4
+# initial_backoff = "200ms"
+# max_backoff = "2s"
+# backoff_multiplier = 2
 
 [connectors]
 config_type = "local"

core/connectors/runtime/src/api/auth.rs

@@ -25,6 +25,7 @@ use axum::{
 };
 use secrecy::ExposeSecret;
 use std::sync::Arc;
+use subtle::ConstantTimeEq;
 
 const API_KEY_HEADER: &str = "api-key";
 const PUBLIC_PATHS: &[&str] = &["/", "/health"];
@@ -50,7 +51,11 @@ pub async fn resolve_api_key(
         return Err(StatusCode::UNAUTHORIZED);
     };
 
-    if api_key != context.api_key.expose_secret() {
+    let matches: bool = api_key
+        .as_bytes()
+        .ct_eq(context.api_key.expose_secret().as_bytes())
+        .into();
+    if !matches {
         return Err(StatusCode::UNAUTHORIZED);
     }
 

core/connectors/runtime/src/api/source.rs

@@ -260,7 +260,6 @@ async fn restart_source(
             context.config_provider.as_ref(),
             &context.iggy_clients.producer,
             &context.metrics,
-            &context.state_path,
             &context,
         )
         .await?;

core/connectors/runtime/src/configs/runtime.rs

@@ -24,11 +24,13 @@ use figment::value::Dict;
 use figment::{Metadata, Profile, Provider};
 use iggy_common::IggyDuration;
 use iggy_common::defaults::{DEFAULT_ROOT_PASSWORD, DEFAULT_ROOT_USERNAME};
+use secrecy::{ExposeSecret, SecretString};
 use serde::{Deserialize, Serialize};
 use serde_with::{DisplayFromStr, serde_as};
 use std::collections::HashMap;
 use std::fmt::{Display, Formatter};
 use std::str::FromStr;
+use std::time::Duration;
 
 #[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)]
 pub struct TelemetryConfig {
@@ -345,11 +347,170 @@ impl Default for ConnectorsConfig {
 #[derive(Debug, Clone, Serialize, Deserialize, ConfigEnv)]
 pub struct StateConfig {
     pub path: String,
+    #[serde(default)]
+    #[config_env(leaf)]
+    pub storage: StateStorageKind,
+    #[serde(default)]
+    pub http: HttpStateConfig,
 }
 
 impl Display for StateConfig {
     fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
-        write!(f, "{{ path: {} }}", self.path)
+        write!(
+            f,
+            "{{ path: {}, storage: {}, http: {} }}",
+            self.path, self.storage, self.http
+        )
+    }
+}
+
+#[derive(
+    Debug,
+    Default,
+    Clone,
+    Copy,
+    PartialEq,
+    Eq,
+    Deserialize,
+    Serialize,
+    strum::Display,
+    strum::EnumString,
+)]
+#[serde(rename_all = "lowercase")]
+#[strum(serialize_all = "lowercase", ascii_case_insensitive)]
+pub enum StateStorageKind {
+    #[default]
+    File,
+    Http,
+}
+
+#[serde_as]
+#[derive(Debug, Clone, Serialize, Deserialize, ConfigEnv)]
+#[serde(default)]
+pub struct HttpStateConfig {
+    pub url: String,
+    #[config_env(leaf)]
+    #[serde_as(as = "DisplayFromStr")]
+    pub timeout: IggyDuration,
+    #[config_env(skip)]
+    #[serde(serialize_with = "serialize_secret_map")]
+    pub request_headers: HashMap<String, SecretString>,
+    pub retry: RetryConfig,
+}
+
+impl Default for HttpStateConfig {
+    fn default() -> Self {
+        Self {
+            url: String::new(),
+            timeout: IggyDuration::new_from_secs(5),
+            request_headers: HashMap::new(),
+            retry: default_state_retry(),
+        }
+    }
+}
+
+impl Display for HttpStateConfig {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        write!(
+            f,
+            "{{ url: {:?}, timeout: {}, request_headers: {:?}, retry: {} }}",
+            self.url,
+            self.timeout,
+            self.request_headers.keys(),
+            self.retry
+        )
+    }
+}
+
+fn default_state_retry() -> RetryConfig {
+    RetryConfig {
+        enabled: true,
+        max_attempts: 4,
+        initial_backoff: IggyDuration::new(Duration::from_millis(200)),
+        max_backoff: IggyDuration::new_from_secs(2),
+        backoff_multiplier: 2,
+    }
+}
+
+/// Exposes the header values on serialization, mirroring the
+/// `serialize_secret` treatment of `http.api_key`: the plaintext is the point
+/// when the config is rendered back out, while `Debug`/`Display` stay
+/// redacted through `SecretString`.
+fn serialize_secret_map<S: serde::Serializer>(
+    headers: &HashMap<String, SecretString>,
+    serializer: S,
+) -> Result<S::Ok, S::Error> {
+    serializer.collect_map(
+        headers
+            .iter()
+            .map(|(name, value)| (name, value.expose_secret())),
+    )
+}
+
+#[cfg(test)]
+mod state_config_tests {
+    use super::*;
+
+    #[test]
+    fn given_legacy_path_only_state_section_when_parsed_should_default_to_file_storage() {
+        let parsed: StateConfig = toml::from_str(r#"path = "local_state""#).expect("parse state");
+        assert_eq!(parsed.path, "local_state");
+        assert_eq!(parsed.storage, StateStorageKind::File);
+        assert!(parsed.http.url.is_empty());
+        assert_eq!(parsed.http.timeout.get_duration(), Duration::from_secs(5));
+    }
+
+    #[test]
+    fn given_http_state_section_when_parsed_should_populate_backend_config() {
+        let toml = r#"
+            path = "local_state"
+            storage = "http"
+
+            [http]
+            url = "http://127.0.0.1:8080/connectors/state"
+            timeout = "10s"
+
+            [http.request_headers]
+            authorization = "Bearer token"
+
+            [http.retry]
+            enabled = true
+            max_attempts = 7
+            initial_backoff = "100ms"
+            max_backoff = "1s"
+            backoff_multiplier = 3
+        "#;
+        let parsed: StateConfig = toml::from_str(toml).expect("parse state");
+        assert_eq!(parsed.storage, StateStorageKind::Http);
+        assert_eq!(parsed.http.url, "http://127.0.0.1:8080/connectors/state");
+        assert_eq!(parsed.http.timeout.get_duration(), Duration::from_secs(10));
+        assert!(parsed.http.request_headers.contains_key("authorization"));
+        assert_eq!(parsed.http.retry.max_attempts, 7);
+        assert_eq!(parsed.http.retry.backoff_multiplier, 3);
+    }
+
+    #[test]
+    fn given_unknown_storage_kind_when_parsed_should_fail() {
+        let result = toml::from_str::<StateConfig>(
+            r#"
+            path = "local_state"
+            storage = "s3"
+        "#,
+        );
+        assert!(result.is_err(), "unknown storage kinds must fail boot");
+    }
+
+    #[test]
+    fn given_state_config_when_displayed_should_not_render_header_values() {
+        let mut config = StateConfig::default();
+        config
+            .http
+            .request_headers
+            .insert("authorization".to_string(), "Bearer top-secret".into());
+        let display_output = config.to_string();
+        let debug_output = format!("{config:?}");
+        assert!(!display_output.contains("top-secret"), "{display_output}");
+        assert!(!debug_output.contains("top-secret"), "{debug_output}");
     }
 }
 
@@ -408,6 +569,8 @@ impl Default for StateConfig {
     fn default() -> Self {
         Self {
             path: "local_state".to_owned(),
+            storage: StateStorageKind::default(),
+            http: HttpStateConfig::default(),
         }
     }
 }