[PLATFORM-1080]: Error handling

Cargo.toml

@@ -16,12 +16,14 @@ all-features = true
 
 [features]
 sql = ["sqlx"]
-postgres = ["sql", "sqlx/postgres", "typed-builder"]
+postgres = ["sql", "sqlx/postgres", "typed-builder", "tokio"]
 rebuilder = []
 kafka = ["rdkafka", "typed-builder"]
 rabbit = ["lapin", "typed-builder"]
 
 [dependencies]
+tokio = { version = "1.6", optional = true }
+
 # Serialization/Deserialization
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
@@ -32,8 +34,6 @@ chrono = { version = "0.4", features = ["serde"] }
 # Build async trait
 async-trait = "0.1.50"
 
-# Atomic Arc
-arc-swap = "1.6"
 # Self-referential structs (for PgStoreLockGuard)
 ouroboros = "0.15"
 
@@ -50,9 +50,10 @@ typed-builder = { version = "0.14.0", optional = true }
 futures = "0.3"
 tracing = "0.1"
 
+thiserror = "1.0"
+
 [dev-dependencies]
 tokio = { version = "1.6", features = ["full"] }
-thiserror = "1.0"
 rand = "0.8"
 
 [[example]]

README.md

@@ -2,7 +2,7 @@
 
 It is an opinionated library used to achieve cqrs/es in Rust.
 
-A set of example snippets can be found in the `example` folder.
+A set of example snippets could be found in the `example` folder.
 
 ## Install
 
@@ -38,12 +38,7 @@ cargo make test
 
 Run linters.
 
