add arbiter-engine error handling

arbiter-engine/Cargo.toml

@@ -23,7 +23,6 @@ serde_json.workspace = true
 serde.workspace = true
 tokio.workspace = true
 async-stream.workspace = true
-anyhow = { version = "=1.0.79" }
 tracing.workspace = true
 tokio-stream = "0.1.14"
 futures = "0.3.30"

arbiter-engine/src/agent.rs

@@ -1,15 +1,7 @@
 //! The agent module contains the core agent abstraction for the Arbiter Engine.
 
-use std::{fmt::Debug, sync::Arc};
-
-use arbiter_core::middleware::RevmMiddleware;
-use serde::{de::DeserializeOwned, Serialize};
-use thiserror::Error;
-
-use crate::{
-    machine::{Behavior, Engine, StateMachine},
-    messager::Messager,
-};
+use super::*;
+use crate::machine::{Behavior, Engine, StateMachine};
 
 /// An agent is an entity capable of processing events and producing actions.
 /// These are the core actors in simulations or in onchain systems.
@@ -21,6 +13,7 @@
 /// each of its [`Behavior`]s `startup()` methods. The [`Behavior`]s themselves
 /// will return a stream of events that then let the [`Behavior`] move into the
 /// `State::Processing` stage.
+#[derive(Debug)]
 pub struct Agent {
     /// Identifier for this agent.
     /// Used for routing messages.
@@ -38,17 +31,6 @@
     pub(crate) behavior_engines: Vec<Box<dyn StateMachine>>,
 }
 
