diff --git a/docs/sinks/redis-streams.md b/docs/sinks/redis-streams.md index ca2bde5..8c61955 100644 --- a/docs/sinks/redis-streams.md +++ b/docs/sinks/redis-streams.md @@ -15,6 +15,10 @@ sink: type: redis-streams url: redis://localhost:6379 stream_name: events + connection_timeout_ms: 1000 + response_timeout_ms: 5000 + connection_retries: 2 + connection_max_delay_ms: 1000 ``` ### With Length Limit @@ -34,6 +38,10 @@ sink: | `url` | string | Yes | - | No | Redis connection URL | | `stream_name` | string | No | - | Yes | Default stream (can be overridden per-event) | | `max_len` | integer | No | - | No | Maximum stream length (uses MAXLEN ~) | +| `connection_timeout_ms` | integer | No | Redis default | No | Timeout for each connection attempt | +| `response_timeout_ms` | integer | No | Redis default | No | Timeout for command responses | +| `connection_retries` | integer | No | Redis default | No | Number of reconnection attempts | +| `connection_max_delay_ms` | integer | No | Redis default | No | Maximum delay between reconnect attempts | ## Dynamic Routing diff --git a/docs/sinks/redis-strings.md b/docs/sinks/redis-strings.md index 58ca694..56183d6 100644 --- a/docs/sinks/redis-strings.md +++ b/docs/sinks/redis-strings.md @@ -14,6 +14,10 @@ docker pull ghcr.io/psteinroe/postgres-stream:redis-strings-latest sink: type: redis-strings url: redis://localhost:6379 + connection_timeout_ms: 1000 + response_timeout_ms: 5000 + connection_retries: 2 + connection_max_delay_ms: 1000 ``` ### With Key Prefix @@ -31,6 +35,10 @@ sink: |--------|------|----------|---------|-------------------|-------------| | `url` | string | Yes | - | No | Redis connection URL | | `key_prefix` | string | No | - | No | Prefix for all keys | +| `connection_timeout_ms` | integer | No | Redis default | No | Timeout for each connection attempt | +| `response_timeout_ms` | integer | No | Redis default | No | Timeout for command responses | +| `connection_retries` | integer | No | Redis default | No | Number of reconnection attempts | +| `connection_max_delay_ms` | integer | No | Redis default | No | Maximum delay between reconnect attempts | | `key` | - | - | - | Yes | Full key (via metadata only) | ## Key Resolution diff --git a/src/config/load.rs b/src/config/load.rs index 9327ee3..951228d 100644 --- a/src/config/load.rs +++ b/src/config/load.rs @@ -74,9 +74,9 @@ impl TryFrom for Environment { "prod" => Ok(Self::Prod), "staging" => Ok(Self::Staging), "dev" => Ok(Self::Dev), - other => Err(io::Error::other(format!( - "{other} is not a supported environment. Use either `prod`/`staging`/`dev`.", - ))), + _ => Err(io::Error::other( + "unsupported APP_ENVIRONMENT value; use `prod`, `staging`, or `dev`", + )), } } } diff --git a/src/config/sink.rs b/src/config/sink.rs index b8f00d8..c79effd 100644 --- a/src/config/sink.rs +++ b/src/config/sink.rs @@ -105,3 +105,48 @@ pub enum SinkConfig { #[serde(rename = "gcp-pubsub")] GcpPubsub(GcpPubsubSinkConfig), } + +impl SinkConfig { + /// Returns the stable, non-sensitive name of the configured sink. + pub(crate) const fn kind(&self) -> &'static str { + match self { + Self::Memory => "memory", + + #[cfg(feature = "sink-elasticsearch")] + Self::Elasticsearch(_) => "elasticsearch", + + #[cfg(feature = "sink-redis-strings")] + Self::RedisStrings(_) => "redis-strings", + + #[cfg(feature = "sink-redis-streams")] + Self::RedisStreams(_) => "redis-streams", + + #[cfg(feature = "sink-nats")] + Self::Nats(_) => "nats", + + #[cfg(feature = "sink-rabbitmq")] + Self::Rabbitmq(_) => "rabbitmq", + + #[cfg(feature = "sink-webhook")] + Self::Webhook(_) => "webhook", + + #[cfg(feature = "sink-kafka")] + Self::Kafka(_) => "kafka", + + #[cfg(feature = "sink-sqs")] + Self::Sqs(_) => "sqs", + + #[cfg(feature = "sink-sns")] + Self::Sns(_) => "sns", + + #[cfg(feature = "sink-kinesis")] + Self::Kinesis(_) => "kinesis", + + #[cfg(feature = "sink-meilisearch")] + Self::Meilisearch(_) => "meilisearch", + + #[cfg(feature = "sink-gcp-pubsub")] + Self::GcpPubsub(_) => "gcp-pubsub", + } + } +} diff --git a/src/core.rs b/src/core.rs index a2088b4..948210e 100644 --- a/src/core.rs +++ b/src/core.rs @@ -17,7 +17,7 @@ use etl::pipeline::Pipeline; use etl::store::both::postgres::PostgresStore; use sqlx::postgres::PgPoolOptions; use tokio::signal::unix::{SignalKind, signal}; -use tracing::{debug, error, info, warn}; +use tracing::{error, info, warn}; /// Starts the pipeline daemon with the provided configuration. /// @@ -28,8 +28,6 @@ use tracing::{debug, error, info, warn}; /// If a replication slot is invalidated, this function will automatically /// recover by setting a failover checkpoint and restarting the pipeline. pub async fn start_pipeline_with_config(config: PipelineConfig) -> EtlResult<()> { - info!("starting pgstream daemon"); - log_config(&config); // Run etl migrations before starting the pipeline @@ -98,27 +96,15 @@ async fn run_pipeline(config: &PipelineConfig) -> EtlResult<()> { #[cfg(feature = "sink-redis-strings")] SinkConfig::RedisStrings(cfg) => { use crate::sink::redis_strings::RedisStringsSink; - let s = RedisStringsSink::new(cfg.clone()).await.map_err(|e| { - etl::etl_error!( - etl::error::ErrorKind::InvalidData, - "Failed to create Redis Strings sink", - e.to_string() - ) - })?; - AnySink::RedisStrings(s) + let sink = RedisStringsSink::new(cfg.clone()).await?; + AnySink::RedisStrings(sink) } #[cfg(feature = "sink-redis-streams")] SinkConfig::RedisStreams(cfg) => { use crate::sink::redis_streams::RedisStreamsSink; - let s = RedisStreamsSink::new(cfg.clone()).await.map_err(|e| { - etl::etl_error!( - etl::error::ErrorKind::InvalidData, - "Failed to create Redis Streams sink", - e.to_string() - ) - })?; - AnySink::RedisStreams(s) + let sink = RedisStreamsSink::new(cfg.clone()).await?; + AnySink::RedisStreams(sink) } #[cfg(feature = "sink-nats")] @@ -256,95 +242,19 @@ async fn run_pipeline(config: &PipelineConfig) -> EtlResult<()> { start_pipeline_with_shutdown(pipeline).await } -/// Logs the daemon configuration (without secrets). +/// Logs an allowlist of safe operational configuration fields. fn log_config(config: &PipelineConfig) { - log_stream_config(config); - log_sink_config(&config.sink); -} - -fn log_stream_config(config: &PipelineConfig) { let stream = &config.stream; - debug!( + info!( stream_id = stream.id, - host = stream.pg_connection.host, - port = stream.pg_connection.port, - dbname = stream.pg_connection.name, - username = stream.pg_connection.username, - tls_enabled = stream.pg_connection.tls.enabled, + sink_type = config.sink.kind(), max_batch_size = stream.batch.max_size, max_batch_fill_ms = stream.batch.max_fill_ms, - "stream configuration" + tls_enabled = stream.pg_connection.tls.enabled, + "pgstream daemon starting" ); } -fn log_sink_config(config: &SinkConfig) { - match config { - SinkConfig::Memory => { - debug!("using memory sink"); - } - - #[cfg(feature = "sink-elasticsearch")] - SinkConfig::Elasticsearch(_cfg) => { - debug!("using elasticsearch sink"); - } - - #[cfg(feature = "sink-redis-strings")] - SinkConfig::RedisStrings(_cfg) => { - debug!("using redis-strings sink"); - } - - #[cfg(feature = "sink-redis-streams")] - SinkConfig::RedisStreams(_cfg) => { - debug!("using redis-streams sink"); - } - - #[cfg(feature = "sink-nats")] - SinkConfig::Nats(_cfg) => { - debug!("using nats sink"); - } - - #[cfg(feature = "sink-rabbitmq")] - SinkConfig::Rabbitmq(_cfg) => { - debug!("using rabbitmq sink"); - } - - #[cfg(feature = "sink-webhook")] - SinkConfig::Webhook(_cfg) => { - debug!("using webhook sink"); - } - - #[cfg(feature = "sink-kafka")] - SinkConfig::Kafka(_cfg) => { - debug!("using kafka sink"); - } - - #[cfg(feature = "sink-sqs")] - SinkConfig::Sqs(_cfg) => { - debug!("using sqs sink"); - } - - #[cfg(feature = "sink-sns")] - SinkConfig::Sns(_cfg) => { - debug!("using sns sink"); - } - - #[cfg(feature = "sink-kinesis")] - SinkConfig::Kinesis(_cfg) => { - debug!("using kinesis sink"); - } - - #[cfg(feature = "sink-meilisearch")] - SinkConfig::Meilisearch(_cfg) => { - debug!("using meilisearch sink"); - } - - #[cfg(feature = "sink-gcp-pubsub")] - SinkConfig::GcpPubsub(_cfg) => { - debug!("using gcp-pubsub sink"); - } - } -} - /// Starts a pipeline and handles graceful shutdown signals. /// /// Launches the pipeline, sets up signal handlers for SIGTERM and SIGINT, diff --git a/src/error.rs b/src/error.rs new file mode 100644 index 0000000..382d6fa --- /dev/null +++ b/src/error.rs @@ -0,0 +1,116 @@ +use etl::error::EtlError; +use std::{backtrace::Backtrace, error::Error, fmt}; + +pub type PgStreamResult = Result; + +/// Captured backtrace wrapper matching etl-replicator's stable error-reporting pattern. +pub struct CapturedBacktrace(Backtrace); + +impl CapturedBacktrace { + fn capture() -> Self { + Self(Backtrace::capture()) + } +} + +impl fmt::Debug for CapturedBacktrace { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(formatter, "{}", self.0) + } +} + +/// Top-level daemon error used to render one consistent failure report. +#[derive(Debug)] +pub enum PgStreamError { + Etl(EtlError), + Config(Box, CapturedBacktrace), + Io(std::io::Error, CapturedBacktrace), +} + +impl PgStreamError { + pub fn config(error: E) -> Self { + Self::Config(Box::new(error), CapturedBacktrace::capture()) + } + + fn category(&self) -> &'static str { + match self { + Self::Etl(_) => "daemon error", + Self::Config(_, _) => "configuration error", + Self::Io(_, _) => "i/o error", + } + } + + fn backtrace(&self) -> Option<&Backtrace> { + match self { + Self::Etl(error) => error.backtrace(), + Self::Config(_, backtrace) | Self::Io(_, backtrace) => Some(&backtrace.0), + } + } + + pub fn render_report(&self) -> String { + let mut report = String::new(); + report.push_str("postgres-stream failed\n"); + report.push_str(&format!("category: {}\n", self.category())); + report.push_str(&format!("error: {self}\n")); + + if !matches!(self, Self::Etl(error) if error.errors().is_some()) { + let mut source = Error::source(self); + let mut index = 1usize; + while let Some(error) = source { + report.push_str(&format!("cause {index}: {error}\n")); + source = error.source(); + index += 1; + } + } + + if should_render_backtrace() + && let Some(backtrace) = self.backtrace() + { + report.push_str("backtrace:\n"); + report.push_str(&backtrace.to_string()); + if !report.ends_with('\n') { + report.push('\n'); + } + } + + report + } +} + +impl fmt::Display for PgStreamError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Etl(error) => write!(formatter, "{error}"), + Self::Config(source, _) => write!(formatter, "configuration error: {source}"), + Self::Io(source, _) => write!(formatter, "i/o error: {source}"), + } + } +} + +impl Error for PgStreamError { + fn source(&self) -> Option<&(dyn Error + 'static)> { + match self { + Self::Etl(error) => error.source(), + Self::Config(source, _) => Some(source.as_ref()), + Self::Io(source, _) => Some(source), + } + } +} + +impl From for PgStreamError { + fn from(error: EtlError) -> Self { + Self::Etl(error) + } +} + +impl From for PgStreamError { + fn from(error: std::io::Error) -> Self { + Self::Io(error, CapturedBacktrace::capture()) + } +} + +fn should_render_backtrace() -> bool { + matches!( + std::env::var("RUST_BACKTRACE").as_deref(), + Ok("1") | Ok("full") + ) +} diff --git a/src/main.rs b/src/main.rs index 571a982..d112f8d 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,10 +1,12 @@ -use etl::error::EtlResult; +use crate::error::{PgStreamError, PgStreamResult}; use postgres_stream::config::{PipelineConfig, load_config}; use postgres_stream::core::start_pipeline_with_config; use postgres_stream::metrics::init_metrics; -use tracing::{error, info}; +use std::process::ExitCode; use tracing_subscriber::{EnvFilter, fmt}; +mod error; + /// Jemalloc allocator for better memory management in high-throughput async workloads. #[cfg(not(target_env = "msvc"))] #[global_allocator] @@ -40,25 +42,23 @@ pub static malloc_conf: &[u8] = /// Loads configuration, initializes tracing, starts the async runtime, /// and launches the replication stream. Handles all errors and ensures /// proper service initialization sequence. -fn main() -> anyhow::Result<()> { - // Initialize tracing subscriber for logging - init_tracing(); - - // Load daemon configuration - let config = load_config::().map_err(|c| { - etl::etl_error!( - etl::error::ErrorKind::ConfigError, - "failed to load configuration: {}", - c - ) - })?; +fn main() -> ExitCode { + match try_main() { + Ok(()) => ExitCode::SUCCESS, + Err(error) => { + eprintln!("{}", error.render_report()); + ExitCode::FAILURE + } + } +} - // Initialize metrics collection - init_metrics()?; +/// Runs the daemon and propagates typed startup errors. +fn try_main() -> PgStreamResult<()> { + init_tracing(); - info!(stream_id = config.stream.id, "pgstream daemon starting"); + let config = load_pipeline_config()?; + init_metrics().map_err(PgStreamError::config)?; - // Start the tokio runtime tokio::runtime::Builder::new_multi_thread() .enable_all() .build()? @@ -67,20 +67,20 @@ fn main() -> anyhow::Result<()> { Ok(()) } +fn load_pipeline_config() -> PgStreamResult { + load_config::().map_err(PgStreamError::config) +} + /// Main async entry point that starts the pipeline. /// /// Launches the stream with the provided configuration and captures /// any errors for logging and error handling. -async fn async_main(config: PipelineConfig) -> EtlResult<()> { +async fn async_main(config: PipelineConfig) -> PgStreamResult<()> { // Start the jemalloc metrics collection background task. #[cfg(not(target_env = "msvc"))] postgres_stream::metrics::spawn_jemalloc_metrics_task(config.stream.id); - // Start the daemon with error handling - if let Err(err) = start_pipeline_with_config(config).await { - error!("an error occurred in the stream daemon: {err}"); - return Err(err); - } + start_pipeline_with_config(config).await?; Ok(()) } @@ -100,3 +100,30 @@ fn init_tracing() { .with_line_number(false) .init(); } + +#[cfg(test)] +mod tests { + use super::*; + use std::fs; + use temp_env::with_vars; + use tempfile::TempDir; + + #[test] + fn malformed_configuration_renders_the_underlying_cause() { + let temp_dir = TempDir::new().unwrap(); + fs::write(temp_dir.path().join("base.yml"), "stream: [\n").unwrap(); + + let report = with_vars( + [ + ("APP_CONFIG_DIR", temp_dir.path().to_str()), + ("APP_ENVIRONMENT", Some("prod")), + ], + || load_pipeline_config().unwrap_err().render_report(), + ); + + assert!(report.contains("category: configuration error")); + assert!(report.contains("failed to initialize configuration builder")); + assert!(report.contains("cause 2:")); + assert!(report.contains("line 2 column 1")); + } +} diff --git a/src/sink/mod.rs b/src/sink/mod.rs index 93f4815..a0b2411 100644 --- a/src/sink/mod.rs +++ b/src/sink/mod.rs @@ -7,6 +7,9 @@ pub mod elasticsearch; #[cfg(feature = "sink-nats")] pub mod nats; +#[cfg(any(feature = "sink-redis-strings", feature = "sink-redis-streams"))] +mod redis_common; + #[cfg(feature = "sink-redis-strings")] pub mod redis_strings; diff --git a/src/sink/redis_common.rs b/src/sink/redis_common.rs new file mode 100644 index 0000000..78a372c --- /dev/null +++ b/src/sink/redis_common.rs @@ -0,0 +1,184 @@ +//! Shared Redis connection and error handling for Redis-backed sinks. + +use etl::error::{ErrorKind, EtlError}; +use redis::{ + RedisError, + aio::{ConnectionManager, ConnectionManagerConfig}, +}; +use std::time::Duration; + +/// Optional Redis connection-manager settings. +#[derive(Clone, Copy, Debug, Default)] +pub(crate) struct RedisConnectionSettings { + pub(crate) connection_timeout_ms: Option, + pub(crate) response_timeout_ms: Option, + pub(crate) connection_retries: Option, + pub(crate) connection_max_delay_ms: Option, +} + +/// Creates a Redis connection manager with the configured timeout and retry behavior. +pub(crate) async fn connect( + url: String, + settings: RedisConnectionSettings, +) -> redis::RedisResult { + let client = redis::Client::open(url)?; + let mut config = ConnectionManagerConfig::new(); + + if let Some(timeout_ms) = settings.connection_timeout_ms { + config = config.set_connection_timeout(Duration::from_millis(timeout_ms)); + } + if let Some(timeout_ms) = settings.response_timeout_ms { + config = config.set_response_timeout(Duration::from_millis(timeout_ms)); + } + if let Some(retries) = settings.connection_retries { + config = config.set_number_of_retries(retries); + } + if let Some(max_delay_ms) = settings.connection_max_delay_ms { + config = config.set_max_delay(max_delay_ms); + } + + ConnectionManager::new_with_config(client, config).await +} + +/// Maps a Redis failure to the ETL retry classification while retaining its source. +pub(crate) fn map_error(error: RedisError, description: &'static str) -> EtlError { + let kind = classify_error(&error); + etl::etl_error!(kind, description, source: error) +} + +fn classify_error(error: &RedisError) -> ErrorKind { + if error.kind() == redis::ErrorKind::AuthenticationFailed { + return ErrorKind::DestinationAuthenticationError; + } + + if is_transient(error) { + ErrorKind::DestinationConnectionFailed + } else { + ErrorKind::InvalidData + } +} + +fn is_transient(error: &RedisError) -> bool { + if error.is_io_error() { + return true; + } + + if error.kind() == redis::ErrorKind::ExtensionError { + return error.code() == Some("OOM"); + } + + matches!( + error.kind(), + redis::ErrorKind::BusyLoadingError + | redis::ErrorKind::Moved + | redis::ErrorKind::Ask + | redis::ErrorKind::TryAgain + | redis::ErrorKind::ClusterDown + | redis::ErrorKind::MasterDown + | redis::ErrorKind::ReadOnly + | redis::ErrorKind::ClusterConnectionNotFound + ) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::{error::Error, io, net::TcpListener, thread, time::Instant}; + + #[test] + fn classifies_transient_io_errors_for_timed_retry() { + let error = RedisError::from(io::Error::new(io::ErrorKind::TimedOut, "timed out")); + + assert_eq!( + classify_error(&error), + ErrorKind::DestinationConnectionFailed + ); + } + + #[test] + fn classifies_authentication_errors_for_manual_intervention() { + let error = RedisError::from(( + redis::ErrorKind::AuthenticationFailed, + "authentication failed", + )); + + assert_eq!( + classify_error(&error), + ErrorKind::DestinationAuthenticationError + ); + } + + #[test] + fn classifies_client_errors_as_invalid_data() { + for kind in [ + redis::ErrorKind::InvalidClientConfig, + redis::ErrorKind::ParseError, + redis::ErrorKind::ResponseError, + ] { + let error = RedisError::from((kind, "permanent error")); + assert_eq!(classify_error(&error), ErrorKind::InvalidData); + } + } + + #[test] + fn classifies_temporary_server_errors_for_timed_retry() { + let error = RedisError::from((redis::ErrorKind::TryAgain, "try again")); + + assert_eq!( + classify_error(&error), + ErrorKind::DestinationConnectionFailed + ); + } + + #[test] + fn classifies_redis_oom_as_transient() { + let error = redis::parse_redis_value(b"-OOM command not allowed\r\n") + .unwrap() + .extract_error() + .unwrap_err(); + + assert_eq!( + classify_error(&error), + ErrorKind::DestinationConnectionFailed + ); + } + + #[tokio::test] + async fn response_timeout_bounds_stalled_redis_responses() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let address = listener.local_addr().unwrap(); + let server = thread::spawn(move || { + let _connection = listener.accept().unwrap(); + thread::sleep(Duration::from_secs(1)); + }); + + let settings = RedisConnectionSettings { + response_timeout_ms: Some(50), + connection_retries: Some(0), + ..RedisConnectionSettings::default() + }; + let started = Instant::now(); + let error = match connect(format!("redis://{address}"), settings).await { + Ok(_) => panic!("stalled Redis response should time out"), + Err(error) => error, + }; + + assert!(error.is_timeout()); + assert!(started.elapsed() < Duration::from_millis(500)); + assert_eq!( + classify_error(&error), + ErrorKind::DestinationConnectionFailed + ); + + server.join().unwrap(); + } + + #[test] + fn mapped_errors_retain_the_redis_error_as_their_source() { + let error = RedisError::from((redis::ErrorKind::TypeError, "wrong type")); + let mapped = map_error(error, "failed to publish to Redis"); + + assert_eq!(mapped.kind(), ErrorKind::InvalidData); + assert!(mapped.source().is_some()); + } +} diff --git a/src/sink/redis_streams.rs b/src/sink/redis_streams.rs index a42f88d..f091546 100644 --- a/src/sink/redis_streams.rs +++ b/src/sink/redis_streams.rs @@ -31,7 +31,7 @@ use serde::{Deserialize, Serialize}; use std::sync::Arc; use tokio::sync::Mutex; -use crate::sink::Sink; +use crate::sink::{Sink, redis_common}; use crate::types::TriggeredEvent; /// Configuration for the Redis Streams sink. @@ -51,6 +51,22 @@ pub struct RedisStreamsSinkConfig { /// Maximum stream length (optional). Uses MAXLEN ~ for approximate trimming. #[serde(default)] pub max_len: Option, + + /// Timeout for establishing a Redis connection, in milliseconds. + #[serde(default)] + pub connection_timeout_ms: Option, + + /// Timeout for Redis command responses, in milliseconds. + #[serde(default)] + pub response_timeout_ms: Option, + + /// Number of reconnection attempts made by the connection manager. + #[serde(default)] + pub connection_retries: Option, + + /// Maximum delay between reconnection attempts, in milliseconds. + #[serde(default)] + pub connection_max_delay_ms: Option, } /// Configuration for the Redis Streams sink without sensitive data. @@ -63,6 +79,18 @@ pub struct RedisStreamsSinkConfigWithoutSecrets { /// Maximum stream length (optional). pub max_len: Option, + + /// Timeout for establishing a Redis connection, in milliseconds. + pub connection_timeout_ms: Option, + + /// Timeout for Redis command responses, in milliseconds. + pub response_timeout_ms: Option, + + /// Number of reconnection attempts made by the connection manager. + pub connection_retries: Option, + + /// Maximum delay between reconnection attempts, in milliseconds. + pub connection_max_delay_ms: Option, } impl From for RedisStreamsSinkConfigWithoutSecrets { @@ -70,6 +98,10 @@ impl From for RedisStreamsSinkConfigWithoutSecrets { Self { stream_name: config.stream_name, max_len: config.max_len, + connection_timeout_ms: config.connection_timeout_ms, + response_timeout_ms: config.response_timeout_ms, + connection_retries: config.connection_retries, + connection_max_delay_ms: config.connection_max_delay_ms, } } } @@ -79,6 +111,10 @@ impl From<&RedisStreamsSinkConfig> for RedisStreamsSinkConfigWithoutSecrets { Self { stream_name: config.stream_name.clone(), max_len: config.max_len, + connection_timeout_ms: config.connection_timeout_ms, + response_timeout_ms: config.response_timeout_ms, + connection_retries: config.connection_retries, + connection_max_delay_ms: config.connection_max_delay_ms, } } } @@ -105,11 +141,18 @@ impl RedisStreamsSink { /// # Errors /// /// Returns an error if the Redis connection cannot be established. - pub async fn new( - config: RedisStreamsSinkConfig, - ) -> Result> { - let client = redis::Client::open(config.url)?; - let connection = ConnectionManager::new(client).await?; + pub async fn new(config: RedisStreamsSinkConfig) -> EtlResult { + let settings = redis_common::RedisConnectionSettings { + connection_timeout_ms: config.connection_timeout_ms, + response_timeout_ms: config.response_timeout_ms, + connection_retries: config.connection_retries, + connection_max_delay_ms: config.connection_max_delay_ms, + }; + let connection = redis_common::connect(config.url, settings) + .await + .map_err(|error| { + redis_common::map_error(error, "failed to create Redis Streams sink") + })?; Ok(Self { connection: Arc::new(Mutex::new(connection)), @@ -165,12 +208,8 @@ impl Sink for RedisStreamsSink { pipe.add_command(cmd); } - pipe.query_async::<()>(&mut *conn).await.map_err(|e| { - etl::etl_error!( - etl::error::ErrorKind::DestinationError, - "Failed to XADD events to Redis stream", - e.to_string() - ) + pipe.query_async::<()>(&mut *conn).await.map_err(|error| { + redis_common::map_error(error, "failed to publish events to Redis stream") })?; Ok(()) @@ -185,4 +224,30 @@ mod tests { fn test_sink_name() { assert_eq!(RedisStreamsSink::name(), "redis-streams"); } + + #[test] + fn test_connection_settings_deserialize_and_remain_safe_to_serialize() { + let config: RedisStreamsSinkConfig = serde_yaml::from_str( + r#" +url: redis://user:secret@localhost:6379 +stream_name: events +max_len: 1000 +connection_timeout_ms: 1000 +response_timeout_ms: 5000 +connection_retries: 2 +connection_max_delay_ms: 750 +"#, + ) + .unwrap(); + + assert_eq!(config.connection_timeout_ms, Some(1000)); + assert_eq!(config.response_timeout_ms, Some(5000)); + assert_eq!(config.connection_retries, Some(2)); + assert_eq!(config.connection_max_delay_ms, Some(750)); + + let serialized = + serde_yaml::to_string(&RedisStreamsSinkConfigWithoutSecrets::from(&config)).unwrap(); + assert!(!serialized.contains("secret")); + assert!(serialized.contains("response_timeout_ms: 5000")); + } } diff --git a/src/sink/redis_strings.rs b/src/sink/redis_strings.rs index 74a4db9..ddb0b47 100644 --- a/src/sink/redis_strings.rs +++ b/src/sink/redis_strings.rs @@ -20,7 +20,7 @@ use serde::{Deserialize, Serialize}; use std::sync::Arc; use tokio::sync::Mutex; -use crate::sink::Sink; +use crate::sink::{Sink, redis_common}; use crate::types::TriggeredEvent; /// Configuration for the Redis Strings sink. @@ -36,6 +36,22 @@ pub struct RedisStringsSinkConfig { /// Optional prefix for all keys. #[serde(default)] pub key_prefix: Option, + + /// Timeout for establishing a Redis connection, in milliseconds. + #[serde(default)] + pub connection_timeout_ms: Option, + + /// Timeout for Redis command responses, in milliseconds. + #[serde(default)] + pub response_timeout_ms: Option, + + /// Number of reconnection attempts made by the connection manager. + #[serde(default)] + pub connection_retries: Option, + + /// Maximum delay between reconnection attempts, in milliseconds. + #[serde(default)] + pub connection_max_delay_ms: Option, } /// Configuration for the Redis Strings sink without sensitive data. @@ -45,12 +61,28 @@ pub struct RedisStringsSinkConfig { pub struct RedisStringsSinkConfigWithoutSecrets { /// Optional prefix for all keys. pub key_prefix: Option, + + /// Timeout for establishing a Redis connection, in milliseconds. + pub connection_timeout_ms: Option, + + /// Timeout for Redis command responses, in milliseconds. + pub response_timeout_ms: Option, + + /// Number of reconnection attempts made by the connection manager. + pub connection_retries: Option, + + /// Maximum delay between reconnection attempts, in milliseconds. + pub connection_max_delay_ms: Option, } impl From for RedisStringsSinkConfigWithoutSecrets { fn from(config: RedisStringsSinkConfig) -> Self { Self { key_prefix: config.key_prefix, + connection_timeout_ms: config.connection_timeout_ms, + response_timeout_ms: config.response_timeout_ms, + connection_retries: config.connection_retries, + connection_max_delay_ms: config.connection_max_delay_ms, } } } @@ -59,6 +91,10 @@ impl From<&RedisStringsSinkConfig> for RedisStringsSinkConfigWithoutSecrets { fn from(config: &RedisStringsSinkConfig) -> Self { Self { key_prefix: config.key_prefix.clone(), + connection_timeout_ms: config.connection_timeout_ms, + response_timeout_ms: config.response_timeout_ms, + connection_retries: config.connection_retries, + connection_max_delay_ms: config.connection_max_delay_ms, } } } @@ -82,11 +118,18 @@ impl RedisStringsSink { /// # Errors /// /// Returns an error if the Redis connection cannot be established. - pub async fn new( - config: RedisStringsSinkConfig, - ) -> Result> { - let client = redis::Client::open(config.url)?; - let connection = ConnectionManager::new(client).await?; + pub async fn new(config: RedisStringsSinkConfig) -> EtlResult { + let settings = redis_common::RedisConnectionSettings { + connection_timeout_ms: config.connection_timeout_ms, + response_timeout_ms: config.response_timeout_ms, + connection_retries: config.connection_retries, + connection_max_delay_ms: config.connection_max_delay_ms, + }; + let connection = redis_common::connect(config.url, settings) + .await + .map_err(|error| { + redis_common::map_error(error, "failed to create Redis Strings sink") + })?; Ok(Self { connection: Arc::new(Mutex::new(connection)), @@ -130,13 +173,9 @@ impl Sink for RedisStringsSink { pipe.set(&key, event.payload.to_string()); } - pipe.query_async::<()>(&mut *conn).await.map_err(|e| { - etl::etl_error!( - etl::error::ErrorKind::DestinationError, - "Failed to publish events to Redis", - e.to_string() - ) - })?; + pipe.query_async::<()>(&mut *conn) + .await + .map_err(|error| redis_common::map_error(error, "failed to publish events to Redis"))?; Ok(()) } @@ -150,4 +189,29 @@ mod tests { fn test_sink_name() { assert_eq!(RedisStringsSink::name(), "redis-strings"); } + + #[test] + fn test_connection_settings_deserialize_and_remain_safe_to_serialize() { + let config: RedisStringsSinkConfig = serde_yaml::from_str( + r#" +url: redis://user:secret@localhost:6379 +key_prefix: events +connection_timeout_ms: 1000 +response_timeout_ms: 5000 +connection_retries: 2 +connection_max_delay_ms: 750 +"#, + ) + .unwrap(); + + assert_eq!(config.connection_timeout_ms, Some(1000)); + assert_eq!(config.response_timeout_ms, Some(5000)); + assert_eq!(config.connection_retries, Some(2)); + assert_eq!(config.connection_max_delay_ms, Some(750)); + + let serialized = + serde_yaml::to_string(&RedisStringsSinkConfigWithoutSecrets::from(&config)).unwrap(); + assert!(!serialized.contains("secret")); + assert!(serialized.contains("response_timeout_ms: 5000")); + } } diff --git a/tests/redis_streams_sink_tests.rs b/tests/redis_streams_sink_tests.rs index f18e3ce..644b82a 100644 --- a/tests/redis_streams_sink_tests.rs +++ b/tests/redis_streams_sink_tests.rs @@ -1,7 +1,7 @@ #![allow(clippy::indexing_slicing)] //! Integration tests for the Redis Streams sink. -#![cfg(feature = "sink-redis-streams")] +#![cfg(all(feature = "sink-redis-streams", feature = "test-utils"))] use postgres_stream::sink::Sink; use postgres_stream::sink::redis_streams::{RedisStreamsSink, RedisStreamsSinkConfig}; @@ -36,6 +36,10 @@ async fn test_redis_streams_sink_publishes_events() { url: redis_url.clone(), stream_name: Some(stream_name.to_string()), max_len: None, + connection_timeout_ms: None, + response_timeout_ms: None, + connection_retries: None, + connection_max_delay_ms: None, }; let sink = RedisStreamsSink::new(config) @@ -96,6 +100,10 @@ async fn test_redis_streams_sink_with_max_len() { url: redis_url.clone(), stream_name: Some(stream_name.to_string()), max_len: Some(5), + connection_timeout_ms: None, + response_timeout_ms: None, + connection_retries: None, + connection_max_delay_ms: None, }; let sink = RedisStreamsSink::new(config) @@ -135,6 +143,10 @@ async fn test_redis_streams_sink_empty_batch() { url: redis_url, stream_name: Some("pgstream:empty-test".to_string()), max_len: None, + connection_timeout_ms: None, + response_timeout_ms: None, + connection_retries: None, + connection_max_delay_ms: None, }; let sink = RedisStreamsSink::new(config) @@ -158,6 +170,10 @@ async fn test_redis_streams_sink_uses_stream_from_metadata() { url: redis_url.clone(), stream_name: None, max_len: None, + connection_timeout_ms: None, + response_timeout_ms: None, + connection_retries: None, + connection_max_delay_ms: None, }; let sink = RedisStreamsSink::new(config) diff --git a/tests/redis_strings_sink_tests.rs b/tests/redis_strings_sink_tests.rs index 383dd64..aaafa1b 100644 --- a/tests/redis_strings_sink_tests.rs +++ b/tests/redis_strings_sink_tests.rs @@ -1,7 +1,7 @@ #![allow(clippy::indexing_slicing)] //! Integration tests for the Redis Strings sink. -#![cfg(feature = "sink-redis-strings")] +#![cfg(all(feature = "sink-redis-strings", feature = "test-utils"))] use postgres_stream::sink::Sink; use postgres_stream::sink::redis_strings::{RedisStringsSink, RedisStringsSinkConfig}; @@ -34,6 +34,10 @@ async fn test_redis_strings_sink_publishes_events() { let config = RedisStringsSinkConfig { url: redis_url.clone(), key_prefix: None, + connection_timeout_ms: None, + response_timeout_ms: None, + connection_retries: None, + connection_max_delay_ms: None, }; let sink = RedisStringsSink::new(config) @@ -78,6 +82,10 @@ async fn test_redis_strings_sink_with_key_prefix() { let config = RedisStringsSinkConfig { url: redis_url.clone(), key_prefix: Some("pgstream".to_string()), + connection_timeout_ms: None, + response_timeout_ms: None, + connection_retries: None, + connection_max_delay_ms: None, }; let sink = RedisStringsSink::new(config) @@ -114,6 +122,10 @@ async fn test_redis_strings_sink_empty_batch() { let config = RedisStringsSinkConfig { url: redis_url, key_prefix: None, + connection_timeout_ms: None, + response_timeout_ms: None, + connection_retries: None, + connection_max_delay_ms: None, }; let sink = RedisStringsSink::new(config) @@ -134,6 +146,10 @@ async fn test_redis_strings_sink_uses_key_from_metadata() { let config = RedisStringsSinkConfig { url: redis_url.clone(), key_prefix: None, + connection_timeout_ms: None, + response_timeout_ms: None, + connection_retries: None, + connection_max_delay_ms: None, }; let sink = RedisStringsSink::new(config)