feat: abstract NotificationBus for swappable dispatch

seriouscoderone · claude · seriouscoderone · commit 495e70595c9c · 2026-02-18T18:59:23.000-07:00
Replace the concrete HashMap-based NotificationBus with a trait-based
dispatch architecture. NotificationBus is now a Clone-able wrapper around
Arc&lt;dyn NotificationDispatch&gt;, enabling alternative notification backends
(e.g. SQS for serverless) without adding generic type parameters anywhere
in the codebase.

- Add NotificationDispatch trait with dispatch() and register_observer()
- Extract current HashMap logic into InProcessDispatch (RwLock + OnceLock)
- Add NotificationBus::from_dispatch() for custom implementations
- Change register_observer from &amp;mut self to &amp;self across all processors
- Add RwLockingError variant to keri-core Error enum

Co-Authored-By: Claude Opus 4.6 &lt;noreply@anthropic.com&gt;
diff --git a/components/controller/src/known_events.rs b/components/controller/src/known_events.rs
@@ -65,7 +65,7 @@ impl KnownEvents {
         let oobi_manager = OobiManager::new(event_database.clone());
 
         let (
-            mut notification_bus,
+            notification_bus,
             (
                 _out_of_order_escrow,
                 _partially_signed_escrow,
diff --git a/components/watcher/src/watcher/watcher_data.rs b/components/watcher/src/watcher/watcher_data.rs
@@ -99,7 +99,7 @@ impl WatcherData {
 
         let oobi_manager = OobiManager::new(events_db.clone());
 
-        let (mut notification_bus, _) = default_escrow_bus(events_db.clone(), escrow_config);
+        let (notification_bus, _) = default_escrow_bus(events_db.clone(), escrow_config);
         let reply_escrow = Arc::new(ReplyEscrow::new(events_db.clone()));
         notification_bus.register_observer(
             reply_escrow.clone(),
diff --git a/components/witness/src/witness.rs b/components/witness/src/witness.rs
@@ -171,7 +171,7 @@ impl Witness {
 
         let events_db =
             Arc::new(RedbDatabase::new(&events_database_path).map_err(|_| Error::DbError)?);
-        let mut witness_processor = WitnessProcessor::new(events_db.clone(), escrow_config);
+        let witness_processor = WitnessProcessor::new(events_db.clone(), escrow_config);
         let event_storage = Arc::new(EventStorage::new_redb(events_db.clone()));
 
         let receipt_generator = Arc::new(WitnessReceiptGenerator::new(
diff --git a/components/witness/src/witness_processor.rs b/components/witness/src/witness_processor.rs
@@ -22,7 +22,7 @@ pub struct WitnessProcessor {
 impl Processor for WitnessProcessor {
     type Database = RedbDatabase;
     fn register_observer(
-        &mut self,
+        &self,
         observer: Arc<dyn Notifier + Send + Sync>,
         notifications: &[JustNotification],
     ) -> Result<(), Error> {
@@ -62,7 +62,7 @@ impl Default for WitnessEscrowConfig {
 
 impl WitnessProcessor {
     pub fn new(redb: Arc<RedbDatabase>, escrow_config: WitnessEscrowConfig) -> Self {
-        let mut bus = NotificationBus::new();
+        let bus = NotificationBus::new();
         let partially_signed_escrow = Arc::new(PartiallySignedEscrow::new(
             redb.clone(),
             escrow_config.partially_signed_timeout,
diff --git a/keriox_core/src/error/mod.rs b/keriox_core/src/error/mod.rs
@@ -79,6 +79,9 @@ pub enum Error {
     #[error("mutex is poisoned")]
     MutexPoisoned,
 
+    #[error("RwLock poisoned")]
+    RwLockingError,
+
     #[error("Incorrect event digest")]
     IncorrectDigest,
 
diff --git a/keriox_core/src/processor/basic_processor.rs b/keriox_core/src/processor/basic_processor.rs
@@ -18,7 +18,7 @@ pub struct BasicProcessor<D: EventDatabase>(EventProcessor<D>);
 impl<D: EventDatabase + 'static> Processor for BasicProcessor<D> {
     type Database = D;
     fn register_observer(
-        &mut self,
+        &self,
         observer: Arc<dyn Notifier + Send + Sync>,
         notification: &[JustNotification],
     ) -> Result<(), Error> {
diff --git a/keriox_core/src/processor/escrow/maybe_out_of_order_escrow.rs b/keriox_core/src/processor/escrow/maybe_out_of_order_escrow.rs
@@ -111,7 +111,7 @@ fn test_out_of_order() -> Result<(), Error> {
         let events_db_path = NamedTempFile::new().unwrap();
         let redb = RedbDatabase::new(events_db_path.path()).unwrap();
         let events_db = Arc::new(redb);
-        let mut processor = BasicProcessor::new(events_db.clone(), None);
+        let processor = BasicProcessor::new(events_db.clone(), None);
 
         // Register out of order escrow, to save and reprocess out of order events
         let new_ooo = Arc::new(MaybeOutOfOrderEscrow::new(
diff --git a/keriox_core/src/processor/escrow/mod.rs b/keriox_core/src/processor/escrow/mod.rs
@@ -54,7 +54,7 @@ pub fn default_escrow_bus<D>(
 where
     D: EventDatabase + EscrowCreator + Sync + Send + 'static,
 {
-    let mut bus = NotificationBus::new();
+    let bus = NotificationBus::new();
 
     // Register out of order escrow, to save and reprocess out of order events
     let ooo_escrow = Arc::new(MaybeOutOfOrderEscrow::new(
diff --git a/keriox_core/src/processor/escrow/partially_signed_escrow.rs b/keriox_core/src/processor/escrow/partially_signed_escrow.rs
@@ -199,7 +199,7 @@ mod tests {
             let path = witness_root.path();
             let events_db_path = NamedTempFile::new().unwrap();
             let events_db = Arc::new(RedbDatabase::new(events_db_path.path()).unwrap());
-            let mut processor = BasicProcessor::new(events_db.clone(), None);
+            let processor = BasicProcessor::new(events_db.clone(), None);
 
             // Register out of order escrow, to save and reprocess out of order events
             let ooo_escrow = Arc::new(MaybeOutOfOrderEscrow::new(
@@ -271,7 +271,7 @@ mod tests {
             std::fs::create_dir_all(path).unwrap();
             let events_db_path = NamedTempFile::new().unwrap();
             let events_db = Arc::new(RedbDatabase::new(events_db_path.path()).unwrap());
-            let mut processor = BasicProcessor::new(events_db.clone(), None);
+            let processor = BasicProcessor::new(events_db.clone(), None);
 
             // Register partially signed escrow, to save and reprocess partially signed events
             let ps_escrow = Arc::new(PartiallySignedEscrow::new(
diff --git a/keriox_core/src/processor/escrow/partially_witnessed_escrow.rs b/keriox_core/src/processor/escrow/partially_witnessed_escrow.rs
@@ -383,7 +383,7 @@ mod tests {
         let redb = RedbDatabase::new(events_db_path.path()).unwrap();
         let log_db = redb.log_db.clone();
         let events_db = Arc::new(redb);
-        let mut event_processor = BasicProcessor::new(events_db.clone(), None);
+        let event_processor = BasicProcessor::new(events_db.clone(), None);
         let event_storage = EventStorage::new(Arc::clone(&events_db));
 
         // Register not fully witnessed escrow, to save and reprocess events
@@ -528,7 +528,7 @@ mod tests {
         let redb = RedbDatabase::new(events_db_path.path()).unwrap();
         let log_db = redb.log_db.clone();
         let events_db = Arc::new(redb);
-        let mut event_processor = BasicProcessor::new(events_db.clone(), None);
+        let event_processor = BasicProcessor::new(events_db.clone(), None);
         let event_storage = EventStorage::new(Arc::clone(&events_db));
 
         // Register not fully witnessed escrow, to save and reprocess events
@@ -671,7 +671,7 @@ mod tests {
         let redb = RedbDatabase::new(events_db_path.path()).unwrap();
         let log_db = redb.log_db.clone();
         let events_db = Arc::new(redb);
-        let mut event_processor = BasicProcessor::new(events_db.clone(), None);
+        let event_processor = BasicProcessor::new(events_db.clone(), None);
         let event_storage = EventStorage::new(Arc::clone(&events_db));
 
         // Register not fully witnessed escrow, to save and reprocess events
@@ -775,7 +775,7 @@ mod tests {
         let redb = RedbDatabase::new(events_db_path.path()).unwrap();
         let log_db = redb.log_db.clone();
         let events_db = Arc::new(redb);
-        let mut event_processor = BasicProcessor::new(events_db.clone(), None);
+        let event_processor = BasicProcessor::new(events_db.clone(), None);
         let event_storage = EventStorage::new(Arc::clone(&events_db));
 
         // Register not fully witnessed escrow, to save and reprocess events
diff --git a/keriox_core/src/processor/escrow/reply_escrow.rs b/keriox_core/src/processor/escrow/reply_escrow.rs
@@ -196,7 +196,7 @@ mod tests {
         fs::create_dir_all(root.path()).unwrap();
         let events_db_path = NamedTempFile::new().unwrap();
         let events_db = Arc::new(RedbDatabase::new(events_db_path.path()).unwrap());
-        let mut event_processor = BasicProcessor::new(events_db.clone(), None);
+        let event_processor = BasicProcessor::new(events_db.clone(), None);
         let rpy_escrow = Arc::new(ReplyEscrow::new(events_db.clone()));
         event_processor.register_observer(
             rpy_escrow.clone(),
diff --git a/keriox_core/src/processor/escrow_tests.rs b/keriox_core/src/processor/escrow_tests.rs
@@ -50,7 +50,7 @@ fn test_out_of_order_cleanup() -> Result<(), Error> {
         let events_db_path = NamedTempFile::new().unwrap();
         let redb = RedbDatabase::new(events_db_path.path()).unwrap();
         let events_db = Arc::new(redb);
-        let mut processor = BasicProcessor::new(events_db.clone(), None);
+        let processor = BasicProcessor::new(events_db.clone(), None);
 
         // Register out of order escrow, to save and reprocess out of order events
         let ooo_escrow = Arc::new(MaybeOutOfOrderEscrow::new(
@@ -145,7 +145,7 @@ fn test_partially_sign_escrow_cleanup() -> Result<(), Error> {
         std::fs::create_dir_all(path).unwrap();
         let events_db_path = NamedTempFile::new().unwrap();
         let events_db = Arc::new(RedbDatabase::new(events_db_path.path()).unwrap());
-        let mut processor = BasicProcessor::new(events_db.clone(), None);
+        let processor = BasicProcessor::new(events_db.clone(), None);
 
         // Register partially signed escrow, to save and reprocess partially signed events
         let ps_escrow = Arc::new(PartiallySignedEscrow::new(
@@ -230,7 +230,7 @@ pub fn test_partially_witnessed_escrow_cleanup() -> Result<(), Error> {
     let redb = RedbDatabase::new(events_db_path.path()).unwrap();
     let log_db = redb.log_db.clone();
     let events_db = Arc::new(redb);
-    let mut event_processor = BasicProcessor::new(events_db.clone(), None);
+    let event_processor = BasicProcessor::new(events_db.clone(), None);
     let event_storage = EventStorage::new(Arc::clone(&events_db));
     // Register not fully witnessed escrow, to save and reprocess events
     let partially_witnessed_escrow = Arc::new(PartiallyWitnessedEscrow::new(
diff --git a/keriox_core/src/processor/mod.rs b/keriox_core/src/processor/mod.rs
@@ -38,7 +38,7 @@ pub trait Processor {
     fn process_op_reply(&self, reply: &SignedReply) -> Result<(), Error>;
 
     fn register_observer(
-        &mut self,
+        &self,
         observer: Arc<dyn Notifier + Send + Sync>,
         notifications: &[JustNotification],
     ) -> Result<(), Error>;
@@ -92,7 +92,7 @@ impl<D: EventDatabase + 'static> EventProcessor<D> {
     }
 
     pub fn register_observer(
-        &mut self,
+        &self,
         observer: Arc<dyn Notifier + Send + Sync>,
         notifications: Vec<JustNotification>,
     ) -> Result<(), Error> {
diff --git a/keriox_core/src/processor/notification.rs b/keriox_core/src/processor/notification.rs
@@ -1,4 +1,7 @@
-use std::{collections::HashMap, sync::Arc};
+use std::{
+    collections::HashMap,
+    sync::{Arc, OnceLock, RwLock},
+};
 
 #[cfg(feature = "query")]
 use crate::query::reply_event::SignedReply;
@@ -10,36 +13,108 @@ use crate::{
     },
 };
 
+/// Internal dispatch strategy — the swappable part.
+/// Implement this trait to change how notifications are delivered
+/// (e.g. in-process HashMap, SQS queue, etc.).
+pub trait NotificationDispatch: Send + Sync {
+    fn dispatch(&self, notification: &Notification) -> Result<(), Error>;
+    fn register_observer(
+        &self,
+        observer: Arc<dyn Notifier + Send + Sync>,
+        notifications: Vec<JustNotification>,
+    ) -> Result<(), Error>;
+}
+
+/// In-process dispatch: preserves the original HashMap-based behavior.
+/// Uses `RwLock` for interior mutability so `register_observer` takes `&self`.
+struct InProcessDispatch {
+    observers: RwLock<HashMap<JustNotification, Vec<Arc<dyn Notifier + Send + Sync>>>>,
+    /// Back-reference to the owning `NotificationBus` so we can pass it
+    /// to `Notifier::notify()` callbacks.
+    bus: OnceLock<NotificationBus>,
+}
+
+impl InProcessDispatch {
+    fn new() -> Self {
+        Self {
+            observers: RwLock::new(HashMap::new()),
+            bus: OnceLock::new(),
+        }
+    }
+}
+
+impl NotificationDispatch for InProcessDispatch {
+    fn dispatch(&self, notification: &Notification) -> Result<(), Error> {
+        let observers = self
+            .observers
+            .read()
+            .map_err(|_| Error::RwLockingError)?;
+        let bus = self.bus.get().ok_or_else(|| {
+            Error::SemanticError("InProcessDispatch: bus back-reference not set".into())
+        })?;
+        if let Some(obs) = observers.get(&notification.into()) {
+            for esc in obs.iter() {
+                esc.notify(notification, bus)?;
+            }
+        }
+        Ok(())
+    }
+
+    fn register_observer(
+        &self,
+        observer: Arc<dyn Notifier + Send + Sync>,
+        notifications: Vec<JustNotification>,
+    ) -> Result<(), Error> {
+        let mut observers = self
+            .observers
+            .write()
+            .map_err(|_| Error::RwLockingError)?;
+        for notification in notifications {
+            observers
+                .entry(notification)
+                .or_default()
+                .push(observer.clone());
+        }
+        Ok(())
+    }
+}
+
+/// Clone-able notification bus that delegates to an internal dispatch strategy.
+#[derive(Clone)]
 pub struct NotificationBus {
-    observers: HashMap<JustNotification, Vec<Arc<dyn Notifier + Send + Sync>>>,
+    inner: Arc<dyn NotificationDispatch>,
 }
 
 impl NotificationBus {
+    /// Create a new bus with the default in-process dispatch.
     pub fn new() -> Self {
-        Self {
-            observers: HashMap::new(),
-        }
+        let dispatch = Arc::new(InProcessDispatch::new());
+        let bus = Self {
+            inner: dispatch.clone(),
+        };
+        // Set the back-reference so InProcessDispatch can pass &NotificationBus
+        // to Notifier::notify() callbacks.
+        let _ = dispatch.bus.set(bus.clone());
+        bus
     }
+
+    /// Create a bus backed by a custom dispatch implementation.
+    pub fn from_dispatch(dispatch: Arc<dyn NotificationDispatch>) -> Self {
+        Self { inner: dispatch }
+    }
+
     pub fn register_observer(
-        &mut self,
+        &self,
         escrow: Arc<dyn Notifier + Send + Sync>,
         notification: Vec<JustNotification>,
     ) {
-        notification.into_iter().for_each(|notification| {
-            self.observers
-                .entry(notification)
-                .or_default()
-                .push(escrow.clone());
-        });
+        // register_observer on InProcessDispatch should not fail in practice,
+        // but if it does we silently ignore to preserve the existing API signature.
+        let _ = self.inner.register_observer(escrow, notification);
     }
 
     pub fn notify(&self, notification: &Notification) -> Result<(), Error> {
-        if let Some(obs) = self.observers.get(&notification.into()) {
-            for esc in obs.iter() {
-                esc.notify(notification, self)?;
-            }
-        };
-        Ok(())
+        self.inner.dispatch(notification)
     }
 }
 
diff --git a/support/teliox/src/processor/escrow/missing_issuer.rs b/support/teliox/src/processor/escrow/missing_issuer.rs
@@ -184,7 +184,7 @@ mod tests {
         // Setup issuer key event log. Without ixn events tel event's can't be validated.
         let keri_root = Builder::new().prefix("test-db").tempfile().unwrap();
         let keri_db = Arc::new(RedbDatabase::new(keri_root.path()).unwrap());
-        let mut keri_processor = BasicProcessor::new(keri_db.clone(), None);
+        let keri_processor = BasicProcessor::new(keri_db.clone(), None);
         let keri_storage = Arc::new(EventStorage::new(keri_db.clone()));
 
         let issuer_kel = r#"{"v":"KERI10JSON00012b_","t":"icp","d":"EETk5xW-rl2TgHTTXr8m5kGXiC30m3gMgsYcBAjOE9eI","i":"EETk5xW-rl2TgHTTXr8m5kGXiC30m3gMgsYcBAjOE9eI","s":"0","kt":"1","k":["DHdoiqT1iac2HI6-HfCYcc01Piz2FTTPvZDFt6vADioD"],"nt":"1","n":["EH8IzIWeQFiUr3rr2dh8xAiW9Akwl6EooDt8iduQYyq_"],"bt":"0","b":[],"c":[],"a":[]}-AABAABvFFeXb9uW2G16o3C9xJZvY3a_utMPxd4NIUcGWRTqykMO1NzKwjsA_AQrOEwgO5jselWHREcK6vcAxRfv6-QC{"v":"KERI10JSON00013a_","t":"ixn","d":"EMOzEVoFjbkS3ZS5JtmJO4LeZ4gydbr8iXNrEQAt1OR2","i":"EETk5xW-rl2TgHTTXr8m5kGXiC30m3gMgsYcBAjOE9eI","s":"1","p":"EETk5xW-rl2TgHTTXr8m5kGXiC30m3gMgsYcBAjOE9eI","a":[{"i":"EF3TVac5quxrbLGLKAHF21laISjMgjYQAIg3OsTen969","s":"0","d":"ENIKpuUkjM-1K2Sv_TZwF_k8FTVkefAgy8sIpiFp0uWh"}]}-AABAACvrSS_EZUMKQ6Ax8FaB_Sf99O0y6MmfoRDBKMphVWWtuCOlFQm6N0XrTwtYxO3pO0AEZkJ1vzu52-RDK-w3YAN{"v":"KERI10JSON00013a_","t":"ixn","d":"EDvnfU2yMZUXEy9D_22YOkeSZOq6YG9zfItawvx3GR_6","i":"EETk5xW-rl2TgHTTXr8m5kGXiC30m3gMgsYcBAjOE9eI","s":"2","p":"EMOzEVoFjbkS3ZS5JtmJO4LeZ4gydbr8iXNrEQAt1OR2","a":[{"i":"EC8Oej-3HAUpBY_kxzBK3B-0RV9j4dXw1H0NRKxJg7g-","s":"0","d":"EDBM1ys50vEJxRzvBjTOrmOhokELjVtozXy3ZbJ8-KFk"}]}-AABAAABtEQ7SoGt2IcZBMX0GaEaMqGdMsrGpj1fABDKgE5dA7s7AGXTkWrZjzA4GXkGXuOspi6upqBhpxr6d5ySeKQH"#;

Original file line number	Diff line number	Diff line change
`@@ -65,7 +65,7 @@ impl KnownEvents {`
`65`	`65`	`let oobi_manager = OobiManager::new(event_database.clone());`
`66`	`66`
`67`	`67`	`let (`
`68`		`- mut notification_bus,`
	`68`	`+ notification_bus,`
`69`	`69`	`(`
`70`	`70`	`_out_of_order_escrow,`
`71`	`71`	`_partially_signed_escrow,`
Original file line number	Diff line number	Diff line change
`@@ -54,7 +54,7 @@ pub fn default_escrow_bus<D>(`
`54`	`54`	`where`
`55`	`55`	`D: EventDatabase + EscrowCreator + Sync + Send + 'static,`
`56`	`56`	`{`
`57`		`- let mut bus = NotificationBus::new();`
	`57`	`+ let bus = NotificationBus::new();`
`58`	`58`
`59`	`59`	`// Register out of order escrow, to save and reprocess out of order events`
`60`	`60`	`let ooo_escrow = Arc::new(MaybeOutOfOrderEscrow::new(`