From 0274642cdbace7f3de76bac31a2be187f4b0ae9f Mon Sep 17 00:00:00 2001 From: Maciej Modzelewski Date: Thu, 20 Aug 2026 21:17:21 +0200 Subject: [PATCH] feat(connectors): store source state on an HTTP state server Source connectors checkpoint through the file state backend, which writes one file per source to local disk. That ties a source's cursor to the node it happens to run on, so replacing the node or failing over to another runtime instance rewinds the source. Add a second backend selected by state.storage. The HTTP provider stores each source at {url}/source_{key} on any HTTP-speaking store, so cursors outlive the process. Concurrency is optimistic: reads remember the returned ETag, writes are conditional (If-Match, or If-None-Match: * before the first write), and each logical save carries a stable Idempotency-Key so a server that committed a write but lost the response replays that outcome instead of failing the retry with a spurious 412. State failures are now classified in the SDK. 5xx, timeouts and connect failures retry with backoff and end as TransientState, so the batch is Nacked and the plugin re-polls. Version conflicts and revoked authorization are PermanentState and latch the provider, failing later saves without touching the network: another writer took over, and retrying would only mask the original error. A classified failure while loading an enabled source's state aborts startup instead of minting a failed plugin, because treating it as "no state" would silently rewind the source. The file backend stays the default and its unclassified load errors keep the per-connector failure path, so existing configs parse and behave as before. --- Cargo.lock | 4 + Cargo.toml | 1 + core/connectors/runtime/Cargo.toml | 4 + core/connectors/runtime/README.md | 54 + core/connectors/runtime/config.toml | 21 +- .../runtime/example_config/config.toml | 20 +- core/connectors/runtime/src/api/auth.rs | 7 +- core/connectors/runtime/src/api/source.rs | 1 - .../connectors/runtime/src/configs/runtime.rs | 165 ++- core/connectors/runtime/src/context.rs | 7 +- core/connectors/runtime/src/error.rs | 10 + core/connectors/runtime/src/main.rs | 8 +- core/connectors/runtime/src/manager/source.rs | 22 +- core/connectors/runtime/src/source.rs | 110 +- core/connectors/runtime/src/state.rs | 132 ++ core/connectors/runtime/src/state/http.rs | 1127 +++++++++++++++++ core/connectors/runtime/src/stream.rs | 102 +- core/connectors/sdk/src/lib.rs | 15 + .../tests/connectors/runtime/http_state.rs | 384 ++++++ .../tests/connectors/runtime/http_state.toml | 34 + .../http_state_config/random_source.toml | 37 + .../tests/connectors/runtime/mod.rs | 1 + 22 files changed, 2185 insertions(+), 81 deletions(-) create mode 100644 core/connectors/runtime/src/state/http.rs create mode 100644 core/integration/tests/connectors/runtime/http_state.rs create mode 100644 core/integration/tests/connectors/runtime/http_state.toml create mode 100644 core/integration/tests/connectors/runtime/http_state_config/random_source.toml diff --git a/Cargo.lock b/Cargo.lock index a08a49adb9..bc9698629a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6723,6 +6723,7 @@ dependencies = [ "async-trait", "axum", "axum-server", + "bytes", "clap", "configs", "configs_derive", @@ -6756,6 +6757,7 @@ dependencies = [ "serde_with", "serde_yaml_ng", "strum 0.28.0", + "subtle", "sysinfo 0.39.6", "system_stats", "tempfile", @@ -6766,6 +6768,8 @@ dependencies = [ "tracing", "tracing-opentelemetry", "tracing-subscriber", + "uuid", + "wiremock", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index eff5c6669e..3e74626b50 100644 --- a/Cargo.toml +++ b/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" diff --git a/core/connectors/runtime/Cargo.toml b/core/connectors/runtime/Cargo.toml index dba184d85f..aac80a8104 100644 --- a/core/connectors/runtime/Cargo.toml +++ b/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 } diff --git a/core/connectors/runtime/README.md b/core/connectors/runtime/README.md index 1c1339f49c..4829db9ef5 100644 --- a/core/connectors/runtime/README.md +++ b/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: ` 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. diff --git a/core/connectors/runtime/config.toml b/core/connectors/runtime/config.toml index 247a67e6b7..3f8d9b3b1b 100644 --- a/core/connectors/runtime/config.toml +++ b/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" diff --git a/core/connectors/runtime/example_config/config.toml b/core/connectors/runtime/example_config/config.toml index 4fad587f22..f33682f008 100644 --- a/core/connectors/runtime/example_config/config.toml +++ b/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" diff --git a/core/connectors/runtime/src/api/auth.rs b/core/connectors/runtime/src/api/auth.rs index b3c2ad594d..02d12be198 100644 --- a/core/connectors/runtime/src/api/auth.rs +++ b/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); } diff --git a/core/connectors/runtime/src/api/source.rs b/core/connectors/runtime/src/api/source.rs index b468f09265..a1c700edfc 100644 --- a/core/connectors/runtime/src/api/source.rs +++ b/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?; diff --git a/core/connectors/runtime/src/configs/runtime.rs b/core/connectors/runtime/src/configs/runtime.rs index 88442e1c8f..fa4e921616 100644 --- a/core/connectors/runtime/src/configs/runtime.rs +++ b/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, + 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( + headers: &HashMap, + serializer: S, +) -> Result { + 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::( + 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(), } } } diff --git a/core/connectors/runtime/src/context.rs b/core/connectors/runtime/src/context.rs index 1e62880fec..6b14066b52 100644 --- a/core/connectors/runtime/src/context.rs +++ b/core/connectors/runtime/src/context.rs @@ -18,6 +18,7 @@ use crate::configs::connectors::{ConnectorsConfigProvider, SinkConfig, SourceConfig}; use crate::configs::runtime::ConnectorsRuntimeConfig; use crate::metrics::Metrics; +use crate::state::StateStorageFactory; use crate::stream::IggyClients; use crate::{ FailedPlugin, SinkConnectorWrapper, SourceConnectorWrapper, @@ -43,7 +44,7 @@ pub struct RuntimeContext { pub metrics: Arc, pub start_time: IggyTimestamp, pub iggy_clients: Arc, - pub state_path: String, + pub state_factory: Arc, } #[allow(clippy::too_many_arguments)] @@ -57,7 +58,7 @@ pub fn init( failed_sources: &[FailedPlugin], config_provider: Box, iggy_clients: Arc, - state_path: String, + state_factory: Arc, ) -> RuntimeContext { let metrics = Arc::new(Metrics::init()); let mut sink_details = map_sinks(sinks_config, sink_wrappers); @@ -79,7 +80,7 @@ pub fn init( metrics, start_time: IggyTimestamp::now(), iggy_clients, - state_path, + state_factory, } } diff --git a/core/connectors/runtime/src/error.rs b/core/connectors/runtime/src/error.rs index a8d0ba7647..35be3bb17c 100644 --- a/core/connectors/runtime/src/error.rs +++ b/core/connectors/runtime/src/error.rs @@ -29,6 +29,15 @@ pub enum RuntimeError { FailedToSerializeRawMessages, #[error("Connector SDK error")] ConnectorSdkError(#[from] iggy_connector_sdk::Error), + /// A classified state-store failure while loading an enabled source's + /// state. Process-level: treating it as "no state" would silently rewind + /// the source, and parking the source as a failed plugin would hide a + /// store outage that a restart could clear. + #[error("Failed to load state for source connector '{connector_key}': {source}")] + StateLoadFailed { + connector_key: String, + source: iggy_connector_sdk::Error, + }, #[error("Iggy client error")] IggyClient(#[from] iggy::prelude::ClientError), #[error("Iggy error")] @@ -71,6 +80,7 @@ impl RuntimeError { RuntimeError::MissingIggyCredentials => "invalid_configuration", RuntimeError::InvalidConfiguration(_) => "invalid_configuration", RuntimeError::HttpRequestFailed(_) => "http_request_failed", + RuntimeError::StateLoadFailed { .. } => "state_load_failed", RuntimeError::TokenFileNotFound(_) => "invalid_configuration", RuntimeError::TokenFileReadError(_, _) => "invalid_configuration", RuntimeError::TokenFileEmpty(_) => "invalid_configuration", diff --git a/core/connectors/runtime/src/main.rs b/core/connectors/runtime/src/main.rs index 25677d3bd2..08311d7d1d 100644 --- a/core/connectors/runtime/src/main.rs +++ b/core/connectors/runtime/src/main.rs @@ -142,9 +142,7 @@ async fn main() -> Result<(), RuntimeError> { log::init_logging(&config.telemetry, &config.logging, VERSION); - std::fs::create_dir_all(&config.state.path).expect("Failed to create state directory"); - - info!("State will be stored in: {}", config.state.path); + let state_factory = state::factory_from_config(&config.state)?; let iggy_clients = Arc::new(stream::init(config.iggy.clone()).await?); @@ -161,7 +159,7 @@ async fn main() -> Result<(), RuntimeError> { let (sources, failed_sources) = source::init( sources_config.clone(), &iggy_clients.producer, - &config.state.path, + &state_factory, ) .await?; @@ -208,7 +206,7 @@ async fn main() -> Result<(), RuntimeError> { &failed_sources, connectors_config_provider, iggy_clients.clone(), - config.state.path.clone(), + state_factory, ); for (key, container) in sink_containers_by_key { if let Some(details) = context.sinks.get(&key).await { diff --git a/core/connectors/runtime/src/manager/source.rs b/core/connectors/runtime/src/manager/source.rs index 674b06e23f..8ef68e8b19 100644 --- a/core/connectors/runtime/src/manager/source.rs +++ b/core/connectors/runtime/src/manager/source.rs @@ -22,7 +22,6 @@ use crate::context::RuntimeContext; use crate::error::RuntimeError; use crate::metrics::Metrics; use crate::source; -use crate::state::{StateProvider, StateStorage}; use dashmap::DashMap; use dlopen2::wrapper::Container; use iggy::prelude::IggyClient; @@ -199,7 +198,6 @@ impl SourceManager { config: &SourceConfig, iggy_client: &IggyClient, metrics: &Arc, - state_path: &str, context: &Arc, ) -> Result<(), RuntimeError> { let details = self @@ -217,10 +215,19 @@ impl SourceManager { let plugin_id = PLUGIN_ID.fetch_add(1, Ordering::SeqCst); - let state_storage = source::get_state_storage(state_path, key); - let state = match &state_storage { - StateStorage::File(file) => file.load().await?, - }; + let state_storage = context.state_factory.storage_for(key)?; + let state = state_storage + .load() + .await + .map_err(|load_error| match load_error { + iggy_connector_sdk::Error::TransientState(_) + | iggy_connector_sdk::Error::PermanentState(_) + | iggy_connector_sdk::Error::StateLatched => RuntimeError::StateLoadFailed { + connector_key: key.to_string(), + source: load_error, + }, + other => RuntimeError::ConnectorSdkError(other), + })?; source::init_source( &container, @@ -268,7 +275,6 @@ impl SourceManager { config_provider: &dyn ConnectorsConfigProvider, iggy_client: &IggyClient, metrics: &Arc, - state_path: &str, context: &Arc, ) -> Result<(), RuntimeError> { let guard = { @@ -294,7 +300,7 @@ impl SourceManager { .map_err(|e| RuntimeError::InvalidConfiguration(e.to_string()))? .ok_or_else(|| RuntimeError::SourceNotFound(key.to_string()))?; - self.start_connector(key, &config, iggy_client, metrics, state_path, context) + self.start_connector(key, &config, iggy_client, metrics, context) .await?; info!("Source connector: {key} restarted successfully."); Ok(()) diff --git a/core/connectors/runtime/src/source.rs b/core/connectors/runtime/src/source.rs index d63496f0be..6b01bcb359 100644 --- a/core/connectors/runtime/src/source.rs +++ b/core/connectors/runtime/src/source.rs @@ -24,7 +24,8 @@ use iggy::prelude::{ }; use iggy_connector_sdk::encoders::avro::{AvroEncoderConfig, AvroStreamEncoder}; use iggy_connector_sdk::{ - ConnectorState, DecodedMessage, ProducedMessages, Schema, StreamEncoder, TopicMetadata, + ConnectorState, DecodedMessage, Error as SdkError, ProducedMessages, Schema, StreamEncoder, + TopicMetadata, source::{BatchResultCallback, HandleCallback, SourceBatchResult}, transforms::Transform, }; @@ -45,7 +46,7 @@ use crate::metrics::SourceLabels; use crate::{ FailedPlugin, PLUGIN_ID, RuntimeError, SourceApi, SourceConnector, SourceConnectorPlugin, SourceConnectorProducer, SourceConnectorWrapper, resolve_plugin_path, - state::{FileStateProvider, StateProvider, StateStorage}, + state::{StateStorage, StateStorageFactory}, transform, }; use iggy_connector_sdk::api::ConnectorStatus; @@ -76,18 +77,22 @@ pub(crate) fn cleanup_sender(plugin_id: u32) { /// Initializes all enabled source connectors. /// -/// Per-connector failures (path resolution, dlopen, state load, plugin init, +/// Per-connector failures (path resolution, dlopen, plugin init, /// producer/encoder/transform setup) are captured against the offending /// connector and do not abort the runtime. Connectors that fail before their /// FFI container can be loaded are returned in the second tuple element so /// they remain visible in health/status output. /// -/// Only system-level errors that prevent any connector from running (e.g. a -/// poisoned global state) are propagated as `Err`. +/// Only system-level errors that prevent any connector from running are +/// propagated as `Err`. That includes classified state-store failures +/// (`TransientState`/`PermanentState`/`StateLatched`) while loading an +/// enabled source's state: the store is unhealthy, so the process must fail +/// rather than rewind the source or mint a `FailedPlugin`. Unclassified +/// state-load failures (the file backend) keep the per-connector path. pub async fn init( source_configs: HashMap, iggy_client: &IggyClient, - state_path: &str, + state_factory: &Arc, ) -> Result<(HashMap, Vec), RuntimeError> { let mut source_connectors: HashMap = HashMap::new(); let mut failed_plugins: Vec = Vec::new(); @@ -124,25 +129,38 @@ pub async fn init( &config.version ); - let state_storage = get_state_storage(state_path, &key); - let state = match &state_storage { - StateStorage::File(file) => match file.load().await { - Ok(state) => state, - Err(error) => { - let message = format!("Failed to load source state: {error}"); - error!("Source: {name} ({key}) - {message}"); - failed_plugins.push(FailedPlugin::new( - plugin_id, - &key, - &name, - &config.path, - config.plugin_config_format, - config.enabled, - message, - )); - continue; - } - }, + let state_storage = state_factory.storage_for(&key)?; + let state = match state_storage.load().await { + Ok(state) => state, + Err( + load_error @ (SdkError::TransientState(_) + | SdkError::PermanentState(_) + | SdkError::StateLatched), + ) => { + // A classified failure means the state store is unhealthy, + // not the plugin. Treating it as "no state" would silently + // rewind the source, and parking it as a failed plugin would + // hide an outage the next restart could clear, so abort boot. + error!("Source: {name} ({key}) - failed to load state: {load_error}"); + return Err(RuntimeError::StateLoadFailed { + connector_key: key, + source: load_error, + }); + } + Err(error) => { + let message = format!("Failed to load source state: {error}"); + error!("Source: {name} ({key}) - {message}"); + failed_plugins.push(FailedPlugin::new( + plugin_id, + &key, + &name, + &config.path, + config.plugin_config_format, + config.enabled, + message, + )); + continue; + } }; if !source_connectors.contains_key(&path) { @@ -287,11 +305,6 @@ pub(crate) fn init_source( } } -pub(crate) fn get_state_storage(state_path: &str, key: &str) -> StateStorage { - let path = format!("{state_path}/source_{key}.state"); - StateStorage::File(FileStateProvider::new(path)) -} - pub(crate) async fn setup_source_producer( key: &str, config: &SourceConfig, @@ -525,25 +538,24 @@ pub(crate) async fn source_forwarding_loop( let mut state_saved = true; if let Some(state) = produced_messages.state { let state_save_start = Instant::now(); - match &state_storage { - StateStorage::File(file) => { - if let Err(error) = file.save(state).await { - state_saved = false; - let error_msg = format!( - "Failed to save state for source connector with ID: {plugin_id}. {error}" - ); - error!("{error_msg}"); - context.metrics.inc_errors_with_labels(&labels.counter); - context.sources.set_error(&plugin_key, &error_msg).await; - } else { - debug!("State saved for source connector with ID: {plugin_id}"); - let state_save_elapsed = state_save_start.elapsed(); - context.metrics.observe_stage_with_labels( - &labels.stage_state_save, - state_save_elapsed, - ); - state_save_us = Some(benchmark::as_micros(state_save_elapsed)); - } + match state_storage.save(state).await { + Ok(()) => { + debug!("State saved for source connector with ID: {plugin_id}"); + let state_save_elapsed = state_save_start.elapsed(); + context.metrics.observe_stage_with_labels( + &labels.stage_state_save, + state_save_elapsed, + ); + state_save_us = Some(benchmark::as_micros(state_save_elapsed)); + } + Err(error) => { + state_saved = false; + let error_msg = format!( + "Failed to save state for source connector with ID: {plugin_id}. {error}" + ); + error!("{error_msg}"); + context.metrics.inc_errors_with_labels(&labels.counter); + context.sources.set_error(&plugin_key, &error_msg).await; } } } else { diff --git a/core/connectors/runtime/src/state.rs b/core/connectors/runtime/src/state.rs index 021f4e1acc..d136e1cad9 100644 --- a/core/connectors/runtime/src/state.rs +++ b/core/connectors/runtime/src/state.rs @@ -15,15 +15,22 @@ // specific language governing permissions and limitations // under the License. +mod http; + +use crate::configs::runtime::{StateConfig, StateStorageKind}; +use crate::error::RuntimeError; use iggy_connector_sdk::{ConnectorState, Error}; use std::io::ErrorKind; use std::path::{Path, PathBuf}; +use std::sync::Arc; use strum::Display; use tokio::fs::{self, OpenOptions}; use tokio::io::AsyncWriteExt; use tokio::sync::Mutex; use tracing::{debug, error, info, warn}; +pub use http::{HttpStateFactory, HttpStateProvider}; + pub trait StateProvider { async fn load(&self) -> Result, Error>; async fn save(&self, state: ConnectorState) -> Result<(), Error>; @@ -34,6 +41,75 @@ pub trait StateProvider { pub enum StateStorage { #[strum(to_string = "file")] File(FileStateProvider), + #[strum(to_string = "http")] + Http(HttpStateProvider), +} + +impl StateStorage { + pub async fn load(&self) -> Result, Error> { + match self { + StateStorage::File(provider) => provider.load().await, + StateStorage::Http(provider) => provider.load().await, + } + } + + pub async fn save(&self, state: ConnectorState) -> Result<(), Error> { + match self { + StateStorage::File(provider) => provider.save(state).await, + StateStorage::Http(provider) => provider.save(state).await, + } + } +} + +/// Builds the configured [`StateStorageFactory`]. The only backend-selection +/// point in the runtime: a bad backend configuration fails here, at boot. +pub fn factory_from_config( + config: &StateConfig, +) -> Result, RuntimeError> { + match config.storage { + StateStorageKind::File => { + std::fs::create_dir_all(&config.path).map_err(|create_error| { + RuntimeError::InvalidConfiguration(format!( + "Failed to create state directory '{}': {create_error}", + config.path + )) + })?; + info!("State will be stored in: {}", config.path); + Ok(Arc::new(FileStateFactory::new(config.path.clone()))) + } + StateStorageKind::Http => { + let factory = HttpStateFactory::new(&config.http)?; + info!("State will be stored via HTTP at: {}", config.http.url); + Ok(Arc::new(factory)) + } + } +} + +/// Builds the per-connector [`StateStorage`]. Resolved once at startup and +/// shared via the runtime context, so API-driven restarts get their storage +/// from the same place as the initial start. Fallible because a storage +/// backend may have per-connector prerequisites, even though the file backend +/// never fails. +pub trait StateStorageFactory: Send + Sync { + fn storage_for(&self, connector_key: &str) -> Result; +} + +#[derive(Debug)] +pub struct FileStateFactory { + path: String, +} + +impl FileStateFactory { + pub fn new(path: String) -> Self { + Self { path } + } +} + +impl StateStorageFactory for FileStateFactory { + fn storage_for(&self, connector_key: &str) -> Result { + let path = format!("{}/source_{connector_key}.state", self.path); + Ok(StateStorage::File(FileStateProvider::new(path))) + } } #[derive(Debug)] @@ -220,6 +296,17 @@ mod tests { FileStateProvider::new(dir.path().join(name).to_string_lossy().to_string()) } + #[tokio::test] + async fn given_file_factory_when_storage_built_should_round_trip_state() { + let dir = TempDir::new().unwrap(); + let factory = FileStateFactory::new(dir.path().to_string_lossy().to_string()); + let storage = factory.storage_for("test").unwrap(); + storage.save(ConnectorState(vec![1, 2, 3])).await.unwrap(); + let loaded = storage.load().await.unwrap().unwrap(); + assert_eq!(loaded.0, vec![1, 2, 3]); + assert!(dir.path().join("source_test.state").is_file()); + } + #[tokio::test] async fn given_no_existing_file_when_loaded_should_return_none() { let dir = TempDir::new().unwrap(); @@ -397,6 +484,51 @@ mod tests { ); } + #[test] + fn given_default_config_when_factory_built_should_use_file_backend() { + let dir = TempDir::new().unwrap(); + let config = StateConfig { + path: dir.path().join("state").to_string_lossy().to_string(), + ..StateConfig::default() + }; + let factory = factory_from_config(&config).expect("file factory should build"); + assert!( + matches!(factory.storage_for("test"), Ok(StateStorage::File(_))), + "default storage must remain the file backend" + ); + assert!( + dir.path().join("state").is_dir(), + "the file arm still creates the state directory at boot" + ); + } + + #[test] + fn given_http_storage_without_url_when_factory_built_should_fail() { + let config = StateConfig { + storage: StateStorageKind::Http, + ..StateConfig::default() + }; + let result = factory_from_config(&config); + assert!( + matches!(result, Err(RuntimeError::InvalidConfiguration(ref message)) if message.contains("state.http.url")), + "http storage without a url must fail boot" + ); + } + + #[test] + fn given_http_storage_with_url_when_factory_built_should_use_http_backend() { + let mut config = StateConfig { + storage: StateStorageKind::Http, + ..StateConfig::default() + }; + config.http.url = "http://localhost:1/state".to_string(); + let factory = factory_from_config(&config).expect("http factory should build"); + assert!(matches!( + factory.storage_for("test"), + Ok(StateStorage::Http(_)) + )); + } + #[tokio::test] async fn given_target_path_is_directory_when_save_called_should_fail() { let dir = TempDir::new().unwrap(); diff --git a/core/connectors/runtime/src/state/http.rs b/core/connectors/runtime/src/state/http.rs new file mode 100644 index 0000000000..d9004e9452 --- /dev/null +++ b/core/connectors/runtime/src/state/http.rs @@ -0,0 +1,1127 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use crate::configs::runtime::{HttpStateConfig, RetryConfig}; +use crate::error::RuntimeError; +use crate::state::{StateProvider, StateStorage, StateStorageFactory}; +use bytes::Bytes; +use iggy_connector_sdk::{ConnectorState, Error}; +use reqwest::header::{ + CONTENT_TYPE, ETAG, HeaderMap, HeaderName, HeaderValue, IF_MATCH, IF_NONE_MATCH, RETRY_AFTER, +}; +use reqwest::{Method, StatusCode, Url}; +use reqwest_middleware::{ClientBuilder, ClientWithMiddleware, RequestBuilder}; +use reqwest_tracing::{SpanBackendWithUrl, TracingMiddleware}; +use secrecy::ExposeSecret; +use std::fmt; +use std::str::FromStr; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::time::Duration; +use tokio::sync::Mutex; +use tracing::{debug, error, info, warn}; +use uuid::Uuid; + +const IDEMPOTENCY_KEY_HEADER: &str = "idempotency-key"; +const OCTET_STREAM: &str = "application/octet-stream"; +const ERROR_BODY_SNIPPET_CHARS: usize = 256; + +/// Builds [`HttpStateProvider`]s that store source state at +/// `{url}/source_{connector_key}` on a generic HTTP state server. The shared +/// client (connection pool, static headers, timeout) is built eagerly so a +/// bad URL, TLS, or header configuration fails at boot. +pub struct HttpStateFactory { + client: ClientWithMiddleware, + base_url: Url, + retry: RetryPolicy, +} + +impl HttpStateFactory { + pub fn new(config: &HttpStateConfig) -> Result { + if config.url.trim().is_empty() { + return Err(RuntimeError::InvalidConfiguration( + "state.http.url is required when state.storage = \"http\"".to_string(), + )); + } + let base_url = Url::parse(&config.url).map_err(|parse_error| { + RuntimeError::InvalidConfiguration(format!( + "Invalid state.http.url '{}': {parse_error}", + config.url + )) + })?; + if base_url.cannot_be_a_base() || !matches!(base_url.scheme(), "http" | "https") { + return Err(RuntimeError::InvalidConfiguration(format!( + "state.http.url must be an http(s) URL, got '{}'", + config.url + ))); + } + + let mut headers = HeaderMap::new(); + for (name, value) in &config.request_headers { + let header_name = HeaderName::from_str(name).map_err(|header_error| { + RuntimeError::InvalidConfiguration(format!( + "Invalid state.http header name '{name}': {header_error}" + )) + })?; + // The parse error never echoes the value, and marking it + // sensitive keeps it out of any Debug output downstream. + let mut header_value = + HeaderValue::from_str(value.expose_secret()).map_err(|header_error| { + RuntimeError::InvalidConfiguration(format!( + "Invalid state.http header value for '{name}': {header_error}" + )) + })?; + header_value.set_sensitive(true); + headers.insert(header_name, header_value); + } + + // Redirects are disabled: a redirected conditional PUT could drop the + // body or method, so any 3xx is surfaced as a protocol violation. + let client = reqwest::Client::builder() + .default_headers(headers) + .timeout(config.timeout.get_duration()) + .redirect(reqwest::redirect::Policy::none()) + .build() + .map_err(|client_error| { + RuntimeError::InvalidConfiguration(format!( + "Failed to build the state HTTP client: {client_error}" + )) + })?; + let client = ClientBuilder::new(client) + .with(TracingMiddleware::::new()) + .build(); + + Ok(Self { + client, + base_url, + retry: RetryPolicy::from(&config.retry), + }) + } +} + +impl StateStorageFactory for HttpStateFactory { + fn storage_for(&self, connector_key: &str) -> Result { + let mut resource_url = self.base_url.clone(); + { + let mut segments = resource_url.path_segments_mut().map_err(|_| { + RuntimeError::InvalidConfiguration(format!( + "State URL cannot host per-connector resources: {}", + self.base_url + )) + })?; + // push() percent-encodes the segment, so any connector key is safe + // in the path. pop_if_empty() tolerates a trailing slash in the + // configured base URL. + segments.pop_if_empty(); + segments.push(&format!("source_{connector_key}")); + } + Ok(StateStorage::Http(HttpStateProvider::new( + self.client.clone(), + resource_url, + self.retry.clone(), + ))) + } +} + +impl fmt::Debug for HttpStateFactory { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("HttpStateFactory") + .field("base_url", &self.base_url.as_str()) + .field("retry", &self.retry) + .finish_non_exhaustive() + } +} + +/// Stores one source connector's state on an HTTP state server with +/// optimistic concurrency: every read remembers the returned `ETag`, every +/// write is conditional (`If-Match`, or `If-None-Match: *` before the first +/// write), and every write carries an `Idempotency-Key` that is stable across +/// the retries of one logical save. +/// +/// Errors are classified per the state protocol contract: `5xx`, timeouts and +/// connect failures are [`Error::TransientState`] after bounded retries; +/// version conflicts, lost authorization and protocol violations are +/// [`Error::PermanentState`]. Any permanent save error latches the provider, +/// so later saves fail fast with [`Error::StateLatched`] without touching the +/// network. +pub struct HttpStateProvider { + client: ClientWithMiddleware, + resource_url: Url, + retry: RetryPolicy, + version: Mutex, + latched: AtomicBool, +} + +impl HttpStateProvider { + pub(crate) fn new(client: ClientWithMiddleware, resource_url: Url, retry: RetryPolicy) -> Self { + Self { + client, + resource_url, + retry, + version: Mutex::new(TrackedVersion::Unknown), + latched: AtomicBool::new(false), + } + } + + async fn load_with_version( + &self, + version: &mut TrackedVersion, + ) -> Result, Error> { + let response = self + .execute_with_retry(Method::GET, "load", |request| request) + .await?; + match response.status() { + StatusCode::OK => { + let etag = required_etag(&response, "load", &self.resource_url)?; + let bytes = response.bytes().await.map_err(|read_error| { + Error::TransientState(format!( + "load GET {} failed reading the state body: {read_error}", + self.resource_url + )) + })?; + debug!( + "Loaded state from {} ({} bytes)", + self.resource_url, + bytes.len() + ); + *version = TrackedVersion::Etag(etag); + // A zero-length body is valid state, unlike the file backend + // where an empty file means "no state yet". + Ok(Some(ConnectorState(bytes.to_vec()))) + } + StatusCode::NOT_FOUND => { + info!("No state stored at {}, starting fresh", self.resource_url); + *version = TrackedVersion::Absent; + Ok(None) + } + status => Err(Error::PermanentState( + describe_failure("load", status, &self.resource_url, response).await, + )), + } + } + + async fn execute_with_retry( + &self, + method: Method, + operation: &str, + build_request: F, + ) -> Result + where + F: Fn(RequestBuilder) -> RequestBuilder, + { + let max_attempts = if self.retry.enabled { + self.retry.max_attempts + } else { + 0 + }; + let mut attempt = 0u32; + loop { + let request = build_request( + self.client + .request(method.clone(), self.resource_url.clone()), + ); + let failure = match request.send().await { + Ok(response) if !is_transient_status(response.status()) => return Ok(response), + Ok(response) => TransientFailure::Status(response), + Err(send_error) => TransientFailure::Send(send_error), + }; + if attempt >= max_attempts { + return Err(Error::TransientState(failure.describe( + operation, + &self.resource_url, + attempt + 1, + ))); + } + let delay = self.retry.delay_for(attempt, failure.retry_after()); + warn!( + "{operation} against {} hit a transient failure, retrying in {delay:?} (attempt {} of {})", + self.resource_url, + attempt + 1, + max_attempts + 1 + ); + tokio::time::sleep(delay).await; + attempt += 1; + } + } + + fn latch(&self) { + self.latched.store(true, Ordering::Release); + error!( + "State provider for {} latched after a permanent save error; further saves fail fast until restart", + self.resource_url + ); + } +} + +impl StateProvider for HttpStateProvider { + async fn load(&self) -> Result, Error> { + let mut version = self.version.lock().await; + self.load_with_version(&mut version).await + } + + async fn save(&self, state: ConnectorState) -> Result<(), Error> { + if self.latched.load(Ordering::Acquire) { + return Err(Error::StateLatched); + } + let mut version = self.version.lock().await; + // Re-check under the lock: a concurrent save may have latched while + // this one was waiting. + if self.latched.load(Ordering::Acquire) { + return Err(Error::StateLatched); + } + if matches!(*version, TrackedVersion::Unknown) { + // Never issue an unconditional write. The runtime always loads at + // init, so this safety net should not trigger in practice. + warn!( + "Saving state to {} before any load; loading first to resolve the stored version", + self.resource_url + ); + self.load_with_version(&mut version).await?; + } + let (condition_name, condition_value) = match &*version { + TrackedVersion::Etag(etag) => ( + IF_MATCH, + HeaderValue::from_str(etag).map_err(|_| { + Error::PermanentState(format!( + "Tracked ETag for {} is not a valid header value", + self.resource_url + )) + })?, + ), + TrackedVersion::Absent => (IF_NONE_MATCH, HeaderValue::from_static("*")), + TrackedVersion::Unknown => { + return Err(Error::PermanentState(format!( + "State version for {} is unresolved after load", + self.resource_url + ))); + } + }; + + // One key per logical save, byte-identical across every retry of that + // save, so a server that committed the write but lost the response + // can replay the original outcome instead of failing the retry with a + // spurious 412. + let idempotency_key = Uuid::new_v4().to_string(); + let body = Bytes::from(state.0); + let response = self + .execute_with_retry(Method::PUT, "save", |request| { + request + .header(condition_name.clone(), condition_value.clone()) + .header(IDEMPOTENCY_KEY_HEADER, idempotency_key.as_str()) + .header(CONTENT_TYPE, OCTET_STREAM) + .body(body.clone()) + }) + .await?; + + match response.status() { + StatusCode::OK | StatusCode::CREATED | StatusCode::NO_CONTENT => { + match required_etag(&response, "save", &self.resource_url) { + Ok(etag) => { + debug!( + "Saved state to {} ({} bytes)", + self.resource_url, + body.len() + ); + *version = TrackedVersion::Etag(etag); + Ok(()) + } + Err(etag_error) => { + self.latch(); + Err(etag_error) + } + } + } + status => { + let mut message = + describe_failure("save", status, &self.resource_url, response).await; + if matches!( + status, + StatusCode::PRECONDITION_FAILED | StatusCode::CONFLICT + ) { + message.push_str( + "; the stored state changed under this writer or its write authority was revoked", + ); + } + self.latch(); + Err(Error::PermanentState(message)) + } + } + } +} + +impl fmt::Debug for HttpStateProvider { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("HttpStateProvider") + .field("resource_url", &self.resource_url.as_str()) + .field("retry", &self.retry) + .field("latched", &self.latched.load(Ordering::Relaxed)) + .finish_non_exhaustive() + } +} + +#[derive(Debug)] +enum TrackedVersion { + Unknown, + Absent, + Etag(String), +} + +#[derive(Debug, Clone)] +pub(crate) struct RetryPolicy { + enabled: bool, + max_attempts: u32, + initial_backoff: Duration, + max_backoff: Duration, + backoff_multiplier: u32, +} + +impl RetryPolicy { + fn delay_for(&self, attempt: u32, retry_after: Option) -> Duration { + // Retry-After is honored but capped at max_backoff so a hostile or + // misconfigured server cannot stall the forwarding loop indefinitely. + if let Some(retry_after) = retry_after { + return retry_after.min(self.max_backoff); + } + let factor = self + .backoff_multiplier + .checked_pow(attempt) + .unwrap_or(u32::MAX); + self.initial_backoff + .saturating_mul(factor) + .min(self.max_backoff) + } +} + +impl From<&RetryConfig> for RetryPolicy { + fn from(config: &RetryConfig) -> Self { + Self { + enabled: config.enabled, + max_attempts: config.max_attempts, + initial_backoff: config.initial_backoff.get_duration(), + max_backoff: config.max_backoff.get_duration(), + backoff_multiplier: config.backoff_multiplier, + } + } +} + +enum TransientFailure { + Status(reqwest::Response), + Send(reqwest_middleware::Error), +} + +impl TransientFailure { + fn retry_after(&self) -> Option { + match self { + TransientFailure::Status(response) => response + .headers() + .get(RETRY_AFTER) + .and_then(|value| value.to_str().ok()) + .and_then(|value| value.trim().parse::().ok()) + .map(Duration::from_secs), + TransientFailure::Send(_) => None, + } + } + + fn describe(&self, operation: &str, url: &Url, attempts: u32) -> String { + match self { + TransientFailure::Status(response) => format!( + "{operation} against {url} still failing with HTTP {} after {attempts} attempts", + response.status() + ), + TransientFailure::Send(send_error) => { + format!( + "{operation} against {url} still failing after {attempts} attempts: {send_error}" + ) + } + } + } +} + +fn is_transient_status(status: StatusCode) -> bool { + status == StatusCode::TOO_EARLY + || status == StatusCode::TOO_MANY_REQUESTS + || status.is_server_error() +} + +fn required_etag( + response: &reqwest::Response, + operation: &str, + url: &Url, +) -> Result { + response + .headers() + .get(ETAG) + .and_then(|value| value.to_str().ok()) + .map(str::to_owned) + .ok_or_else(|| { + Error::PermanentState(format!( + "{operation} against {url} succeeded with HTTP {} but returned no usable ETag; \ + the state server violates the protocol contract", + response.status() + )) + }) +} + +async fn describe_failure( + operation: &str, + status: StatusCode, + url: &Url, + response: reqwest::Response, +) -> String { + let detail = match response.text().await { + Ok(body) if !body.is_empty() => { + let snippet: String = body.chars().take(ERROR_BODY_SNIPPET_CHARS).collect(); + format!(" - {snippet}") + } + _ => String::new(), + }; + format!("{operation} against {url} returned HTTP {status}{detail}") +} + +#[cfg(test)] +mod tests { + use super::*; + use iggy_common::IggyDuration; + use secrecy::SecretString; + use std::collections::HashMap; + use std::sync::Arc; + use std::sync::atomic::AtomicU64; + use std::time::Instant; + use wiremock::matchers::{header, method, path}; + use wiremock::{Mock, MockServer, Request, Respond, ResponseTemplate}; + + const RESOURCE_PATH: &str = "/source_test"; + + fn test_config(url: &str) -> HttpStateConfig { + HttpStateConfig { + url: url.to_string(), + timeout: IggyDuration::new(Duration::from_secs(5)), + request_headers: HashMap::new(), + retry: RetryConfig { + enabled: true, + max_attempts: 2, + initial_backoff: IggyDuration::new(Duration::from_millis(1)), + max_backoff: IggyDuration::new(Duration::from_millis(5)), + backoff_multiplier: 2, + }, + } + } + + fn storage_for(config: &HttpStateConfig) -> StateStorage { + HttpStateFactory::new(config) + .expect("test factory should build") + .storage_for("test") + .expect("test storage should build") + } + + fn storage(server: &MockServer) -> StateStorage { + storage_for(&test_config(&server.uri())) + } + + fn ok_with_etag(etag: &str) -> ResponseTemplate { + ResponseTemplate::new(200).insert_header("etag", etag) + } + + async fn requests_of(server: &MockServer, http_method: &str) -> Vec { + server + .received_requests() + .await + .expect("request recording is enabled") + .into_iter() + .filter(|request| request.method.as_str() == http_method) + .collect() + } + + #[tokio::test] + async fn given_stored_state_when_loaded_should_return_bytes_and_track_etag() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path(RESOURCE_PATH)) + .respond_with(ok_with_etag("\"v1\"").set_body_bytes(vec![1, 2, 3])) + .mount(&server) + .await; + Mock::given(method("PUT")) + .and(path(RESOURCE_PATH)) + .and(header("if-match", "\"v1\"")) + .respond_with(ok_with_etag("\"v2\"")) + .mount(&server) + .await; + + let storage = storage(&server); + let loaded = storage.load().await.unwrap().unwrap(); + assert_eq!(loaded.0, vec![1, 2, 3]); + storage + .save(ConnectorState(vec![4, 5])) + .await + .expect("save with the tracked ETag should hit the If-Match mock"); + } + + #[tokio::test] + async fn given_empty_body_when_loaded_should_return_empty_state() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path(RESOURCE_PATH)) + .respond_with(ok_with_etag("\"v1\"")) + .mount(&server) + .await; + + let loaded = storage(&server).load().await.unwrap(); + assert_eq!( + loaded.expect("zero-length body is valid state").0, + Vec::::new() + ); + } + + #[tokio::test] + async fn given_missing_state_when_loaded_should_return_none_then_create_on_save() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path(RESOURCE_PATH)) + .respond_with(ResponseTemplate::new(404)) + .mount(&server) + .await; + Mock::given(method("PUT")) + .and(path(RESOURCE_PATH)) + .and(header("if-none-match", "*")) + .and(header("content-type", OCTET_STREAM)) + .respond_with(ResponseTemplate::new(201).insert_header("etag", "\"v1\"")) + .mount(&server) + .await; + + let storage = storage(&server); + assert!(storage.load().await.unwrap().is_none()); + storage + .save(ConnectorState(vec![7])) + .await + .expect("first save should create via If-None-Match: *"); + let puts = requests_of(&server, "PUT").await; + assert_eq!(puts.len(), 1); + assert!(puts[0].headers.get(IDEMPOTENCY_KEY_HEADER).is_some()); + } + + #[tokio::test] + async fn given_success_without_etag_when_loaded_should_classify_permanent() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path(RESOURCE_PATH)) + .respond_with(ResponseTemplate::new(200).set_body_bytes(vec![1])) + .mount(&server) + .await; + + let result = storage(&server).load().await; + assert!( + matches!(result, Err(Error::PermanentState(_))), + "missing ETag on load must be a protocol violation, got {result:?}" + ); + } + + #[tokio::test] + async fn given_terminal_statuses_when_loaded_should_classify_permanent() { + for status in [401u16, 403, 400] { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path(RESOURCE_PATH)) + .respond_with(ResponseTemplate::new(status)) + .mount(&server) + .await; + + let result = storage(&server).load().await; + assert!( + matches!(result, Err(Error::PermanentState(_))), + "HTTP {status} on load must be permanent, got {result:?}" + ); + } + } + + #[tokio::test] + async fn given_server_error_then_success_when_loaded_should_retry() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path(RESOURCE_PATH)) + .respond_with(ResponseTemplate::new(503)) + .up_to_n_times(1) + .mount(&server) + .await; + Mock::given(method("GET")) + .and(path(RESOURCE_PATH)) + .respond_with(ok_with_etag("\"v1\"").set_body_bytes(vec![9])) + .mount(&server) + .await; + + let loaded = storage(&server).load().await.unwrap().unwrap(); + assert_eq!(loaded.0, vec![9]); + } + + #[tokio::test] + async fn given_persistent_server_errors_when_loaded_should_exhaust_as_transient() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path(RESOURCE_PATH)) + .respond_with(ResponseTemplate::new(503)) + .expect(3) // initial try + max_attempts retries + .mount(&server) + .await; + + let result = storage(&server).load().await; + assert!( + matches!(result, Err(Error::TransientState(_))), + "exhausted retries must be transient, got {result:?}" + ); + } + + #[tokio::test] + async fn given_retry_after_when_retrying_should_wait_at_least_that_long() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path(RESOURCE_PATH)) + .respond_with(ResponseTemplate::new(429).insert_header("retry-after", "1")) + .up_to_n_times(1) + .mount(&server) + .await; + Mock::given(method("GET")) + .and(path(RESOURCE_PATH)) + .respond_with(ok_with_etag("\"v1\"")) + .mount(&server) + .await; + + let mut config = test_config(&server.uri()); + config.retry.max_backoff = IggyDuration::new(Duration::from_secs(2)); + let started = Instant::now(); + storage_for(&config).load().await.unwrap(); + assert!( + started.elapsed() >= Duration::from_secs(1), + "Retry-After: 1 should delay the retry, elapsed {:?}", + started.elapsed() + ); + } + + #[tokio::test] + async fn given_unresponsive_server_when_loaded_should_classify_transient() { + // Bound but never accepted: the request times out instead of racing + // other tests for a recycled port. + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let port = listener.local_addr().unwrap().port(); + let mut config = test_config(&format!("http://127.0.0.1:{port}")); + config.timeout = IggyDuration::new(Duration::from_millis(200)); + config.retry.enabled = false; + + let result = storage_for(&config).load().await; + assert!( + matches!(result, Err(Error::TransientState(_))), + "a timed-out request must be transient, got {result:?}" + ); + } + + #[tokio::test] + async fn given_version_conflict_when_saved_should_latch() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path(RESOURCE_PATH)) + .respond_with(ResponseTemplate::new(404)) + .mount(&server) + .await; + Mock::given(method("PUT")) + .and(path(RESOURCE_PATH)) + .respond_with(ResponseTemplate::new(412)) + .mount(&server) + .await; + + let storage = storage(&server); + storage.load().await.unwrap(); + let result = storage.save(ConnectorState(vec![1])).await; + assert!(matches!(result, Err(Error::PermanentState(_)))); + + let requests_before = requests_of(&server, "PUT").await.len(); + let latched = storage.save(ConnectorState(vec![2])).await; + assert!( + matches!(latched, Err(Error::StateLatched)), + "expected StateLatched, got {latched:?}" + ); + assert_eq!( + requests_of(&server, "PUT").await.len(), + requests_before, + "a latched save must not touch the network" + ); + } + + #[tokio::test] + async fn given_forbidden_save_should_latch() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path(RESOURCE_PATH)) + .respond_with(ResponseTemplate::new(404)) + .mount(&server) + .await; + Mock::given(method("PUT")) + .and(path(RESOURCE_PATH)) + .respond_with(ResponseTemplate::new(403)) + .mount(&server) + .await; + + let storage = storage(&server); + storage.load().await.unwrap(); + assert!(matches!( + storage.save(ConnectorState(vec![1])).await, + Err(Error::PermanentState(_)) + )); + assert!(matches!( + storage.save(ConnectorState(vec![2])).await, + Err(Error::StateLatched) + )); + } + + #[tokio::test] + async fn given_save_success_without_etag_should_latch() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path(RESOURCE_PATH)) + .respond_with(ResponseTemplate::new(404)) + .mount(&server) + .await; + Mock::given(method("PUT")) + .and(path(RESOURCE_PATH)) + .respond_with(ResponseTemplate::new(200)) + .mount(&server) + .await; + + let storage = storage(&server); + storage.load().await.unwrap(); + assert!(matches!( + storage.save(ConnectorState(vec![1])).await, + Err(Error::PermanentState(_)) + )); + assert!(matches!( + storage.save(ConnectorState(vec![2])).await, + Err(Error::StateLatched) + )); + } + + #[tokio::test] + async fn given_transient_save_failures_when_retried_should_reuse_idempotency_key_and_body() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path(RESOURCE_PATH)) + .respond_with(ResponseTemplate::new(404)) + .mount(&server) + .await; + Mock::given(method("PUT")) + .and(path(RESOURCE_PATH)) + .respond_with(ResponseTemplate::new(503)) + .up_to_n_times(2) + .mount(&server) + .await; + Mock::given(method("PUT")) + .and(path(RESOURCE_PATH)) + .respond_with(ok_with_etag("\"v1\"")) + .mount(&server) + .await; + + let storage = storage(&server); + storage.load().await.unwrap(); + storage.save(ConnectorState(vec![1, 2, 3])).await.unwrap(); + + let puts = requests_of(&server, "PUT").await; + assert_eq!(puts.len(), 3, "one initial try plus two retries"); + let first_key = puts[0] + .headers + .get(IDEMPOTENCY_KEY_HEADER) + .expect("idempotency key must be present") + .clone(); + for put in &puts { + assert_eq!( + put.headers.get(IDEMPOTENCY_KEY_HEADER), + Some(&first_key), + "every retry of one logical save must reuse the same key" + ); + assert_eq!(put.body, vec![1u8, 2, 3], "retries must be byte-identical"); + assert_eq!( + put.headers.get("if-none-match").map(|v| v.as_bytes()), + Some(b"*".as_slice()) + ); + } + } + + #[tokio::test] + async fn given_two_logical_saves_should_mint_fresh_idempotency_keys() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path(RESOURCE_PATH)) + .respond_with(ResponseTemplate::new(404)) + .mount(&server) + .await; + Mock::given(method("PUT")) + .and(path(RESOURCE_PATH)) + .respond_with(ok_with_etag("\"v1\"")) + .mount(&server) + .await; + + let storage = storage(&server); + storage.load().await.unwrap(); + storage.save(ConnectorState(vec![1])).await.unwrap(); + storage.save(ConnectorState(vec![2])).await.unwrap(); + + let puts = requests_of(&server, "PUT").await; + assert_eq!(puts.len(), 2); + assert_ne!( + puts[0].headers.get(IDEMPOTENCY_KEY_HEADER), + puts[1].headers.get(IDEMPOTENCY_KEY_HEADER), + "a new state value must get a new idempotency key" + ); + } + + #[tokio::test] + async fn given_persistent_save_errors_should_be_transient_and_not_latch() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path(RESOURCE_PATH)) + .respond_with(ResponseTemplate::new(404)) + .mount(&server) + .await; + Mock::given(method("PUT")) + .and(path(RESOURCE_PATH)) + .respond_with(ResponseTemplate::new(503)) + .mount(&server) + .await; + + let storage = storage(&server); + storage.load().await.unwrap(); + assert!(matches!( + storage.save(ConnectorState(vec![1])).await, + Err(Error::TransientState(_)) + )); + let puts_after_first = requests_of(&server, "PUT").await.len(); + assert!( + matches!( + storage.save(ConnectorState(vec![2])).await, + Err(Error::TransientState(_)) + ), + "transient failures must not latch" + ); + assert!( + requests_of(&server, "PUT").await.len() > puts_after_first, + "the next save must reach the network after a transient failure" + ); + } + + #[tokio::test] + async fn given_unloaded_provider_when_saved_should_load_before_writing() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path(RESOURCE_PATH)) + .respond_with(ResponseTemplate::new(404)) + .mount(&server) + .await; + Mock::given(method("PUT")) + .and(path(RESOURCE_PATH)) + .and(header("if-none-match", "*")) + .respond_with(ok_with_etag("\"v1\"")) + .mount(&server) + .await; + + storage(&server) + .save(ConnectorState(vec![1])) + .await + .expect("save before load should resolve the version first"); + + let all = server.received_requests().await.unwrap(); + assert_eq!( + all[0].method.as_str(), + "GET", + "version must be resolved before writing" + ); + assert_eq!(all[1].method.as_str(), "PUT"); + } + + /// Conditional PUT responder: enforces `If-None-Match: *` before the first + /// write and `If-Match: "v{n}"` afterwards, answering 412 on any mismatch. + struct VersionedPut(Arc); + + impl Respond for VersionedPut { + fn respond(&self, request: &Request) -> ResponseTemplate { + let current = self.0.load(Ordering::SeqCst); + let matches = if current == 0 { + request + .headers + .get("if-none-match") + .map(|value| value.as_bytes() == b"*") + .unwrap_or(false) + } else { + request + .headers + .get("if-match") + .and_then(|value| value.to_str().ok()) + .map(|value| value == format!("\"v{current}\"")) + .unwrap_or(false) + }; + if !matches { + return ResponseTemplate::new(412); + } + let next = current + 1; + self.0.store(next, Ordering::SeqCst); + ResponseTemplate::new(200).insert_header("etag", format!("\"v{next}\"").as_str()) + } + } + + #[tokio::test] + async fn given_concurrent_saves_when_completed_should_serialize_and_chain_etags() { + let server = MockServer::start().await; + let version = Arc::new(AtomicU64::new(0)); + Mock::given(method("GET")) + .and(path(RESOURCE_PATH)) + .respond_with(ResponseTemplate::new(404)) + .mount(&server) + .await; + Mock::given(method("PUT")) + .and(path(RESOURCE_PATH)) + .respond_with(VersionedPut(version.clone())) + .mount(&server) + .await; + + let storage = Arc::new(storage(&server)); + storage.load().await.unwrap(); + let mut handles = Vec::new(); + for value in 0u8..8 { + let storage = storage.clone(); + handles.push(tokio::spawn(async move { + storage.save(ConnectorState(vec![value])).await + })); + } + for handle in handles { + handle + .await + .unwrap() + .expect("serialized saves must all commit against the fresh ETag"); + } + assert_eq!(version.load(Ordering::SeqCst), 8); + } + + #[tokio::test] + async fn given_configured_headers_when_requesting_should_attach_them() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path(RESOURCE_PATH)) + .and(header("authorization", "Bearer secret-token")) + .respond_with(ResponseTemplate::new(404)) + .mount(&server) + .await; + + let mut config = test_config(&server.uri()); + config.request_headers.insert( + "authorization".to_string(), + SecretString::from("Bearer secret-token"), + ); + assert!( + storage_for(&config).load().await.unwrap().is_none(), + "the mock only matches when the configured header is attached" + ); + } + + #[test] + fn given_secret_headers_when_formatted_should_not_leak_values() { + let mut config = test_config("http://localhost:1/state"); + config.request_headers.insert( + "authorization".to_string(), + SecretString::from("Bearer secret-token"), + ); + let debug_output = format!("{config:?}"); + let display_output = format!("{config}"); + assert!(!debug_output.contains("secret-token"), "{debug_output}"); + assert!(!display_output.contains("secret-token"), "{display_output}"); + assert!( + display_output.contains("authorization"), + "header names stay visible for operators: {display_output}" + ); + } + + #[test] + fn given_empty_url_when_factory_built_should_fail() { + let result = HttpStateFactory::new(&test_config(" ")); + assert!( + matches!(result, Err(RuntimeError::InvalidConfiguration(ref message)) if message.contains("state.http.url")), + "expected InvalidConfiguration about state.http.url" + ); + } + + #[test] + fn given_non_http_scheme_when_factory_built_should_fail() { + assert!(HttpStateFactory::new(&test_config("ftp://example.com/state")).is_err()); + assert!(HttpStateFactory::new(&test_config("not a url")).is_err()); + } + + #[test] + fn given_invalid_header_name_when_factory_built_should_fail() { + let mut config = test_config("http://localhost:1/state"); + config + .request_headers + .insert("bad header".to_string(), SecretString::from("value")); + assert!(HttpStateFactory::new(&config).is_err()); + } + + #[test] + fn given_connector_key_when_storage_built_should_percent_encode_the_segment() { + let factory = HttpStateFactory::new(&test_config("http://localhost:1/state")).unwrap(); + let StateStorage::Http(provider) = factory.storage_for("a b/c").unwrap() else { + panic!("http factory must build http storage"); + }; + assert_eq!(provider.resource_url.path(), "/state/source_a%20b%2Fc"); + } + + #[test] + fn given_trailing_slash_base_url_when_storage_built_should_not_double_slash() { + let factory = HttpStateFactory::new(&test_config("http://localhost:1/state/")).unwrap(); + let StateStorage::Http(provider) = factory.storage_for("test").unwrap() else { + panic!("http factory must build http storage"); + }; + assert_eq!(provider.resource_url.path(), "/state/source_test"); + } + + #[test] + fn given_backoff_policy_when_delays_computed_should_grow_and_cap() { + let policy = RetryPolicy { + enabled: true, + max_attempts: 5, + initial_backoff: Duration::from_millis(100), + max_backoff: Duration::from_millis(350), + backoff_multiplier: 2, + }; + assert_eq!(policy.delay_for(0, None), Duration::from_millis(100)); + assert_eq!(policy.delay_for(1, None), Duration::from_millis(200)); + assert_eq!(policy.delay_for(2, None), Duration::from_millis(350)); + assert_eq!(policy.delay_for(30, None), Duration::from_millis(350)); + } + + #[test] + fn given_retry_after_when_delay_computed_should_honor_but_cap_it() { + let policy = RetryPolicy { + enabled: true, + max_attempts: 5, + initial_backoff: Duration::from_millis(1), + max_backoff: Duration::from_secs(2), + backoff_multiplier: 2, + }; + assert_eq!( + policy.delay_for(0, Some(Duration::from_secs(1))), + Duration::from_secs(1) + ); + assert_eq!( + policy.delay_for(0, Some(Duration::from_secs(3600))), + Duration::from_secs(2), + "a hostile Retry-After must not stall the loop past max_backoff" + ); + } +} diff --git a/core/connectors/runtime/src/stream.rs b/core/connectors/runtime/src/stream.rs index 574fdd34c5..4082ffc8ae 100644 --- a/core/connectors/runtime/src/stream.rs +++ b/core/connectors/runtime/src/stream.rs @@ -70,22 +70,28 @@ pub struct IggyClients { } pub async fn init(config: IggyConfig) -> Result { + let consumer = create_client(&config).await?; + let producer = create_client(&config).await?; + let iggy_clients = IggyClients { producer, consumer }; + Ok(iggy_clients) +} + +/// Builds the authenticated connection string for `config`, resolving a +/// `file:`-prefixed token first. Every client of the configured Iggy server +/// goes through this, so all of them authenticate identically. +pub(crate) fn connection_string(config: &IggyConfig) -> Result { let token = if config.token.is_empty() { None } else { Some(resolve_token(&config.token)?) }; - - let consumer = create_client(&config, token.as_deref()).await?; - let producer = create_client(&config, token.as_deref()).await?; - let iggy_clients = IggyClients { producer, consumer }; - Ok(iggy_clients) + connection_string_with_token(config, token.as_deref()) } -async fn create_client( +fn connection_string_with_token( config: &IggyConfig, token: Option<&str>, -) -> Result { +) -> Result { let address = config.address.to_owned(); let username = config.username.to_owned(); let password = config.password.to_owned(); @@ -114,7 +120,7 @@ async fn create_client( format!("iggy://{username}:{password}@{address}") }; - let connection_string = if config.tls.enabled { + if config.tls.enabled { let ca_file = &config.tls.ca_file; if ca_file.is_empty() { error!("TLS CA file must be provided when TLS is enabled."); @@ -127,11 +133,16 @@ async fn create_client( .filter(|domain| !domain.is_empty()) .map(|domain| format!("&tls_domain={domain}")) .unwrap_or_default(); - format!("{connection_string}?tls=true&tls_ca_file={ca_file}{domain}") + Ok(format!( + "{connection_string}?tls=true&tls_ca_file={ca_file}{domain}" + )) } else { - connection_string - }; + Ok(connection_string) + } +} +async fn create_client(config: &IggyConfig) -> Result { + let connection_string = connection_string(config)?; let client = IggyClientBuilder::from_connection_string(&connection_string)?.build()?; client.connect().await?; Ok(client) @@ -239,4 +250,73 @@ mod tests { assert!(result.is_err()); assert!(matches!(result, Err(RuntimeError::TokenFileEmpty(_)))); } + + #[test] + fn test_connection_string_with_username_and_password() { + let config = IggyConfig::default(); + let result = connection_string(&config).unwrap(); + assert_eq!( + result, + format!( + "iggy://{}:{}@{}", + config.username, config.password, config.address + ) + ); + } + + #[test] + fn test_connection_string_with_token() { + let config = IggyConfig { + token: "my-secret-token".to_owned(), + ..IggyConfig::default() + }; + let result = connection_string(&config).unwrap(); + assert_eq!(result, format!("iggy://my-secret-token@{}", config.address)); + } + + #[test] + fn test_connection_string_resolves_token_file() { + let mut temp_file = NamedTempFile::new().unwrap(); + writeln!(temp_file, "token-from-file").unwrap(); + let config = IggyConfig { + token: format!("file:{}", temp_file.path().display()), + ..IggyConfig::default() + }; + let result = connection_string(&config).unwrap(); + assert_eq!(result, format!("iggy://token-from-file@{}", config.address)); + } + + #[test] + fn test_connection_string_without_credentials_fails() { + let config = IggyConfig { + username: String::new(), + ..IggyConfig::default() + }; + let result = connection_string(&config); + assert!(matches!(result, Err(RuntimeError::MissingIggyCredentials))); + } + + #[test] + fn test_connection_string_with_tls_appends_parameters() { + let mut config = IggyConfig::default(); + config.tls.enabled = true; + config.tls.ca_file = "/certs/ca.pem".to_owned(); + config.tls.domain = Some("iggy.internal".to_owned()); + let result = connection_string(&config).unwrap(); + assert!( + result.ends_with("?tls=true&tls_ca_file=/certs/ca.pem&tls_domain=iggy.internal"), + "unexpected connection string: {result}" + ); + } + + #[test] + fn test_connection_string_with_tls_without_ca_file_fails() { + let mut config = IggyConfig::default(); + config.tls.enabled = true; + let result = connection_string(&config); + assert!(matches!( + result, + Err(RuntimeError::MissingTlsCertificateFile) + )); + } } diff --git a/core/connectors/sdk/src/lib.rs b/core/connectors/sdk/src/lib.rs index 294fae8b6f..7aeb02bc1a 100644 --- a/core/connectors/sdk/src/lib.rs +++ b/core/connectors/sdk/src/lib.rs @@ -455,4 +455,19 @@ pub enum Error { /// be duplicated. #[error("Catalog commit error: {0}")] CatalogCommitError(String), + /// The state store is temporarily unavailable (5xx, timeout, connect + /// failure) and bounded retries were exhausted. The operation may succeed + /// later; the batch-ack path Nacks and the plugin re-polls. + #[error("Transient state error: {0}")] + TransientState(String), + /// The state store rejected the operation in a way retrying cannot fix + /// (version conflict, revoked authorization, protocol violation). No + /// write from this process can be expected to succeed again. + #[error("Permanent state error: {0}")] + PermanentState(String), + /// A previous save failed permanently, so the state provider refuses + /// further saves without touching the network. Fail-fast marker, never + /// retried. + #[error("State provider latched after a permanent state error")] + StateLatched, } diff --git a/core/integration/tests/connectors/runtime/http_state.rs b/core/integration/tests/connectors/runtime/http_state.rs new file mode 100644 index 0000000000..bf9cbe07d2 --- /dev/null +++ b/core/integration/tests/connectors/runtime/http_state.rs @@ -0,0 +1,384 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Runtime-level tests for the HTTP state storage backend. +//! +//! Each test runs the real connectors runtime against an in-process wiremock +//! state server that enforces the protocol contract: conditional writes +//! (`If-None-Match: *` / `If-Match`), strong ETags per committed write, and +//! injectable failure modes (`412` conflicts, `503` bursts). The scenarios: +//! * state store unreachable at boot with an enabled source -> the runtime +//! process exits instead of minting an idle `FailedPlugin`, +//! * `404` at boot -> the source runs from its default state and its +//! checkpoints land on the stub, +//! * `412` mid-stream -> the provider latches, the checkpoint stops +//! advancing, and no further PUTs reach the server, +//! * `503` burst mid-stream -> saves Nack, then recover once the store is +//! healthy again, +//! * API-driven restart -> the source reloads the served state and resumes +//! the ETag chain. + +use assert_cmd::prelude::CommandCargoExt; +use async_trait::async_trait; +use iggy_connector_sdk::api::{ConnectorStatus, SourceInfoResponse}; +use integration::harness::config::TestServerConfig; +use integration::harness::{TestBinaryError, TestFixture, TestHarness, seeds}; +use integration::iggy_harness; +use reqwest::Client; +use std::collections::HashMap; +use std::process::{Command, Stdio}; +use std::sync::Arc; +use std::sync::Mutex as StdMutex; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::time::{Duration, Instant}; +use tokio::time::sleep; +use wiremock::matchers::path; +use wiremock::{Mock, MockServer, Request, Respond, ResponseTemplate}; + +const SOURCE_KEY: &str = "random_http_state"; +const RESOURCE_PATH: &str = "/source_random_http_state"; +const RUNTIME_CONFIG_PATH: &str = "tests/connectors/runtime/http_state.toml"; +const WAIT_DEADLINE: Duration = Duration::from_secs(15); +const POLL_INTERVAL: Duration = Duration::from_millis(50); + +/// In-memory state server backing the wiremock responder. Enforces the +/// conditional-write contract and exposes counters plus injectable failure +/// modes for the tests. +#[derive(Default)] +struct SharedStore { + version: AtomicU64, + body: StdMutex>, + conflict_mode: AtomicBool, + fail_next_puts: AtomicU64, + get_count: AtomicU64, + put_count: AtomicU64, +} + +impl SharedStore { + fn etag(version: u64) -> String { + format!("\"v{version}\"") + } +} + +struct StateStoreResponder(Arc); + +impl Respond for StateStoreResponder { + fn respond(&self, request: &Request) -> ResponseTemplate { + let store = &self.0; + match request.method.as_str() { + "GET" => { + store.get_count.fetch_add(1, Ordering::SeqCst); + let version = store.version.load(Ordering::SeqCst); + if version == 0 { + return ResponseTemplate::new(404); + } + let body = store.body.lock().expect("store lock").clone(); + ResponseTemplate::new(200) + .insert_header("etag", SharedStore::etag(version).as_str()) + .set_body_bytes(body) + } + "PUT" => { + store.put_count.fetch_add(1, Ordering::SeqCst); + if store + .fail_next_puts + .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |remaining| { + remaining.checked_sub(1) + }) + .is_ok() + { + return ResponseTemplate::new(503); + } + if store.conflict_mode.load(Ordering::SeqCst) { + return ResponseTemplate::new(412); + } + let version = store.version.load(Ordering::SeqCst); + let condition_ok = if version == 0 { + request + .headers + .get("if-none-match") + .map(|value| value.as_bytes() == b"*") + .unwrap_or(false) + } else { + request + .headers + .get("if-match") + .and_then(|value| value.to_str().ok()) + .map(|value| value == SharedStore::etag(version)) + .unwrap_or(false) + }; + if !condition_ok { + return ResponseTemplate::new(412); + } + let next = version + 1; + *store.body.lock().expect("store lock") = request.body.clone(); + store.version.store(next, Ordering::SeqCst); + ResponseTemplate::new(200).insert_header("etag", SharedStore::etag(next).as_str()) + } + _ => ResponseTemplate::new(405), + } + } +} + +pub struct HttpStateStoreFixture { + server: MockServer, + store: Arc, +} + +#[async_trait] +impl TestFixture for HttpStateStoreFixture { + async fn setup() -> Result { + let server = MockServer::start().await; + let store = Arc::new(SharedStore::default()); + Mock::given(path(RESOURCE_PATH)) + .respond_with(StateStoreResponder(store.clone())) + .mount(&server) + .await; + Ok(Self { server, store }) + } + + fn connectors_runtime_envs(&self) -> HashMap { + HashMap::from([( + "IGGY_CONNECTORS_STATE_HTTP_URL".to_string(), + self.server.uri(), + )]) + } +} + +async fn wait_until bool>(what: &str, condition: F) { + let deadline = Instant::now() + WAIT_DEADLINE; + while !condition() { + if Instant::now() >= deadline { + panic!("timed out after {WAIT_DEADLINE:?} waiting for {what}"); + } + sleep(POLL_INTERVAL).await; + } +} + +async fn fetch_source(harness: &TestHarness) -> SourceInfoResponse { + let api_url = harness + .connectors_runtime() + .expect("connectors runtime handle should be available") + .http_url(); + let sources: Vec = Client::new() + .get(format!("{api_url}/sources")) + .send() + .await + .expect("Failed to query /sources") + .json() + .await + .expect("Failed to parse sources"); + sources + .into_iter() + .find(|source| source.key == SOURCE_KEY) + .expect("HTTP-state source should be reported") +} + +async fn wait_for_status(harness: &TestHarness, expected: ConnectorStatus) { + let deadline = Instant::now() + WAIT_DEADLINE; + loop { + let source = fetch_source(harness).await; + if source.status == expected { + return; + } + if Instant::now() >= deadline { + panic!( + "timed out after {WAIT_DEADLINE:?} waiting for source status {expected:?}, last seen {:?} ({:?})", + source.status, source.last_error + ); + } + sleep(POLL_INTERVAL).await; + } +} + +#[tokio::test] +async fn given_unavailable_state_store_when_booting_should_fail_startup() { + let mut harness = TestHarness::builder() + .server(TestServerConfig::default()) + .build() + .expect("harness should build"); + harness + .start_with_seed(|client| async move { seeds::connector_stream(&client).await }) + .await + .expect("iggy-server should start"); + let iggy_address = harness + .server() + .tcp_addr() + .expect("server TCP address should be known"); + + // Bound but never accepted, so every state request times out instead of + // racing other tests for a recycled port. + let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("reserve a port"); + let state_url = format!("http://127.0.0.1:{}", listener.local_addr().unwrap().port()); + + let mut command = Command::cargo_bin("iggy-connectors").expect("iggy-connectors binary"); + command + .env("IGGY_CONNECTORS_CONFIG_PATH", RUNTIME_CONFIG_PATH) + .env("IGGY_CONNECTORS_IGGY_ADDRESS", iggy_address.to_string()) + .env("IGGY_CONNECTORS_HTTP_ADDRESS", "127.0.0.1:0") + .env("IGGY_CONNECTORS_STATE_HTTP_URL", state_url) + .env("IGGY_CONNECTORS_STATE_HTTP_TIMEOUT", "200ms") + .stdin(Stdio::null()); + + let output = tokio::time::timeout( + Duration::from_secs(60), + tokio::task::spawn_blocking(move || command.output()), + ) + .await + .expect("the runtime must exit instead of running with an unavailable state store") + .expect("join spawned command") + .expect("spawn iggy-connectors"); + + assert!( + !output.status.success(), + "boot must fail when the state store is unreachable for an enabled source" + ); + let logs = format!( + "{}\n{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + assert!( + logs.contains("failed to load state") || logs.contains("StateLoadFailed"), + "startup failure should point at the state load, got:\n{logs}" + ); +} + +#[iggy_harness( + server(connectors_runtime(config_path = "tests/connectors/runtime/http_state.toml")), + seed = seeds::connector_stream +)] +async fn given_empty_state_store_when_booted_should_run_and_checkpoint( + harness: &TestHarness, + fixture: HttpStateStoreFixture, +) { + crate::connectors::random_source_liveness::assert_produces_messages(harness).await; + wait_for_status(harness, ConnectorStatus::Running).await; + wait_until("the first checkpoint PUT to reach the state server", || { + fixture.store.version.load(Ordering::SeqCst) > 0 + }) + .await; + assert!( + !fixture.store.body.lock().expect("store lock").is_empty(), + "the stored checkpoint must carry the serialized source state" + ); +} + +#[iggy_harness( + server(connectors_runtime(config_path = "tests/connectors/runtime/http_state.toml")), + seed = seeds::connector_stream +)] +async fn given_conflict_mid_stream_should_nack_and_latch( + harness: &TestHarness, + fixture: HttpStateStoreFixture, +) { + wait_until("the first committed checkpoint", || { + fixture.store.version.load(Ordering::SeqCst) > 0 + }) + .await; + + fixture.store.conflict_mode.store(true, Ordering::SeqCst); + wait_for_status(harness, ConnectorStatus::Error).await; + + let version_after_conflict = fixture.store.version.load(Ordering::SeqCst); + let puts_after_conflict = fixture.store.put_count.load(Ordering::SeqCst); + sleep(Duration::from_millis(500)).await; + assert_eq!( + fixture.store.put_count.load(Ordering::SeqCst), + puts_after_conflict, + "a latched provider must not send further PUTs" + ); + assert_eq!( + fixture.store.version.load(Ordering::SeqCst), + version_after_conflict, + "the checkpoint must not advance after a 412" + ); +} + +#[iggy_harness( + server(connectors_runtime(config_path = "tests/connectors/runtime/http_state.toml")), + seed = seeds::connector_stream +)] +async fn given_unavailable_burst_mid_stream_should_nack_then_recover( + harness: &TestHarness, + fixture: HttpStateStoreFixture, +) { + wait_until("the first committed checkpoint", || { + fixture.store.version.load(Ordering::SeqCst) > 0 + }) + .await; + + // 6 failed PUTs = 3 failed saves at max_attempts = 1, i.e. 3 Nacks - + // safely below the plugin's 5-consecutive-Nack cutoff. + let version_before_burst = fixture.store.version.load(Ordering::SeqCst); + fixture.store.fail_next_puts.store(6, Ordering::SeqCst); + wait_until("checkpoints to resume after the 503 burst", || { + fixture.store.version.load(Ordering::SeqCst) >= version_before_burst + 2 + }) + .await; + let _ = harness; +} + +#[iggy_harness( + server(connectors_runtime(config_path = "tests/connectors/runtime/http_state.toml")), + seed = seeds::connector_stream +)] +async fn given_restart_when_state_exists_should_resume_from_served_state( + harness: &TestHarness, + fixture: HttpStateStoreFixture, +) { + wait_until("the first committed checkpoint", || { + fixture.store.version.load(Ordering::SeqCst) > 0 + }) + .await; + + let gets_before_restart = fixture.store.get_count.load(Ordering::SeqCst); + let version_before_restart = fixture.store.version.load(Ordering::SeqCst); + + let api_url = harness + .connectors_runtime() + .expect("connectors runtime handle should be available") + .http_url(); + let response = Client::new() + .post(format!("{api_url}/sources/{SOURCE_KEY}/restart")) + .send() + .await + .expect("restart request should be sent"); + assert!( + response.status().is_success(), + "restart should succeed, got {}", + response.status() + ); + + wait_until("the restarted source to load state from the server", || { + fixture.store.get_count.load(Ordering::SeqCst) > gets_before_restart + }) + .await; + wait_until("the restarted source to resume the ETag chain", || { + fixture.store.version.load(Ordering::SeqCst) > version_before_restart + }) + .await; + + let runtime = harness + .connectors_runtime() + .expect("connectors runtime handle should be available"); + let (stdout, stderr) = runtime.collect_logs(); + let logs = format!("{stdout}\n{stderr}"); + assert!( + logs.contains("Restored state for Random source"), + "the plugin should restore the served state on restart" + ); +} diff --git a/core/integration/tests/connectors/runtime/http_state.toml b/core/integration/tests/connectors/runtime/http_state.toml new file mode 100644 index 0000000000..3345453fb4 --- /dev/null +++ b/core/integration/tests/connectors/runtime/http_state.toml @@ -0,0 +1,34 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +[connectors] +config_type = "local" +config_dir = "tests/connectors/runtime/http_state_config" + +# The state server URL is injected per test via IGGY_CONNECTORS_STATE_HTTP_URL. +[state] +storage = "http" + +[state.http] +timeout = "2s" + +[state.http.retry] +enabled = true +max_attempts = 1 +initial_backoff = "10ms" +max_backoff = "50ms" +backoff_multiplier = 2 diff --git a/core/integration/tests/connectors/runtime/http_state_config/random_source.toml b/core/integration/tests/connectors/runtime/http_state_config/random_source.toml new file mode 100644 index 0000000000..66558c7fea --- /dev/null +++ b/core/integration/tests/connectors/runtime/http_state_config/random_source.toml @@ -0,0 +1,37 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +type = "source" +key = "random_http_state" +enabled = true +version = 0 +name = "Random source with HTTP state" +path = "../../target/debug/libiggy_connector_random_source" +plugin_config_format = "json" + +[[streams]] +stream = "test_stream" +topic = "test_topic" +schema = "json" +batch_length = 100 +linger_time = "5ms" + +[plugin_config] +interval = "100ms" +max_count = 1_000_000 +messages_range = [1, 3] +payload_size = 50 diff --git a/core/integration/tests/connectors/runtime/mod.rs b/core/integration/tests/connectors/runtime/mod.rs index 86294e8daa..72f3ed271f 100644 --- a/core/integration/tests/connectors/runtime/mod.rs +++ b/core/integration/tests/connectors/runtime/mod.rs @@ -17,3 +17,4 @@ mod benchmark; mod error_isolation; +mod http_state;