fix(saga): require Arc<S> in SagaAdapter to guarantee shared state across bus subscriptions

SagaAdapter previously accepted S by value, meaning saga.clone() could
produce independent copies with diverging state for callers whose saga
struct does not internally Arc its own mutable fields.

Changes:
- Add blanket SubscriberId impl for Arc<S> in epoch_core::subscriber_id
- Add blanket Saga<ED> impl for Arc<S> in epoch_core::saga (requires
  S::EventType: Sync for async future Send bounds)
- Change SagaAdapter.saga field and new() parameter from S to Arc<S>
- Update saga_adapter_integration_tests to use Arc::new(saga)

Author: Istar-Eldritch

epoch_core/src/saga.rs

@@ -176,6 +176,49 @@ where
     }
 }
 
+/// Blanket [`Saga`] implementation for `Arc<S>` where `S: Saga<ED>`.
+///
+/// This allows an `Arc`-wrapped saga to be used directly with [`SagaHandler`] and
+/// [`SagaAdapter`], enabling shared ownership across multiple bus subscriptions:
+///
+/// ```ignore
+/// let saga = Arc::new(MySaga::new(...));
+/// native_bus.subscribe(SagaHandler::new(saga.clone())).await?;
+/// foreign_bus.subscribe(SagaAdapter::new(saga.clone(), "saga:my:foreign", |e| ...)).await?;
+/// ```
+#[async_trait]
+impl<ED, S> Saga<ED> for Arc<S>
+where
+    ED: EventData + Send + Sync + 'static,
+    S: Saga<ED> + Send + Sync,
+    S::EventType: Sync,
+{
+    type State = S::State;
+    type StateStore = S::StateStore;
+    type SagaError = S::SagaError;
+    type EventType = S::EventType;
+
+    fn get_state_store(&self) -> Self::StateStore {
+        (**self).get_state_store()
+    }
+
+    async fn handle_event(
+        &self,
+        state: Self::State,
+        event: &Event<Self::EventType>,
+    ) -> Result<Option<Self::State>, Self::SagaError> {
+        (**self).handle_event(state, event).await
+    }
+
+    fn get_id_from_event(&self, event: &Event<Self::EventType>) -> Uuid {
+        (**self).get_id_from_event(event)
+    }
+
+    fn priority(&self) -> u8 {
+        (**self).priority()
+    }
+}
+
 /// A wrapper type that provides an [`EventObserver`] implementation for [`Saga`] types.
 ///
 /// Since Rust doesn't allow multiple blanket implementations of the same trait, and
