bridge: Error refactoring (#1342)

Preparation for adding kafka `ReceiverOutput`.

Part of svix/monorepo-private#8508.

bridge/.config/nextest.toml

@@ -0,0 +1,2 @@
+[profile.default]
+slow-timeout = { period = "30s", terminate-after = 2 }

bridge/svix-bridge-plugin-kafka/src/input.rs

@@ -199,7 +199,7 @@ impl SenderInput for KafkaConsumer {
         self.transformer_tx = tx;
     }
 
-    async fn run(&self) -> std::io::Result<()> {
+    async fn run(&self) {
         let mut fails: u64 = 0;
         let mut last_fail = Instant::now();
 

bridge/svix-bridge-plugin-kafka/tests/it/kafka_consumer.rs

@@ -173,7 +173,7 @@ async fn test_consume_ok() {
     let plugin = get_test_plugin(mock_server.uri(), topic, None);
 
     let handle = tokio::spawn(async move {
-        plugin.run().await.unwrap();
+        plugin.run().await;
     });
     // Wait for the consumer to connect
     tokio::time::sleep(CONNECT_WAIT_TIME).await;
@@ -245,7 +245,7 @@ async fn test_consume_transformed_json_ok() {
     plugin.set_transformer(Some(transformer_tx));
 
     let handle = tokio::spawn(async move {
-        plugin.run().await.unwrap();
+        plugin.run().await;
     });
     // Wait for the consumer to connect
     tokio::time::sleep(CONNECT_WAIT_TIME).await;
@@ -330,7 +330,7 @@ async fn test_consume_transformed_string_ok() {
     plugin.set_transformer(Some(transformer_tx));
 
     let handle = tokio::spawn(async move {
-        plugin.run().await.unwrap();
+        plugin.run().await;
     });
     // Wait for the consumer to connect
     tokio::time::sleep(CONNECT_WAIT_TIME).await;
@@ -365,7 +365,7 @@ async fn test_missing_app_id_nack() {
     let plugin = get_test_plugin(mock_server.uri(), topic, None);
 
     let handle = tokio::spawn(async move {
-        plugin.run().await.unwrap();
+        plugin.run().await;
     });
 
     // Wait for the consumer to connect
@@ -415,7 +415,7 @@ async fn test_missing_event_type_nack() {
     let plugin = get_test_plugin(mock_server.uri(), topic, None);
 
     let handle = tokio::spawn(async move {
-        plugin.run().await.unwrap();
+        plugin.run().await;
     });
 
     // Wait for the consumer to connect
@@ -467,7 +467,7 @@ async fn test_consume_svix_503() {
     let plugin = get_test_plugin(mock_server.uri(), topic, None);
 
     let handle = tokio::spawn(async move {
-        plugin.run().await.unwrap();
+        plugin.run().await;
     });
     // Wait for the consumer to connect
     tokio::time::sleep(CONNECT_WAIT_TIME).await;
@@ -510,7 +510,7 @@ async fn test_consume_svix_offline() {
     drop(mock_server);
 
     let handle = tokio::spawn(async move {
-        plugin.run().await.unwrap();
+        plugin.run().await;
     });
     // Wait for the consumer to connect
     tokio::time::sleep(CONNECT_WAIT_TIME).await;

bridge/svix-bridge-plugin-queue/Cargo.toml

@@ -9,6 +9,7 @@ edition = "2021"
 serde_json = "1.0"
 serde = { version = "1.0", features = ["derive"] }
 svix-bridge-types = { path = "../svix-bridge-types" }
+thiserror = "1.0.61"
 tokio = { version = "1", features = ["full"] }
 tokio-executor-trait = "2.1"
 tokio-reactor-trait = "1.1"

bridge/svix-bridge-plugin-queue/src/error.rs

@@ -1,43 +1,23 @@
 pub use omniqueue::QueueError;
 use svix_bridge_types::svix;
+use thiserror::Error;
 
+#[derive(Debug, Error)]
 pub enum Error {
-    Payload(String),
-    Json(serde_json::Error),
-    Queue(QueueError),
-    Svix(svix::error::Error),
+    #[error("json error: {0}")]
+    Json(#[from] serde_json::Error),
+    #[error("queue error: {0}")]
+    Queue(#[from] QueueError),
+    #[error("svix API error: {0}")]
+    Svix(#[from] svix::error::Error),
+    #[error("{0}")]
     Generic(String),
 }
-pub type Result<T> = std::result::Result<T, Error>;
-
-impl From<svix::error::Error> for Error {
-    fn from(value: svix::error::Error) -> Self {
-        Error::Svix(value)
-    }
-}
-
-impl From<serde_json::Error> for Error {
-    fn from(value: serde_json::Error) -> Self {
-        Error::Json(value)
-    }
-}
-
-impl From<QueueError> for Error {
-    fn from(value: QueueError) -> Self {
-        Error::Queue(value)
-    }
-}
-
-impl From<String> for Error {
-    fn from(value: String) -> Self {
-        Self::Generic(value)
-    }
-}
+pub type Result<T, E = Error> = std::result::Result<T, E>;
 
 impl From<Error> for std::io::Error {
     fn from(value: Error) -> Self {
         match value {
-            Error::Payload(e) => std::io::Error::new(std::io::ErrorKind::Other, e),
             Error::Json(e) => std::io::Error::new(std::io::ErrorKind::Other, e),
             Error::Queue(e) => std::io::Error::new(std::io::ErrorKind::Other, e),
             Error::Svix(e) => std::io::Error::new(std::io::ErrorKind::Other, e),

bridge/svix-bridge-plugin-queue/src/lib.rs

@@ -197,7 +197,7 @@ trait Consumer {
     }
 }
 
-async fn run_inner(consumer: &(impl Consumer + Send + Sync)) -> std::io::Result<()> {
+async fn run_inner(consumer: &(impl Consumer + Send + Sync)) -> ! {
     let mut fails: u64 = 0;
     let mut last_fail = Instant::now();
     let system_name = consumer.system();

bridge/svix-bridge-plugin-queue/src/receiver_output/mod.rs

@@ -1,7 +1,7 @@
 use std::sync::Arc;
 
 use omniqueue::DynProducer;
-use svix_bridge_types::{async_trait, ForwardRequest, ReceiverOutput};
+use svix_bridge_types::{async_trait, BoxError, ForwardRequest, ReceiverOutput};
 
 use crate::{config::QueueOutputOpts, error::Result};
 
@@ -42,11 +42,9 @@ impl ReceiverOutput for QueueForwarder {
     fn name(&self) -> &str {
         &self.name
     }
-    async fn handle(&self, request: ForwardRequest) -> std::io::Result<()> {
-        Ok(self
-            .sender
-            .send_serde_json(&request.payload)
-            .await
-            .map_err(crate::Error::from)?)
+
+    async fn handle(&self, request: ForwardRequest) -> Result<(), BoxError> {
+        self.sender.send_serde_json(&request.payload).await?;
+        Ok(())
     }
 }

bridge/svix-bridge-plugin-queue/src/sender_input/mod.rs

@@ -101,10 +101,12 @@ impl SenderInput for QueueSender {
     fn name(&self) -> &str {
         &self.name
     }
+
     fn set_transformer(&mut self, tx: Option<TransformerTx>) {
         self.transformer_tx = tx;
     }
-    async fn run(&self) -> std::io::Result<()> {
+
+    async fn run(&self) {
         run_inner(self).await
     }
 }

bridge/svix-bridge-types/src/lib.rs

@@ -137,17 +137,19 @@ pub trait SenderInput: Send {
     /// For plugins that want to run JS transformations on payloads.
     /// Giving them a sender lets them pass messages to the JS executor.
     fn set_transformer(&mut self, _tx: Option<TransformerTx>) {}
-    async fn run(&self) -> std::io::Result<()>;
+    async fn run(&self);
 }
 
+pub type BoxError = Box<dyn std::error::Error + Send + Sync>;
+
 /// Represents something we can hand a webhook payload to.
 /// Aka a "forwarder."
 ///
 /// To start, we're only using this in conjunction with an HTTP server "owned" by the bridge binary.
 #[async_trait]
 pub trait ReceiverOutput: Send + Sync {
     fn name(&self) -> &str;
-    async fn handle(&self, request: ForwardRequest) -> std::io::Result<()>;
+    async fn handle(&self, request: ForwardRequest) -> Result<(), BoxError>;
 }
 
 #[derive(Deserialize, Debug, Clone, Default)]

bridge/svix-bridge/src/main.rs

@@ -166,17 +166,10 @@ async fn supervise_senders(inputs: Vec<Box<dyn SenderInput>>) -> Result<()> {
         set.spawn(async move {
             // FIXME: needs much better signaling for termination
             loop {
-                let fut = input.run();
                 // If this future returns, the consumer terminated unexpectedly.
-                if let Err(e) = fut.await {
-                    tracing::warn!(
-                        "sender input {} unexpectedly terminated: {}",
-                        input.name(),
-                        e
-                    );
-                } else {
-                    tracing::warn!("sender input {} unexpectedly terminated", input.name());
-                }
+                input.run().await;
+
+                tracing::warn!("sender input {} unexpectedly terminated", input.name());
                 tokio::time::sleep(Duration::from_secs(1)).await;
             }
         });

bridge/svix-bridge/src/webhook_receiver/tests.rs

@@ -6,8 +6,9 @@ use axum::{
 };
 use serde_json::json;
 use svix_bridge_types::{
-    async_trait, svix::webhooks::Webhook, ForwardRequest, ReceiverOutput, TransformationConfig,
-    TransformerInput, TransformerInputFormat, TransformerJob, TransformerOutput,
+    async_trait, svix::webhooks::Webhook, BoxError, ForwardRequest, ReceiverOutput,
+    TransformationConfig, TransformerInput, TransformerInputFormat, TransformerJob,
+    TransformerOutput,
 };
 use tower::{Service, ServiceExt};
 
@@ -37,8 +38,8 @@ impl ReceiverOutput for FakeReceiverOutput {
         "fake output"
     }
 
-    async fn handle(&self, request: ForwardRequest) -> std::io::Result<()> {
-        self.tx.send(request.payload).unwrap();
+    async fn handle(&self, request: ForwardRequest) -> Result<(), BoxError> {
+        self.tx.send(request.payload)?;
         Ok(())
     }
 }