-```
+```shell
 cargo make clippy
 ```
 
-Finally eventually unset `DATABASE_URL` environment variable.
-
-```shell
-unset DATABASE_URL
-```

entrypoint

@@ -1,10 +1,7 @@
 #!/usr/bin/env bash
 
-source /setup_common.sh
-source /decrypt_secrets.sh
+export DATABASE_URL="postgresql://postgres:postgres@postgres:5432/postgres"
+export RABBIT_URL="amqp://rabbit:rabbit@rabbit/rabbit"
+export KAFKA_BROKERS_URL="kafka:9092"
 
-if [ -n "$1" ]; then
-  sh -c "$@"
-else
-  cargo watch -x 'run'
-fi
+sh -c "$@"

examples/event_bus/main.rs

@@ -91,7 +91,7 @@ async fn main() {
         .await
         .unwrap();
 
-    let manager: AggregateManager<BasicAggregate> = AggregateManager::new(store);
+    let manager: AggregateManager<PgStore<BasicAggregate>> = AggregateManager::new(store);
 
     let content: &str = "view row content";
     let aggregate_state: AggregateState<()> = AggregateState::default();

examples/locking_strategies/main.rs

@@ -6,18 +6,21 @@ use tokio::sync::Mutex;
 use uuid::Uuid;
 
 use esrs::postgres::{PgStore, PgStoreBuilder};
-use esrs::{AggregateManager, AggregateState, EventStore};
+use esrs::{AggregateManager, AggregateManagerError, AggregateState, EventStore};
 
-use crate::common::{new_pool, BasicAggregate, BasicCommand, BasicError, BasicEvent};
+use crate::common::{new_pool, BasicAggregate, BasicCommand, BasicEvent};
 
 #[path = "../common/lib.rs"]
 mod common;
 
-type Agg = AggregateManager<BasicAggregate>;
+type Agg = AggregateManager<PgStore<BasicAggregate>>;
 
 /// Increment the value behind this `aggregate_id` as soon as atomical access can be obtained.
 /// The lock can be obtained even if there are current optimistic accesses! Avoid mixing the two strategies when writing.
-pub async fn increment_atomically(manager: Agg, aggregate_id: Uuid) -> Result<(), BasicError> {
+pub async fn increment_atomically(
+    manager: Agg,
+    aggregate_id: Uuid,
+) -> Result<(), AggregateManagerError<PgStore<BasicAggregate>>> {
     let aggregate_state = manager.lock_and_load(aggregate_id).await?.unwrap_or_default();
     manager
         .handle_command(
@@ -32,7 +35,10 @@ pub async fn increment_atomically(manager: Agg, aggregate_id: Uuid) -> Result<()
 
 /// Increment the value behind this `aggregate_id` with an optimistic locking strategy.
 /// Optimistic access ignores any current active lock! Avoid mixing the two strategies when writing.
-pub async fn increment_optimistically(manager: Agg, aggregate_id: Uuid) -> Result<(), BasicError> {
+pub async fn increment_optimistically(
+    manager: Agg,
+    aggregate_id: Uuid,
+) -> Result<(), AggregateManagerError<PgStore<BasicAggregate>>> {
     // Every optimistic access can take place concurrently...
     let aggregate_state = manager.load(aggregate_id).await?.unwrap_or_default();
     // ...and events are persisted in non-deterministic order.
@@ -51,7 +57,10 @@ pub async fn increment_optimistically(manager: Agg, aggregate_id: Uuid) -> Resul
 /// Load the aggregate state for read-only purposes, preventing others (that use locking) from modifying it.
 /// Avoid using atomic reads if writes are optimistic, as the state would be modified anyway!
 /// If writes are atomic, it is perfectly fine to use a mixture of atomic and optimistic reads.
-pub async fn with_atomic_read(manager: Agg, aggregate_id: Uuid) -> Result<(), BasicError> {
+pub async fn with_atomic_read(
+    manager: Agg,
+    aggregate_id: Uuid,
+) -> Result<(), AggregateManagerError<PgStore<BasicAggregate>>> {
     let mut aggregate_state = manager.lock_and_load(aggregate_id).await?.unwrap_or_default();
     // No one else (employing locking!) can read or modify the state just loaded here,
     // ensuring this really is the *latest* aggregate state.
@@ -65,7 +74,10 @@ pub async fn with_atomic_read(manager: Agg, aggregate_id: Uuid) -> Result<(), Ba
 /// Load the aggregate state for read-only purposes, optimistically assuming nothing is modifying it.
 /// If writes are atomic, it is perfectly fine to use a mixture of atomic and optimistic reads.
 /// Otherwise, optimistic reads are allowed: beware there are no guarantees the state loaded is actually the latest.
-pub async fn with_optimistic_read(manager: Agg, aggregate_id: Uuid) -> Result<(), BasicError> {
+pub async fn with_optimistic_read(
+    manager: Agg,
+    aggregate_id: Uuid,
+) -> Result<(), AggregateManagerError<PgStore<BasicAggregate>>> {
     // Read the state now, ignoring any explicit locking...
     let mut aggregate_state = manager.load(aggregate_id).await?.unwrap_or_default();
     // ...but nothing prevents something else from updating the data in the store in the meanwhile,
@@ -91,7 +103,7 @@ async fn main() {
     };
     let _ = store.persist(&mut aggregate_state, vec![event]).await.unwrap();
 
-    let manager: AggregateManager<BasicAggregate> = AggregateManager::new(store.clone());
+    let manager: AggregateManager<PgStore<BasicAggregate>> = AggregateManager::new(store.clone());
 
     // It is possible to load the aggregate state multiple times.
     let _state_1 = manager.load(aggregate_id).await.unwrap().unwrap();

examples/multi_aggregate_rebuild/main.rs

@@ -2,11 +2,11 @@ use futures::StreamExt;
 use sqlx::{PgConnection, Pool, Postgres, Transaction};
 use uuid::Uuid;
 
-use esrs::postgres::{PgStore, PgStoreBuilder};
+use esrs::postgres::{PgStore, PgStoreBuilder, PgStoreError};
 use esrs::{AggregateManager, AggregateState, ReplayableEventHandler, StoreEvent, TransactionalEventHandler};
 
 use crate::common::{
-    new_pool, AggregateA, AggregateB, CommandA, CommandB, CommonError, EventA, EventB, SharedEventHandler, SharedView,
+    new_pool, AggregateA, AggregateB, CommandA, CommandB, EventA, EventB, SharedEventHandler, SharedView,
 };
 use crate::transactional_event_handler::SharedTransactionalEventHandler;
 
@@ -49,17 +49,20 @@ async fn rebuild_multi_aggregate(
     let event_handlers_a: Vec<Box<dyn ReplayableEventHandler<AggregateA>>> = vec![event_handler.clone()];
     let event_handlers_b: Vec<Box<dyn ReplayableEventHandler<AggregateB>>> = vec![event_handler.clone()];
 
-    let transactional_event_handlers_a: Vec<Box<dyn TransactionalEventHandler<AggregateA, PgConnection>>> =
-        vec![transactional_event_handler.clone()];
-    let transactional_event_handlers_b: Vec<Box<dyn TransactionalEventHandler<AggregateB, PgConnection>>> =
-        vec![transactional_event_handler.clone()];
+    let transactional_event_handlers_a: Vec<
+        Box<dyn TransactionalEventHandler<AggregateA, PgStoreError, PgConnection>>,
+    > = vec![transactional_event_handler.clone()];
+
+    let transactional_event_handlers_b: Vec<
+        Box<dyn TransactionalEventHandler<AggregateB, PgStoreError, PgConnection>>,
+    > = vec![transactional_event_handler.clone()];
 
     let mut events_a = store_a.stream_events(&pool);
     let mut events_b = store_b.stream_events(&pool);
 
     // Fetch first element of both the tables
-    let mut event_a_opt: Option<Result<StoreEvent<EventA>, CommonError>> = events_a.next().await;
-    let mut event_b_opt: Option<Result<StoreEvent<EventB>, CommonError>> = events_b.next().await;
+    let mut event_a_opt: Option<Result<StoreEvent<EventA>, PgStoreError>> = events_a.next().await;
+    let mut event_b_opt: Option<Result<StoreEvent<EventB>, PgStoreError>> = events_b.next().await;
 
     // At this point it's possible to open a transaction
     let mut transaction: Transaction<Postgres> = pool.begin().await.unwrap();
@@ -152,7 +155,7 @@ async fn setup(shared_id: Uuid, pool: Pool<Postgres>, view: SharedView, transact
         .await
         .unwrap();
 
-    let manager: AggregateManager<AggregateA> = AggregateManager::new(pg_store_a);
+    let manager: AggregateManager<PgStore<AggregateA>> = AggregateManager::new(pg_store_a);
     manager
         .handle_command(AggregateState::default(), CommandA { v: 10, shared_id })
         .await
@@ -170,7 +173,7 @@ async fn setup(shared_id: Uuid, pool: Pool<Postgres>, view: SharedView, transact
         .await
         .unwrap();
 
-    let manager: AggregateManager<AggregateB> = AggregateManager::new(pg_store_b);
+    let manager: AggregateManager<PgStore<AggregateB>> = AggregateManager::new(pg_store_b);
     manager
         .handle_command(AggregateState::default(), CommandB { v: 7, shared_id })
         .await

examples/multi_aggregate_rebuild/transactional_event_handler.rs

@@ -1,18 +1,19 @@
 use async_trait::async_trait;
 use sqlx::PgConnection;
 
+use esrs::postgres::PgStoreError;
 use esrs::{StoreEvent, TransactionalEventHandler};
 
-use crate::common::{AggregateA, AggregateB, CommonError, EventA, EventB, SharedView, UpsertSharedView};
+use crate::common::{AggregateA, AggregateB, EventA, EventB, SharedView, UpsertSharedView};
 
 #[derive(Clone)]
 pub struct SharedTransactionalEventHandler {
     pub view: SharedView,
 }
 
 #[async_trait]
-impl TransactionalEventHandler<AggregateA, PgConnection> for SharedTransactionalEventHandler {
-    async fn handle(&self, event: &StoreEvent<EventA>, executor: &mut PgConnection) -> Result<(), CommonError> {
+impl TransactionalEventHandler<AggregateA, PgStoreError, PgConnection> for SharedTransactionalEventHandler {
+    async fn handle(&self, event: &StoreEvent<EventA>, executor: &mut PgConnection) -> Result<(), PgStoreError> {
         Ok(self
             .view
             .upsert(
@@ -28,8 +29,8 @@ impl TransactionalEventHandler<AggregateA, PgConnection> for SharedTransactional
 }
 
 #[async_trait]
-impl TransactionalEventHandler<AggregateB, PgConnection> for SharedTransactionalEventHandler {
-    async fn handle(&self, event: &StoreEvent<EventB>, executor: &mut PgConnection) -> Result<(), CommonError> {
+impl TransactionalEventHandler<AggregateB, PgStoreError, PgConnection> for SharedTransactionalEventHandler {
+    async fn handle(&self, event: &StoreEvent<EventB>, executor: &mut PgConnection) -> Result<(), PgStoreError> {
         Ok(self
             .view
             .upsert(

examples/rebuilder/main.rs

@@ -81,7 +81,7 @@ async fn setup(aggregate_id: Uuid, view: BasicView, transactional_view: BasicVie
         .await
         .unwrap();
 
-    let manager: AggregateManager<BasicAggregate> = AggregateManager::new(pg_store);
+    let manager: AggregateManager<PgStore<BasicAggregate>> = AggregateManager::new(pg_store);
     manager
         .handle_command(
             AggregateState::with_id(aggregate_id),

examples/rebuilder/transactional_event_handler.rs

@@ -1,9 +1,10 @@
 use async_trait::async_trait;
 use sqlx::PgConnection;
 
+use esrs::postgres::PgStoreError;
 use esrs::{StoreEvent, TransactionalEventHandler};
 
-use crate::common::{BasicAggregate, BasicError, BasicEvent, BasicView};
+use crate::common::{BasicAggregate, BasicEvent, BasicView};
 
 /// The `BasicTransactionalEventHandlerV1` and `BasicTransactionalEventHandlerV1` exists in this
 /// example just for the sake of showing how a single transactional event handler, in this case called
@@ -19,8 +20,8 @@ pub struct BasicTransactionalEventHandlerV2 {
 }
 
 #[async_trait]
-impl TransactionalEventHandler<BasicAggregate, PgConnection> for BasicTransactionalEventHandlerV1 {
-    async fn handle(&self, event: &StoreEvent<BasicEvent>, transaction: &mut PgConnection) -> Result<(), BasicError> {
+impl TransactionalEventHandler<BasicAggregate, PgStoreError, PgConnection> for BasicTransactionalEventHandlerV1 {
+    async fn handle(&self, event: &StoreEvent<BasicEvent>, transaction: &mut PgConnection) -> Result<(), PgStoreError> {
         Ok(self
             .view
             .upsert(
@@ -33,8 +34,8 @@ impl TransactionalEventHandler<BasicAggregate, PgConnection> for BasicTransactio
 }
 
 #[async_trait]
-impl TransactionalEventHandler<BasicAggregate, PgConnection> for BasicTransactionalEventHandlerV2 {
-    async fn handle(&self, event: &StoreEvent<BasicEvent>, transaction: &mut PgConnection) -> Result<(), BasicError> {
+impl TransactionalEventHandler<BasicAggregate, PgStoreError, PgConnection> for BasicTransactionalEventHandlerV2 {
+    async fn handle(&self, event: &StoreEvent<BasicEvent>, transaction: &mut PgConnection) -> Result<(), PgStoreError> {
         Ok(self
             .view
             .upsert(

examples/saga/main.rs

@@ -29,9 +29,9 @@ async fn main() {
         side_effect_mutex: side_effect_mutex.clone(),
     };
 
-    store.add_event_handler(saga_event_handler);
+    store.add_event_handler(saga_event_handler).await;
 
-    let manager: AggregateManager<SagaAggregate> = AggregateManager::new(store.clone());
+    let manager: AggregateManager<PgStore<SagaAggregate>> = AggregateManager::new(store.clone());
 
     let state: AggregateState<()> = AggregateState::default();
     let id: Uuid = *state.id();

examples/transactional_view/main.rs

@@ -31,7 +31,7 @@ async fn main() {
         content: content.to_string(),
     };
 
-    let manager: AggregateManager<BasicAggregate> = AggregateManager::new(store.clone());
+    let manager: AggregateManager<PgStore<BasicAggregate>> = AggregateManager::new(store.clone());
 
     let result = manager.handle_command(state, command).await;
 

examples/transactional_view/transactional_event_handler.rs

@@ -1,20 +1,23 @@
 use async_trait::async_trait;
 use sqlx::PgConnection;
 
+use esrs::postgres::PgStoreError;
 use esrs::{StoreEvent, TransactionalEventHandler};
 
-use crate::common::{BasicAggregate, BasicError, BasicEvent, BasicView};
+use crate::common::{BasicAggregate, BasicEvent, BasicView};
 
 pub struct BasicTransactionalEventHandler {
     pub view: BasicView,
 }
 
 #[async_trait]
-impl TransactionalEventHandler<BasicAggregate, PgConnection> for BasicTransactionalEventHandler {
-    async fn handle(&self, event: &StoreEvent<BasicEvent>, transaction: &mut PgConnection) -> Result<(), BasicError> {
+impl TransactionalEventHandler<BasicAggregate, PgStoreError, PgConnection> for BasicTransactionalEventHandler {
+    async fn handle(&self, event: &StoreEvent<BasicEvent>, transaction: &mut PgConnection) -> Result<(), PgStoreError> {
         // This is to show that event is rollbacked
         if event.payload.content.eq("error") {
-            return Err(BasicError::Custom("Event contains `error` string".to_string()));
+            return Err(PgStoreError::Custom(Box::new(BasicEventHandlerError::Custom(
+                "Event contains `error` string".to_string(),
+            ))));
         }
 
         let result = self
@@ -30,3 +33,9 @@ impl TransactionalEventHandler<BasicAggregate, PgConnection> for BasicTransactio
         }
     }
 }
+
+#[derive(thiserror::Error, Debug)]
+pub enum BasicEventHandlerError {
+    #[error("{0}")]
+    Custom(String),
+}

src/esrs/aggregate.rs

@@ -19,15 +19,15 @@ pub trait Aggregate {
 
     /// Internal aggregate state. This will be wrapped in [`AggregateState`] and could be used to validate
     /// commands.
-    type State: Default + Clone + Send + Sync;
+    type State: Default;
 
     /// A command is an action that the caller can execute over an aggregate in order to let it emit
     /// an event.
-    type Command: Send;
+    type Command;
 
     /// An event represents a fact that took place in the domain. They are the source of truth;
     /// your current state is derived from the events.
-    type Event: Send + Sync;
+    type Event;
 
     /// This associated type is used to get domain errors while handling a command.
     type Error;