feat: added new provider crate for emitters and remote runtime

Author: raphael-goetz

crates/taurus-provider/Cargo.toml

@@ -0,0 +1,19 @@
+[package]
+name = "taurus-provider"
+version.workspace = true
+edition.workspace = true
+
+[dependencies]
+code0-flow = { workspace = true, features = ["flow_service"] }
+tucana = { workspace = true }
+tokio = { workspace = true }
+log = { workspace = true }
+futures-lite = { workspace = true }
+rand = { workspace = true }
+base64 = { workspace = true }
+env_logger = { workspace = true }
+async-nats = { workspace = true }
+prost = { workspace = true }
+tonic-health = { workspace = true }
+tonic = { workspace = true }
+taurus-core = { workspace = true }

crates/taurus-provider/src/lib.rs

@@ -0,0 +1 @@
+pub mod providers;

crates/taurus-provider/src/providers/emitter/mod.rs

@@ -0,0 +1 @@
+pub mod nats_emitter;

crates/taurus-provider/src/providers/emitter/nats_emitter.rs

@@ -0,0 +1,93 @@
+use async_nats::Client;
+use prost::Message;
+use std::collections::HashMap;
+use taurus_core::runtime::engine::{EmitType, ExecutionId, RespondEmitter};
+use tokio::sync::mpsc;
+use tucana::shared::value::Kind::{StringValue, StructValue};
+use tucana::shared::{Struct, Value};
+
+const DEFAULT_TOPIC_PREFIX: &str = "runtime.emitter";
+
+pub struct NATSRespondEmitter {
+    tx: mpsc::UnboundedSender<NATSEmitMessage>,
+}
+
+struct NATSEmitMessage {
+    execution_id: ExecutionId,
+    emit_type: EmitType,
+    value: Value,
+}
+
+impl NATSRespondEmitter {
+    pub fn new(client: Client) -> Self {
+        Self::with_topic_prefix(client, DEFAULT_TOPIC_PREFIX)
+    }
+
+    pub fn with_topic_prefix(client: Client, topic_prefix: impl Into<String>) -> Self {
+        let topic_prefix = topic_prefix.into();
+        let (tx, mut rx) = mpsc::unbounded_channel::<NATSEmitMessage>();
+
+        // Keep the public emitter API synchronous while publishing asynchronously.
+        // This worker serializes outbound lifecycle events to one NATS topic per execution:
+        // `<topic_prefix>.<execution_id>`.
+        // Event type is embedded in the payload so subscribers do not need four topic bindings.
+        tokio::spawn(async move {
+            while let Some(message) = rx.recv().await {
+                let topic = format!("{}.{}", topic_prefix, message.execution_id);
+                let encoded_payload = encode_emit_message(message.emit_type, message.value);
+
+                if let Err(err) = client
+                    .publish(topic.clone(), encoded_payload.encode_to_vec().into())
+                    .await
+                {
+                    log::error!(
+                        "Failed to publish runtime emit message on '{}': {:?}",
+                        topic,
+                        err
+                    );
+                }
+            }
+        });
+
+        Self { tx }
+    }
+}
+
+impl RespondEmitter for NATSRespondEmitter {
+    fn emit(&self, execution_id: ExecutionId, emit_type: EmitType, value: Value) {
+        if let Err(err) = self.tx.send(NATSEmitMessage {
+            execution_id,
+            emit_type,
+            value,
+        }) {
+            log::debug!(
+                "Dropped runtime emit message because NATS emitter worker is unavailable: {:?}",
+                err
+            );
+        }
+    }
+}
+
+fn encode_emit_message(emit_type: EmitType, payload: Value) -> Value {
+    let emit_type_value = Value {
+        kind: Some(StringValue(emit_type_as_str(emit_type).to_string())),
+    };
+
+    Value {
+        kind: Some(StructValue(Struct {
+            fields: HashMap::from([
+                ("emit_type".to_string(), emit_type_value),
+                ("payload".to_string(), payload),
+            ]),
+        })),
+    }
+}
+
+fn emit_type_as_str(emit_type: EmitType) -> &'static str {
+    match emit_type {
+        EmitType::StartingExec => "starting",
+        EmitType::OngoingExec => "ongoing",
+        EmitType::FinishedExec => "finished",
+        EmitType::FailedExec => "failed",
+    }
+}

crates/taurus-provider/src/providers/mod.rs

@@ -0,0 +1,5 @@
+/// Provider for Remote Runtimes
+pub mod remote;
+
+/// Provider for Remote Emitters
+pub mod emitter;

crates/taurus-provider/src/providers/remote/mod.rs

@@ -0,0 +1 @@
+pub mod nats_remote_runtime;

crates/taurus-provider/src/providers/remote/nats_remote_runtime.rs

@@ -0,0 +1,81 @@
+use async_nats::Client;
+use prost::Message;
+use taurus_core::runtime::remote::{RemoteExecution, RemoteRuntime};
+use taurus_core::types::errors::runtime_error::RuntimeError;
+use tonic::async_trait;
+use tucana::aquila::ExecutionResult;
+use tucana::shared::Value;
+
+pub struct NATSRemoteRuntime {
+    client: Client,
+}
+
+impl NATSRemoteRuntime {
+    pub fn new(client: Client) -> Self {
+        NATSRemoteRuntime { client }
+    }
+}
+
+#[async_trait]
+impl RemoteRuntime for NATSRemoteRuntime {
+    async fn execute_remote(&self, execution: RemoteExecution) -> Result<Value, RuntimeError> {
+        let topic = format!(
+            "action.{}.{}",
+            execution.target_service, execution.request.execution_identifier
+        );
+        let payload = execution.request.encode_to_vec();
+
+        log::info!("Request Remote Runtime Execution with topic: : {}", topic);
+        let res = self.client.request(topic, payload.into()).await;
+        let message = match res {
+            Ok(r) => r,
+            Err(err) => {
+                log::error!(
+                    "RemoteRuntimeExeption: failed to handle NATS message: {}",
+                    err
+                );
+                return Err(RuntimeError::new(
+                    "T-PROV-000001",
+                    "RemoteRuntimeExeption",
+                    "Failed to receive any response messages from a remote runtime.",
+                ));
+            }
+        };
+
+        let decode_result = ExecutionResult::decode(message.payload);
+        let execution_result = match decode_result {
+            Ok(r) => r,
+            Err(err) => {
+                log::error!(
+                    "RemoteRuntimeExeption: failed to decode NATS message: {}",
+                    err
+                );
+                return Err(RuntimeError::new(
+                    "T-PROV-000002",
+                    "RemoteRuntimeExeption",
+                    "Failed to read Remote Response",
+                ));
+            }
+        };
+
+        match execution_result.result {
+            Some(result) => match result {
+                tucana::aquila::execution_result::Result::Success(value) => Ok(value),
+                tucana::aquila::execution_result::Result::Error(err) => {
+                    let code = err.code.to_string();
+                    let description = match err.description {
+                        Some(string) => string,
+                        None => "Unknown Error".to_string(),
+                    };
+                    let error = RuntimeError::new(code, "RemoteExecutionError", description);
+                    Err(error)
+                }
+            },
+            None => Err(RuntimeError::new(
+                "T-PROV-000003",
+                "RemoteRuntimeExeption",
+                "Result of Remote Response was empty.",
+            )),
+        }
+    }
+}

crates/taurus/Cargo.toml

@@ -17,3 +17,4 @@ prost = { workspace = true }
 tonic-health = { workspace = true }
 tonic = { workspace = true }
 taurus-core = { workspace = true }
+taurus-provider = { workspace = true }