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;