fix(omnievent): replace protobuf-based uuidv5 with cbor for stable ids (#44)

Author: azixus

Cargo.lock

@@ -60,6 +60,7 @@ prost-build = { version = "0.13" }
 prost-types = { version = "0.13" }
 serde = { version = "1.0", features = ["derive"], default-features = false }
 serde_json = "1.0"
+serde_cbor = "0.11"
 
 # storage
 sqlx = { version = "0.8" }

Cargo.toml

@@ -31,6 +31,7 @@ prost.workspace = true
 prost-types.workspace = true
 serde.workspace = true
 serde_json.workspace = true
+serde_cbor.workspace = true
 
 # storage
 sqlx = { workspace = true, features = ["runtime-tokio"], optional = true }
@@ -44,6 +45,7 @@ uuid = { version = "1.0", features = ["v5"] }
 
 [dev-dependencies]
 alloy = { workspace = true, features = ["default", "provider-ws", "provider-anvil-node"] }
+hex.workspace = true
 
 [build-dependencies]
 tonic-build.workspace = true

omnievent/Cargo.toml

@@ -13,7 +13,10 @@ use crate::event_manager::listener::{
     EventListener, EventListenerHandle, EventReceiverHandleError,
 };
 use crate::proto_types::EventOccurrenceFilter;
-use crate::types::{EventFieldData, EventId, EventOccurrence, ParsedRegisterNewEventRequest};
+use crate::types::{
+    EventFieldData, EventId, EventOccurrence, NewRegisteredEventSpecError,
+    ParsedRegisterNewEventRequest, RegisteredEventSpec,
+};
 use alloy::primitives::Address;
 use alloy::rpc::types::Log;
 use futures_util::stream::SelectAll;
@@ -78,6 +81,9 @@ pub(crate) enum EventManagerError {
 
     #[error("filter error")]
     Filter(#[from] FilterError),
+
+    #[error("failed to convert event registration into spec")]
+    EventRegistrationIntoSpec(#[from] NewRegisteredEventSpecError),
 }
 
 // export other event_manager's module errors
@@ -146,11 +152,12 @@ where
         &self,
         req: ParsedRegisterNewEventRequest,
     ) -> Result<EventId, EventManagerError> {
-        let event_id = req.id;
-        let chain_id = req.chain_id;
-        let address = req.address;
-        let event_name = req.event_name.clone();
-        self.internal_register_ethereum_event(req)
+        let event_spec = RegisteredEventSpec::try_from(req)?;
+        let event_id = event_spec.id;
+        let chain_id = event_spec.chain_id;
+        let address = event_spec.address;
+        let event_name = event_spec.event_name.clone();
+        self.internal_register_ethereum_event(event_spec)
             .instrument(tracing::info_span!("register_ethereum_event", %event_id, %chain_id, %address, %event_name))
             .await
     }
@@ -300,7 +307,7 @@ pub(crate) mod tests {
     pub(crate) mod test_contracts {
         use crate::event_manager::tests::test_contracts::EventEmitter::EventEmitterInstance;
         use crate::proto_types::{BlockSafety, EventField, RegisterNewEventRequest};
-        use crate::types::{ParsedRegisterNewEventRequest, RegisteredEvent};
+        use crate::types::{ParsedRegisterNewEventRequest, RegisteredEventSpec};
         use alloy::network::Network;
         use alloy::providers::Provider;
 
@@ -330,12 +337,12 @@ pub(crate) mod tests {
 
         pub(crate) async fn get_string_registered_event<P, N>(
             instance: &EventEmitterInstance<P, N>,
-        ) -> RegisteredEvent
+        ) -> RegisteredEventSpec
         where
             P: Provider<N>,
             N: Network,
         {
-            RegisteredEvent::try_from_req(get_string_register_req(instance).await).unwrap()
+            RegisteredEventSpec::try_from(get_string_register_req(instance).await).unwrap()
         }
 
         pub(crate) async fn get_string_register_req<P, N>(

omnievent/src/event_manager.rs

@@ -1,6 +1,6 @@
 //! Database used to store events and their occurrences.
 
-use crate::types::{EventId, EventOccurrence, RegisteredEvent};
+use crate::types::{EventId, EventOccurrence, RegisteredEventSpec};
 
 pub mod in_memory;
 
@@ -13,7 +13,7 @@ pub trait EventsDatabase {
     /// Store an event in the database.
     fn store_event(
         &self,
-        event: RegisteredEvent,
+        event: RegisteredEventSpec,
     ) -> impl Future<Output = Result<(), Self::Error>> + Send;
 
     /// Store an event occurrence in the database.

omnievent/src/event_manager/db.rs

@@ -1,14 +1,14 @@
 //! Non-persistent in-memory database.
 
 use crate::event_manager::db::EventsDatabase;
-use crate::types::{EventId, EventOccurrence, RegisteredEvent};
+use crate::types::{EventId, EventOccurrence, RegisteredEventSpec};
 use itertools::Itertools;
 use std::collections::HashMap;
 use std::sync::Arc;
 
 pub struct InMemoryDatabaseEntry {
     #[allow(unused)]
-    registered_event: RegisteredEvent,
+    registered_event: RegisteredEventSpec,
     occurrences: Vec<EventOccurrence>,
 }
 
@@ -27,7 +27,7 @@ pub(crate) enum InMemoryDatabaseError {
 impl EventsDatabase for InMemoryDatabase {
     type Error = InMemoryDatabaseError;
 
-    async fn store_event(&self, registered_event: RegisteredEvent) -> Result<(), Self::Error> {
+    async fn store_event(&self, registered_event: RegisteredEventSpec) -> Result<(), Self::Error> {
         let mut db = self.0.write().await;
         db.0.insert(
             registered_event.id,

omnievent/src/event_manager/db/in_memory.rs

@@ -1,7 +1,7 @@
 //! A sqlite-based [`EventsDatabase`]
 
 use crate::event_manager::db::EventsDatabase;
-use crate::types::{BlockInfo, EventId, EventOccurrence, RegisteredEvent};
+use crate::types::{BlockInfo, EventId, EventOccurrence, RegisteredEventSpec};
 use alloy::primitives::Address;
 use chrono::{DateTime, Utc};
 use sqlx::{FromRow, QueryBuilder, Row, Sqlite, SqlitePool};
@@ -84,7 +84,7 @@ impl SqliteEventDatabase {
 impl EventsDatabase for SqliteEventDatabase {
     type Error = SqliteEventDatabaseError;
 
-    async fn store_event(&self, event: RegisteredEvent) -> Result<(), Self::Error> {
+    async fn store_event(&self, event: RegisteredEventSpec) -> Result<(), Self::Error> {
         let event_id = Uuid::from(event.id);
         let event_chain_id = event.chain_id.to_string();
         let event_address = event.address.to_vec();
@@ -248,7 +248,7 @@ mod tests {
     use crate::event_manager::db::EventsDatabase;
     use crate::event_manager::db::sql::sqlite::SqliteEventDatabase;
     use crate::proto_types::BlockSafety;
-    use crate::types::{BlockInfo, EventId, EventOccurrence, RegisteredEvent};
+    use crate::types::{BlockInfo, EventId, EventOccurrence, RegisteredEventSpec};
     use alloy::primitives::{Address, LogData};
 
     #[tokio::test]
@@ -272,7 +272,7 @@ mod tests {
 
         let res = db
             .store_event(
-                RegisteredEvent::try_new(
+                RegisteredEventSpec::try_new(
                     EventId::new(b"test_event"),
                     0u64,
                     Address::default(),
@@ -325,7 +325,7 @@ mod tests {
 
         let event_id = EventId::new(b"test_event");
         db.store_event(
-            RegisteredEvent::try_new(
+            RegisteredEventSpec::try_new(
                 event_id,
                 0u64,
                 Address::default(),
@@ -367,7 +367,7 @@ mod tests {
 
         let event_id = EventId::new(b"test_event");
         db.store_event(
-            RegisteredEvent::try_new(
+            RegisteredEventSpec::try_new(
                 event_id,
                 0u64,
                 Address::default(),

omnievent/src/event_manager/db/sql/sqlite.rs

@@ -1,7 +1,7 @@
 //! Manages event streams and sends back decoded event occurrences.
 
 use crate::event_manager::DecodedEvent;
-use crate::types::{EventFieldData, EventId, RegisteredEvent};
+use crate::types::{EventFieldData, EventId, RegisteredEventSpec};
 use alloy::rpc::types::Log;
 use futures_util::StreamExt;
 use futures_util::stream::{BoxStream, SelectAll};
@@ -83,13 +83,13 @@ impl EventListener {
 pub(crate) type LogStreamWithId = BoxStream<'static, (EventId, Log)>;
 
 pub(crate) struct InternalEventStreamRegistration {
-    event: RegisteredEvent,
+    event: RegisteredEventSpec,
     stream: LogStreamWithId,
 }
 
 impl InternalEventStreamRegistration {
     /// Register a new event stream.
-    pub fn new(event: RegisteredEvent, stream: LogStreamWithId) -> Self {
+    pub fn new(event: RegisteredEventSpec, stream: LogStreamWithId) -> Self {
         Self { event, stream }
     }
 }
@@ -217,11 +217,11 @@ pub enum EventLogDecodeError {
 /// Try to decode a log with a given event specification and return the decoded fields.
 ///
 /// # Panics
-/// Panics if the provided [`RegisteredEvent`] is inconsistent, e.g., the sol_event is inconsistent
+/// Panics if the provided [`RegisteredEventSpec`] is inconsistent, e.g., the sol_event is inconsistent
 /// with the vector of fields.
 fn decode_log(
     log: &Log,
-    event: &RegisteredEvent,
+    event: &RegisteredEventSpec,
 ) -> Result<Vec<EventFieldData>, EventLogDecodeError> {
     let decoded = event.sol_event.decode_log_data(log.data()).map_err(|e| {
         tracing::error!(error = ?e, ?log, ?event, "Failed to decode log with given spec");
@@ -309,7 +309,7 @@ mod tests {
             ..Default::default()
         };
 
-        let event = RegisteredEvent::try_new(
+        let event = RegisteredEventSpec::try_new(
             EventId::new(b"RandomnessRequested -- test"),
             137,
             address,

omnievent/src/event_manager/listener.rs

@@ -3,9 +3,7 @@
 use crate::event_manager::db::EventsDatabase;
 use crate::event_manager::listener::InternalEventStreamRegistration;
 use crate::event_manager::{EventManager, EventManagerError, RegisteredEventEntry};
-use crate::types::{
-    EventId, NewRegisteredEventError, ParsedRegisterNewEventRequest, RegisteredEvent,
-};
+use crate::types::{EventId, NewRegisteredEventSpecError, RegisteredEventSpec};
 use alloy::network::{Ethereum, Network};
 use alloy::providers::Provider;
 use alloy::pubsub::SubscriptionStream;
@@ -20,7 +18,7 @@ where
 {
     pub(super) async fn internal_register_ethereum_event(
         &self,
-        req: ParsedRegisterNewEventRequest,
+        event_spec: RegisteredEventSpec,
     ) -> Result<EventId, EventManagerError> {
         tracing::debug!("Registering new event");
 
@@ -30,22 +28,21 @@ where
         };
 
         // Do nothing if the event is already registered
-        let event_id = req.id;
+        let event_id = event_spec.id;
         if self.active_events_map.read().await.contains_key(&event_id) {
             tracing::debug!("Event already registered");
             return Ok(event_id);
         }
 
-        let (event, stream) =
-            create_stream_and_spec::<_, Ethereum>(req, &self.multi_provider).await?;
+        let stream = create_stream::<_, Ethereum>(&event_spec, &self.multi_provider).await?;
 
         let reg = InternalEventStreamRegistration::new(
-            event.clone(),
+            event_spec.clone(),
             stream.map(move |l| (event_id, l)).boxed(), // boxing :( but we need type erasure due to the closure
         );
 
         // Save the event in the database
-        if let Err(e) = self.events_db.store_event(event).await {
+        if let Err(e) = self.events_db.store_event(event_spec).await {
             tracing::error!(event = ?event_id, error = ?e, "Failed to store event in database");
             Err(EventManagerError::Database(e.into()))?
         }
@@ -77,7 +74,7 @@ where
 #[derive(thiserror::Error, Debug)]
 pub(crate) enum CreateStreamError {
     #[error("failed to create event")]
-    FailedToCreateEvent(#[from] NewRegisteredEventError),
+    FailedToCreateEvent(#[from] NewRegisteredEventSpecError),
 
     #[error("unsupported chain")]
     UnsupportedChain,
@@ -88,25 +85,23 @@ pub(crate) enum CreateStreamError {
     ),
 }
 
-pub(crate) async fn create_stream_and_spec<MP, N>(
-    req: ParsedRegisterNewEventRequest,
+pub(crate) async fn create_stream<MP, N>(
+    spec: &RegisteredEventSpec,
     multi_provider: &MP,
-) -> Result<(RegisteredEvent, SubscriptionStream<Log>), CreateStreamError>
+) -> Result<SubscriptionStream<Log>, CreateStreamError>
 where
     MP: MultiChainProvider<u64>,
     N: Network,
 {
-    let registered_event = RegisteredEvent::try_from_req(req)?;
-
     // Obtain a provider for the specified chainid and network
-    let Some(provider) = multi_provider.get_provider::<N>(&registered_event.chain_id) else {
+    let Some(provider) = multi_provider.get_provider::<N>(&spec.chain_id) else {
         Err(CreateStreamError::UnsupportedChain)?
     };
 
     // Create a new subscription for the specified event
     let stream = provider
-        .subscribe_logs(&Filter::from(&registered_event))
+        .subscribe_logs(&Filter::from(spec))
         .await?
         .into_stream();
-    Ok((registered_event, stream))
+    Ok(stream)
 }

omnievent/src/event_manager/register.rs

@@ -16,6 +16,25 @@ mod events {
         }
     }
 
+    impl serde::Serialize for BlockSafety {
+        fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+        where
+            S: serde::Serializer,
+        {
+            serializer.serialize_str(self.as_str_name())
+        }
+    }
+
+    impl<'de> serde::Deserialize<'de> for BlockSafety {
+        fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+        where
+            D: serde::Deserializer<'de>,
+        {
+            let s = String::deserialize(deserializer)?;
+            Self::from_str_name(&s).ok_or(serde::de::Error::unknown_variant(&s, &[]))
+        }
+    }
+
     impl From<alloy::dyn_abi::DynSolValue> for event_data::Value {
         fn from(value: alloy::dyn_abi::DynSolValue) -> Self {
             match value {