ref(error): Improve error handling (#597)

* ref(error): Improve error handling

* Improve

* Improve

Author: iambriccardo

Cargo.lock

@@ -1,15 +1,16 @@
 use anyhow::{Context, anyhow};
 use etl_api::{config::ApiConfig, startup::Application};
-use etl_config::{Environment, load_config, shared::PgConnectionConfig};
+use etl_config::{load_config, shared::PgConnectionConfig};
 use etl_telemetry::tracing::init_tracing;
-use secrecy::ExposeSecret;
 use std::env;
-use std::sync::{Arc, Once};
+use std::sync::Once;
 use tracing::{error, info};
 
 /// Ensures crypto provider is only initialized once.
 static INIT_CRYPTO: Once = Once::new();
 
+mod sentry;
+
 /// Installs the default cryptographic provider for rustls.
 ///
 /// Uses AWS LC cryptographic provider and ensures it's only installed once
@@ -38,7 +39,7 @@ fn main() -> anyhow::Result<()> {
     let _log_flusher = init_tracing(env!("CARGO_BIN_NAME"))?;
 
     // Initialize Sentry before the async runtime starts
-    let _sentry_guard = init_sentry()?;
+    let _sentry_guard = sentry::init()?;
 
     // We start the runtime.
     actix_web::rt::System::new().block_on(async_main())?;
@@ -86,62 +87,6 @@ async fn async_main() -> anyhow::Result<()> {
     Ok(())
 }
 
-/// Initializes Sentry error tracking and performance monitoring.
-///
-/// Configures Sentry with environment-specific settings, service tagging, and a custom
-/// traces sampler that reduces sampling for high-frequency endpoints like `/metrics` and
-/// `/health_check`.
-/// Returns `None` if Sentry configuration is not provided.
-fn init_sentry() -> anyhow::Result<Option<sentry::ClientInitGuard>> {
-    if let Ok(config) = load_config::<ApiConfig>()
-        && let Some(sentry_config) = &config.sentry
-    {
-        info!("initializing sentry with supplied dsn");
-
-        let environment = Environment::load()?;
-        let guard = sentry::init(sentry::ClientOptions {
-            dsn: Some(sentry_config.dsn.expose_secret().parse()?),
-            environment: Some(environment.to_string().into()),
-            traces_sampler: Some(Arc::new(|ctx: &sentry::TransactionContext| {
-                let transaction_name = ctx.name();
-                let endpoint = transaction_name
-                    .split_once(' ')
-                    .and_then(|(method, path)| {
-                        if path.starts_with('/') && method.chars().all(|c| c.is_ascii_uppercase()) {
-                            Some(path)
-                        } else {
-                            None
-                        }
-                    })
-                    .unwrap_or(transaction_name);
-
-                // Sample verbose endpoints at a much lower rate (0.1%)
-                match endpoint {
-                    "/metrics" | "/health_check" => 0.001,
-                    // All other endpoints sampled at 1%
-                    _ => 0.01,
-                }
-            })),
-            max_request_body_size: sentry::MaxRequestBodySize::Always,
-            integrations: vec![Arc::new(
-                sentry::integrations::panic::PanicIntegration::new(),
-            )],
-            ..Default::default()
-        });
-
-        // Set service tag to differentiate API from other services
-        sentry::configure_scope(|scope| {
-            scope.set_tag("service", "api");
-        });
-
-        return Ok(Some(guard));
-    }
-
-    info!("sentry not configured for api, skipping initialization");
-
-    Ok(None)
-}
-
 fn log_pg_connection_config(config: &PgConnectionConfig) {
     info!(
         host = config.host,

etl-api/src/main.rs

@@ -0,0 +1,61 @@
+use anyhow::Result;
+use etl_api::config::ApiConfig;
+use etl_config::{Environment, load_config};
+use secrecy::ExposeSecret;
+use std::sync::Arc;
+use tracing::info;
+
+/// Initializes Sentry error tracking and performance monitoring for the API.
+///
+/// Returns [`None`] when no Sentry configuration is provided.
+pub fn init() -> Result<Option<sentry::ClientInitGuard>> {
+    if let Ok(config) = load_config::<ApiConfig>()
+        && let Some(sentry_config) = &config.sentry
+    {
+        info!("initializing sentry with supplied dsn");
+
+        let environment = Environment::load()?;
+        let guard = sentry::init(sentry::ClientOptions {
+            dsn: Some(sentry_config.dsn.expose_secret().parse()?),
+            environment: Some(environment.to_string().into()),
+            traces_sampler: Some(Arc::new(|ctx: &sentry::TransactionContext| {
+                sample_trace_rate(ctx)
+            })),
+            max_request_body_size: sentry::MaxRequestBodySize::Always,
+            integrations: vec![Arc::new(
+                sentry::integrations::panic::PanicIntegration::new(),
+            )],
+            attach_stacktrace: true,
+            ..Default::default()
+        });
+
+        sentry::configure_scope(|scope| {
+            scope.set_tag("service", "api");
+        });
+
+        return Ok(Some(guard));
+    }
+
+    info!("sentry not configured for api, skipping initialization");
+    Ok(None)
+}
+
+/// Computes the trace sampling rate based on endpoint path.
+fn sample_trace_rate(ctx: &sentry::TransactionContext) -> f32 {
+    let transaction_name = ctx.name();
+    let endpoint = transaction_name
+        .split_once(' ')
+        .and_then(|(method, path)| {
+            if path.starts_with('/') && method.chars().all(|c| c.is_ascii_uppercase()) {
+                Some(path)
+            } else {
+                None
+            }
+        })
+        .unwrap_or(transaction_name);
+
+    match endpoint {
+        "/metrics" | "/health_check" => 0.001,
+        _ => 0.01,
+    }
+}

etl-api/src/sentry.rs

@@ -17,7 +17,6 @@ etl-config = { workspace = true, features = ["supabase"] }
 etl-destinations = { workspace = true, features = ["bigquery", "iceberg"] }
 etl-telemetry = { workspace = true }
 
-anyhow = { workspace = true, features = ["std"] }
 configcat = { workspace = true }
 metrics = { workspace = true }
 reqwest = { workspace = true, features = ["rustls-tls", "json"] }

etl-replicator/Cargo.toml

@@ -1,14 +1,15 @@
-use anyhow::Context;
 use etl_config::load_config;
 use etl_config::shared::ReplicatorConfig;
 
+use crate::error::{ReplicatorError, ReplicatorResult};
+
 /// Loads and validates the replicator configuration.
 ///
 /// Uses the standard configuration loading mechanism from [`etl_config`] and
 /// validates the resulting [`ReplicatorConfig`] before returning it.
-pub fn load_replicator_config() -> anyhow::Result<ReplicatorConfig> {
-    let config = load_config::<ReplicatorConfig>().context("loading replicator configuration")?;
-    config.validate()?;
+pub fn load_replicator_config() -> ReplicatorResult<ReplicatorConfig> {
+    let config = load_config::<ReplicatorConfig>().map_err(ReplicatorError::config)?;
+    config.validate().map_err(ReplicatorError::config)?;
 
     Ok(config)
 }

etl-replicator/src/config.rs

@@ -1,5 +1,6 @@
 use std::collections::HashMap;
 
+use crate::error::{ReplicatorError, ReplicatorResult};
 use crate::migrations::migrate_state_store;
 use etl::destination::memory::MemoryDestination;
 use etl::pipeline::Pipeline;
@@ -31,7 +32,7 @@ use tracing::{debug, info, warn};
 /// destinations with proper initialization and error handling.
 pub async fn start_replicator_with_config(
     replicator_config: ReplicatorConfig,
-) -> anyhow::Result<()> {
+) -> ReplicatorResult<()> {
     info!("starting replicator service");
 
     log_config(&replicator_config);
@@ -85,7 +86,7 @@ pub async fn start_replicator_with_config(
                     s3_region,
                 },
         } => {
-            let env = Environment::load()?;
+            let env = Environment::load().map_err(ReplicatorError::config)?;
             let client = IcebergClient::new_with_supabase_catalog(
                 project_ref,
                 env.get_supabase_domain(),
@@ -95,7 +96,8 @@ pub async fn start_replicator_with_config(
                 s3_secret_access_key.expose_secret().to_string(),
                 s3_region.clone(),
             )
-            .await?;
+            .await
+            .map_err(ReplicatorError::config)?;
             let namespace = match namespace {
                 Some(ns) => DestinationNamespace::Single(ns.to_string()),
                 None => DestinationNamespace::OnePerSchema,
@@ -125,7 +127,8 @@ pub async fn start_replicator_with_config(
                     s3_endpoint.clone(),
                 ),
             )
-            .await?;
+            .await
+            .map_err(ReplicatorError::config)?;
             let namespace = match namespace {
                 Some(ns) => DestinationNamespace::Single(ns.to_string()),
                 None => DestinationNamespace::OnePerSchema,
@@ -258,7 +261,7 @@ fn log_batch_config(config: &BatchConfig) {
 async fn init_store(
     pipeline_id: PipelineId,
     pg_connection_config: PgConnectionConfig,
-) -> anyhow::Result<impl StateStore + SchemaStore + CleanupStore + Clone> {
+) -> ReplicatorResult<impl StateStore + SchemaStore + CleanupStore + Clone> {
     migrate_state_store(&pg_connection_config).await?;
 
     Ok(PostgresStore::new(pipeline_id, pg_connection_config))
@@ -270,7 +273,7 @@ async fn init_store(
 /// and ensures proper cleanup on shutdown. The pipeline will attempt to
 /// finish processing current batches before terminating.
 #[tracing::instrument(skip(pipeline))]
-async fn start_pipeline<S, D>(mut pipeline: Pipeline<S, D>) -> anyhow::Result<()>
+async fn start_pipeline<S, D>(mut pipeline: Pipeline<S, D>) -> ReplicatorResult<()>
 where
     S: StateStore + SchemaStore + CleanupStore + Clone + Send + Sync + 'static,
     D: Destination + Clone + Send + Sync + 'static,
@@ -315,7 +318,7 @@ where
     shutdown_handle.abort();
     let _ = shutdown_handle.await;
 
-    // Propagate any pipeline error as anyhow error.
+    // Propagate any pipeline error.
     result?;
 
     Ok(())