@@ -300,7 +343,7 @@ where
     TargetEvent: EventData + Send + Sync + 'static,
     F: Fn(&SourceEvent) -> Option<TargetEvent> + Send + Sync,
 {
-    saga: S,
+    saga: Arc<S>,
     subscriber_id: String,
     converter: F,
     _marker: PhantomData<fn(SourceEvent) -> TargetEvent>,
@@ -315,15 +358,17 @@ where
 {
     /// Creates a new `SagaAdapter`.
     ///
-    /// * `saga` - The saga to wrap, shared via `Arc` so it can be subscribed
-    ///   to multiple buses.
+    /// * `saga` - `Arc`-wrapped saga shared across bus subscriptions. Both
+    ///   [`SagaHandler`] and `SagaAdapter` accept `Arc<S>` (via the blanket
+    ///   `Saga` impl for `Arc<S>`), so a single `Arc::new(saga)` can back
+    ///   any number of adapters without cloning internal state.
     /// * `subscriber_id` - Unique identifier for *this specific subscription*.
     ///   Used by the event bus to track an independent checkpoint per bus.
     ///   Convention: `"saga:<saga-name>:<source-bus-name>"`.
     /// * `converter` - Closure mapping the foreign bus's event variants into
     ///   the saga's native parent event type. Return `None` to skip events
     ///   the saga does not care about.
-    pub fn new(saga: S, subscriber_id: impl Into<String>, converter: F) -> Self {
+    pub fn new(saga: Arc<S>, subscriber_id: impl Into<String>, converter: F) -> Self {
         Self {
             saga,
             subscriber_id: subscriber_id.into(),
@@ -341,7 +386,7 @@ where
 impl<S, SourceEvent, TargetEvent, F> crate::SubscriberId
     for SagaAdapter<S, SourceEvent, TargetEvent, F>
 where
-    S: Saga<TargetEvent>,
+    S: Saga<TargetEvent> + Send + Sync,
     SourceEvent: EventData + Send + Sync + 'static,
     TargetEvent: EventData + Send + Sync + 'static,
     F: Fn(&SourceEvent) -> Option<TargetEvent> + Send + Sync,
@@ -384,7 +429,7 @@ where
             causation_id: event.causation_id,
             correlation_id: event.correlation_id,
         };
-        self.saga.process_event(&target_event).await?;
+        self.saga.as_ref().process_event(&target_event).await?;
         Ok(())
     }
 

epoch_core/src/subscriber_id.rs

@@ -83,3 +83,15 @@ pub trait SubscriberId {
     /// The ID must be stable across deployments and restarts.
     fn subscriber_id(&self) -> &str;
 }
+
+/// Delegates [`SubscriberId`] through an `Arc` so `Arc<S>` can be used
+/// wherever `S: SubscriberId` is required (e.g. with [`SagaHandler`] and
+/// [`SagaAdapter`]).
+///
+/// [`SagaHandler`]: crate::saga::SagaHandler
+/// [`SagaAdapter`]: crate::saga::SagaAdapter
+impl<S: SubscriberId + ?Sized> SubscriberId for std::sync::Arc<S> {
+    fn subscriber_id(&self) -> &str {
+        (**self).subscriber_id()
+    }
+}

epoch_pg/tests/saga_adapter_integration_tests.rs

@@ -229,7 +229,7 @@ async fn saga_adapter_receives_events_from_foreign_bus() {
     let target_stream = Uuid::new_v4();
     let source_stream = Uuid::new_v4();
 
-    let saga = CounterSaga::new(&native_sub_id, saga_id);
+    let saga = Arc::new(CounterSaga::new(&native_sub_id, saga_id));
 
     target_bus
         .subscribe(SagaHandler::new(saga.clone()))
@@ -293,7 +293,7 @@ async fn saga_adapter_preserves_event_metadata_through_conversion() {
     let saga_id = Uuid::new_v4();
     let stream_id = Uuid::new_v4();
 
-    let saga = CounterSaga::new(format!("saga:meta:{}", tag), saga_id);
+    let saga = Arc::new(CounterSaga::new(format!("saga:meta:{}", tag), saga_id));
     source_bus
         .subscribe(SagaAdapter::new(
             saga.clone(),
@@ -350,7 +350,7 @@ async fn saga_adapter_advances_independent_checkpoints() {
     let target_stream = Uuid::new_v4();
     let source_stream = Uuid::new_v4();
 
-    let saga = CounterSaga::new(&native_sub_id, saga_id);
+    let saga = Arc::new(CounterSaga::new(&native_sub_id, saga_id));
     target_bus
         .subscribe(SagaHandler::new(saga.clone()))
         .await
@@ -425,7 +425,7 @@ async fn saga_adapter_resumes_from_checkpoint_after_restart() {
 
     // First run.
     {
-        let saga = CounterSaga::new(format!("saga:restart:{}", tag), saga_id);
+        let saga = Arc::new(CounterSaga::new(format!("saga:restart:{}", tag), saga_id));
         source_bus
             .subscribe(SagaAdapter::new(
                 saga.clone(),
@@ -463,7 +463,7 @@ async fn saga_adapter_resumes_from_checkpoint_after_restart() {
     let source_store2 = PgEventStore::new(pool.clone(), source_bus2.clone());
     source_bus2.setup_trigger().await.unwrap();
 
-    let saga2 = CounterSaga::new(format!("saga:restart:{}", tag), saga_id);
+    let saga2 = Arc::new(CounterSaga::new(format!("saga:restart:{}", tag), saga_id));
     source_bus2
         .subscribe(SagaAdapter::new(
             saga2.clone(),