-impl Debug for Agent {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        f.debug_struct("Agent")
-            .field("id", &self.id)
-            .field("messager", &self.messager)
-            .field("client", &self.client)
-            .field("behavior_engines", &self.behavior_engines)
-            .finish()
-    }
-}
-
 impl Agent {
     /// Creates a new [`AgentBuilder`] instance with a specified identifier.
     ///
@@ -119,7 +101,7 @@
     /// # Returns
     ///
     /// Returns the `AgentBuilder` instance to allow for method chaining.
-    pub fn with_engine(mut self, engine: Box<dyn StateMachine>) -> Self {
+    pub(crate) fn with_engine(mut self, engine: Box<dyn StateMachine>) -> Self {
         if let Some(engines) = &mut self.behavior_engines {
             engines.push(engine);
         } else {
@@ -165,24 +147,17 @@
         self,
         client: Arc<RevmMiddleware>,
         messager: Messager,
-    ) -> Result<Agent, AgentBuildError> {
+    ) -> Result<Agent, ArbiterEngineError> {
         match self.behavior_engines {
             Some(engines) => Ok(Agent {
                 id: self.id,
                 messager,
                 client,
                 behavior_engines: engines,
             }),
-            None => Err(AgentBuildError::MissingBehaviorEngines),
+            None => Err(ArbiterEngineError::AgentBuildError(
+                "Missing behavior engines".to_owned(),
+            )),
         }
     }
 }
-
-/// enum representing the possible error states encountered by the agent builder
-#[derive(Debug, Error, Clone, PartialEq, Eq)]
-pub enum AgentBuildError {
-    /// Error representing the case where the agent is missing behavior engines;
-    /// an agent has to have behaviors to be useful!
-    #[error("Agent is missing behavior engines")]
-    MissingBehaviorEngines,
-}

arbiter-engine/src/errors.rs

@@ -0,0 +1,45 @@
+//! Error types for the arbiter engine.
+
+use thiserror::Error;
+
+use super::*;
+
+/// Errors that can occur in the arbiter engine.
+#[derive(Debug, Error)]
+pub enum ArbiterEngineError {
+    /// Error occurred with the [`Messager`].
+    #[error("MessagerError: {0}")]
+    MessagerError(String),
+
+    /// Error occurred with the [`crate::agent::Agent`].
+    #[error("AgentBuildError: {0}")]
+    AgentBuildError(String),
+
+    /// Error occurred with the [`crate::world::World`].
+    #[error("WorldError: {0}")]
+    WorldError(String),
+
+    /// Error occurred with the [`crate::universe::Universe`].
+    #[error("UniverseError: {0}")]
+    UniverseError(String),
+
+    /// Error occurred in joining a task.
+    #[error(transparent)]
+    JoinError(#[from] tokio::task::JoinError),
+
+    /// Error occurred in sending a message.
+    #[error(transparent)]
+    SendError(#[from] tokio::sync::broadcast::error::SendError<crate::messager::Message>),
+
+    /// Error occurred in deserializing json.
+    #[error(transparent)]
+    SerdeJsonError(#[from] serde_json::Error),
+
+    /// Error occurred in reading in a file.
+    #[error(transparent)]
+    IoError(#[from] std::io::Error),
+
+    /// Error occurred in deserializing toml.
+    #[error(transparent)]
+    TomlError(#[from] toml::de::Error),
+}

arbiter-engine/src/examples/minter/behaviors/token_admin.rs

@@ -31,7 +31,7 @@ impl Behavior<Message> for TokenAdmin {
         &mut self,
         client: Arc<RevmMiddleware>,
         messager: Messager,
-    ) -> EventStream<Message> {
+    ) -> Result<EventStream<Message>, ArbiterEngineError> {
         self.messager = Some(messager.clone());
         self.client = Some(client.clone());
         for token_data in self.token_data.values_mut() {
@@ -53,12 +53,12 @@ impl Behavior<Message> for TokenAdmin {
                 .get_or_insert_with(HashMap::new)
                 .insert(token_data.name.clone(), token.clone());
         }
-        Box::pin(messager.stream())
+        messager.stream()
     }
 
     #[tracing::instrument(skip(self), fields(id =
  self.messager.as_ref().unwrap().id.as_deref()))]
-    async fn process(&mut self, event: Message) -> Option<MachineHalt> {
+    async fn process(&mut self, event: Message) -> Result<ControlFlow, ArbiterEngineError> {
         if self.tokens.is_none() {
             error!(
                 "There were no tokens to deploy! You must add tokens to
@@ -98,10 +98,10 @@ impl Behavior<Message> for TokenAdmin {
                 self.count += 1;
                 if self.count == self.max_count.unwrap_or(u64::MAX) {
                     warn!("Reached max count. Halting behavior.");
-                    return Some(MachineHalt);
+                    return Ok(ControlFlow::Halt);
                 }
             }
         }
-        None
+        Ok(ControlFlow::Continue)
     }
 }

arbiter-engine/src/examples/minter/behaviors/token_requester.rs

@@ -2,7 +2,10 @@ use arbiter_bindings::bindings::arbiter_token::TransferFilter;
 use arbiter_core::data_collection::EventLogger;
 use token_admin::{MintRequest, TokenAdminQuery};
 
-use self::{examples::minter::agents::token_requester::TokenRequester, machine::EventStream};
+use self::{
+    errors::ArbiterEngineError, examples::minter::agents::token_requester::TokenRequester,
+    machine::EventStream,
+};
 use super::*;
 
 #[async_trait::async_trait]
@@ -12,14 +15,14 @@ impl Behavior<TransferFilter> for TokenRequester {
         &mut self,
         client: Arc<RevmMiddleware>,
         mut messager: Messager,
-    ) -> EventStream<TransferFilter> {
+    ) -> Result<EventStream<TransferFilter>, ArbiterEngineError> {
         messager
             .send(
                 To::Agent(self.request_to.clone()),
                 &TokenAdminQuery::AddressOf(self.token_data.name.clone()),
             )
             .await;
-        let message = messager.get_next().await;
+        let message = messager.get_next().await.unwrap();
         let token_address = serde_json::from_str::<Address>(&message.data).unwrap();
         let token = ArbiterToken::new(token_address, client.clone());
         self.token_data.address = Some(token_address);
@@ -35,18 +38,18 @@ impl Behavior<TransferFilter> for TokenRequester {
 
         self.messager = Some(messager.clone());
         self.client = Some(client.clone());
-        return Box::pin(
+        Ok(Box::pin(
             EventLogger::builder()
                 .add_stream(token.transfer_filter())
                 .stream()
                 .unwrap()
                 .map(|value| serde_json::from_str(&value).unwrap()),
-        );
+        ))
     }
 
     #[tracing::instrument(skip(self), fields(id =
  self.messager.as_ref().unwrap().id.as_deref()))]
-    async fn process(&mut self, event: TransferFilter) -> Option<MachineHalt> {
+    async fn process(&mut self, event: TransferFilter) -> Result<ControlFlow, ArbiterEngineError> {
         let messager = self.messager.as_ref().unwrap();
         while (self.count < self.max_count.unwrap()) {
             debug!("sending message from requester");
@@ -60,6 +63,6 @@ impl Behavior<TransferFilter> for TokenRequester {
                 .await;
             self.count += 1;
         }
-        Some(MachineHalt)
+        Ok(ControlFlow::Halt)
     }
 }

arbiter-engine/src/examples/minter/mod.rs

@@ -14,7 +14,7 @@ use tracing::error;
 use super::*;
 use crate::{
     agent::Agent,
-    machine::{Behavior, MachineHalt, MachineInstruction, StateMachine},
+    machine::{Behavior, ControlFlow, MachineInstruction, StateMachine},
     messager::To,
     world::World,
 };

arbiter-engine/src/examples/mod.rs

@@ -11,8 +11,9 @@ use futures_util::{stream, StreamExt};
 use super::*;
 use crate::{
     agent::Agent,
+    errors::ArbiterEngineError,
     machine::{
-        Behavior, CreateStateMachine, Engine, EventStream, MachineHalt, State, StateMachine,
+        Behavior, ControlFlow, CreateStateMachine, Engine, EventStream, State, StateMachine,
     },
     messager::{Message, Messager, To},
     world::World,

arbiter-engine/src/examples/timed_message/mod.rs

@@ -16,10 +16,6 @@ fn default_max_count() -> Option<u64> {
     Some(3)
 }
 
-fn default_max_count() -> Option<u64> {
-    Some(3)
-}
-
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub(crate) struct TimedMessage {
     delay: u64,
@@ -60,24 +56,24 @@ impl Behavior<Message> for TimedMessage {
         &mut self,
         _client: Arc<RevmMiddleware>,
         messager: Messager,
-    ) -> EventStream<Message> {
+    ) -> Result<EventStream<Message>, ArbiterEngineError> {
         if let Some(startup_message) = &self.startup_message {
             messager.send(To::All, startup_message).await;
         }
         self.messager = Some(messager.clone());
-        return messager.stream();
+        messager.stream()
     }
 
-    async fn process(&mut self, event: Message) -> Option<MachineHalt> {
+    async fn process(&mut self, event: Message) -> Result<ControlFlow, ArbiterEngineError> {
         if event.data == serde_json::to_string(&self.receive_data).unwrap() {
             let messager = self.messager.clone().unwrap();
             messager.send(To::All, self.send_data.clone()).await;
             self.count += 1;
         }
         if self.count == self.max_count.unwrap_or(u64::MAX) {
-            return Some(MachineHalt);
+            return Ok(ControlFlow::Halt);
         }
-        return None;
+        Ok(ControlFlow::Continue)
     }
 }
 
@@ -98,7 +94,7 @@ async fn echoer() {
 
     world.run().await;
 
-    let mut stream = Box::pin(messager.stream());
+    let mut stream = messager.stream().unwrap();
     let mut idx = 0;
 
     loop {
@@ -140,7 +136,7 @@ async fn ping_pong() {
     let messager = world.messager.for_agent("outside_world");
     world.run().await;
 
-    let mut stream = Box::pin(messager.stream());
+    let mut stream = messager.stream().unwrap();
     let mut idx = 0;
 
     loop {
@@ -181,7 +177,7 @@ async fn ping_pong_two_agent() {
     let messager = world.messager.for_agent("outside_world");
     world.run().await;
 
-    let mut stream = Box::pin(messager.stream());
+    let mut stream = messager.stream().unwrap();
     let mut idx = 0;
 
     loop {

arbiter-engine/src/lib.rs

@@ -5,14 +5,18 @@
 //! distributed fashion where each agent is running in its own process and
 //! communicating with other agents via a messaging layer.
 
-use std::collections::HashMap;
+use std::{collections::HashMap, fmt::Debug, sync::Arc};
 
-use anyhow::{anyhow, Result};
-use serde::{Deserialize, Serialize};
-#[allow(unused)]
+use arbiter_core::middleware::RevmMiddleware;
+use futures_util::future::join_all;
+use serde::{de::DeserializeOwned, Deserialize, Serialize};
+use tokio::task::{spawn, JoinError};
 use tracing::{debug, trace, warn};
 
+use crate::{errors::ArbiterEngineError, messager::Messager};
+
 pub mod agent;
+pub mod errors;
 pub mod examples;
 pub mod machine;
 pub mod messager;