Partial transition to anyhow / thiserror for errors

Signed-off-by: Heinz N. Gies <[email protected]>

.github/workflows/checks.yaml

@@ -24,27 +24,11 @@ jobs:
   unused-deps:
     runs-on: ubuntu-latest
     steps:
-      - name: Install deps
-        run: sudo snap install remarshal
-      - name: Install deps
-        run: sudo snap install --classic ripgrep
       - uses: actions/checkout@v3
-      - name: Check for unused dependencies (tremor-runtime)
-        run: ./.github/checks/deps.sh .
-      - name: Check for unused dependencies (tremor-influx)
-        run: ./.github/checks/deps.sh tremor-influx
-      - name: Check for unused dependencies (tremor-pipeline)
-        run: ./.github/checks/deps.sh tremor-pipeline
-      - name: Check for unused dependencies (tremor-script)
-        run: ./.github/checks/deps.sh tremor-script
-      - name: Check for unused dependencies (tremor-cli)
-        run: ./.github/checks/deps.sh tremor-cli
-      - name: Check for unused dependencies (tremor-common)
-        run: ./.github/checks/deps.sh tremor-common
-      - name: Check for unused dependencies (tremor-value)
-        run: ./.github/checks/deps.sh tremor-value
-      - name: Check for unused dependencies (tremor-codec)
-        run: ./.github/checks/deps.sh tremor-codec
+      - name: Install deps
+        run: cargo install cargo-machete
+      - name: Check for unused dependencies
+        run: cargo machete --with-metadata
 
   format:
     runs-on: ubuntu-latest

.gitignore

@@ -54,3 +54,5 @@ tremor-script/current_counterexample.eqc
 .vscode/dryrun.log
 tremor-script-nif/libpath
 docs-test
+*.mm_profdata
+chrome_profiler.json

Cargo.toml

@@ -33,6 +33,7 @@ lto = "thin"
 opt-level = 3
 
 [dependencies]
+thiserror = "1"
 tokio = { version = "1.34", features = ["full"] }
 tokio-stream = "0.1"
 anyhow = "1"
@@ -64,11 +65,9 @@ jumphash = "0.1"
 clickhouse-rs = { git = "https://github.com/suharev7/clickhouse-rs", rev = "73d39ba" }
 dashmap = "5.5"
 either = { version = "1.9", features = ["serde"] }
-error-chain = "0.12"
 file-mode = "0.1"
 futures = "0.3.29"
 event-listener = "3"
-glob = "0.3"
 halfbrown = "0.2"
 hashbrown = { version = "0.14", features = ["serde"] }
 hex = "0.4"
@@ -138,9 +137,6 @@ rdkafka-sys = { version = "4.6", features = [
 # crononome
 cron = "0.12"
 
-# logstash grok patterns
-grok = "2"
-
 # discord
 serenity = { version = "0.11", default-features = false, features = [
   "client",
@@ -168,7 +164,6 @@ tremor-otelapis = { version = "=0.2.4" }
 aws-sdk-s3 = "0.35"
 aws-types = "0.57"
 aws-config = "0.57"
-aws-smithy-http = "0.57"
 
 # gcp
 googapis = { version = "0.6", default-features = false, features = [
@@ -204,7 +199,6 @@ sha2 = "0.10"
 rmp-serde = "1"
 
 [dev-dependencies]
-port_scanner = "0.1"
 serial_test = { version = "2.0", features = ["logging"] }
 env_logger = "0.10"
 matches = "0.1"
@@ -213,12 +207,9 @@ proptest = "1.1"
 regex = "1"
 # We downgraded to 0.6 because:
 # in the face of high concurrency serial_test 0.7.0 panics after 60 seconds
-signal-hook = "0.3"
-signal-hook-tokio = "0.3"
 tempfile = { version = "3.8" }
 test-case = "3.1"
 testcontainers = { version = "0.14", features = ["watchdog"] }
-num_cpus = "1"
 bytes = "1"
 
 [features]

src/channel.rs

@@ -18,6 +18,25 @@ pub(crate) type UnboundedSender<T> = tokio::sync::mpsc::UnboundedSender<T>;
 pub(crate) type UnboundedReceiver<T> = tokio::sync::mpsc::UnboundedReceiver<T>;
 pub(crate) type OneShotSender<T> = tokio::sync::oneshot::Sender<T>;
 // pub(crate) type OneShotReceiver<T> = tokio::sync::oneshot::Receiver<T>;
-pub(crate) use tokio::sync::mpsc::error::{SendError, TryRecvError};
+pub(crate) use tokio::sync::mpsc::error::SendError;
 pub(crate) use tokio::sync::mpsc::{channel as bounded, unbounded_channel as unbounded};
 pub(crate) use tokio::sync::oneshot::channel as oneshot;
+
+#[derive(Debug, thiserror::Error)]
+pub(crate) enum ChannelError {
+    #[error("Could not send")]
+    Send,
+    #[error("Could not recv")]
+    Recv,
+    #[error("The channel is empty")]
+    Empty,
+}
+
+#[allow(clippy::needless_pass_by_value)]
+pub(crate) fn send_e<T>(_: crate::channel::SendError<T>) -> anyhow::Error {
+    ChannelError::Send.into()
+}
+
+pub(crate) fn empty_e() -> ChannelError {
+    ChannelError::Empty
+}

src/config.rs

@@ -113,7 +113,7 @@ impl Connector {
         defn: &ast::ConnectorDefinition<'static>,
     ) -> crate::Result<Self> {
         let aggr_reg = tremor_script::registry::aggr();
-        let reg = &*FN_REGISTRY.read()?;
+        let reg = &*FN_REGISTRY.read().map_err(|_| ErrorKind::ReadLock)?;
 
         let mut helper = Helper::new(reg, &aggr_reg);
         let params = defn.params.clone();
@@ -136,8 +136,8 @@ impl Connector {
             connector_alias: &alias::Connector,
         ) -> Result<()> {
             if v.get(k).is_some() && v.get(k).map(Value::value_type) != Some(t) {
-                return Err(ErrorKind::InvalidConnectorDefinition(
-                    connector_alias.to_string(),
+                return Err(ConnectorError::InvalidDefinition(
+                    connector_alias.clone(),
                     format!(
                         "Expected type {t:?} for key {k} but got {:?}",
                         v.get(k).map_or(ValueType::Null, Value::value_type)
@@ -316,6 +316,6 @@ mod tests {
         let id = alias::Connector::new("my_id");
         let res = Connector::from_config(&id, "fancy_schmancy".into(), &config);
         assert!(res.is_err());
-        assert_eq!(String::from("Invalid Definition for connector \"my_id\": Expected type I64 for key metrics_interval_s but got String"), res.err().map(|e| e.to_string()).unwrap_or_default());
+        assert_eq!(String::from("[my_id] Invalid definition: Expected type I64 for key metrics_interval_s but got String"), res.err().map(|e| e.to_string()).unwrap_or_default());
     }
 }