fix: improve startup diagnostics and Redis resilience

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
 

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

src/config/load.rs

@@ -74,9 +74,9 @@ impl TryFrom<String> 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`",
+            )),
         }
     }
 }

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",
+        }
+    }
+}

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,

src/error.rs

@@ -0,0 +1,116 @@
+use etl::error::EtlError;
+use std::{backtrace::Backtrace, error::Error, fmt};
+
+pub type PgStreamResult<T> = Result<T, PgStreamError>;
+
+/// 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<dyn Error + Send + Sync>, CapturedBacktrace),
+    Io(std::io::Error, CapturedBacktrace),
+}
+
+impl PgStreamError {
+    pub fn config<E: Error + Send + Sync + 'static>(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<EtlError> for PgStreamError {
+    fn from(error: EtlError) -> Self {
+        Self::Etl(error)
+    }
+}
+
+impl From<std::io::Error> 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")
+    )
+}