Add server-side event filtering by event name (#4)

* Add server-side event filtering by event name

* Fix u128 parsing for some events

* add serializable timestamp, strongly typed event names

* clean up channel names, ws client management

---------

Co-authored-by: 0xflashboy <jake@monad.foundation>

Author: iamvukasin

Cargo.lock

@@ -8,6 +8,7 @@ use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
 use execution_events_example::event_listener;
 use execution_events_example::event_listener::EventData;
 use execution_events_example::server;
+use tracing::warn;
 
 #[derive(Debug, Parser)]
 #[command(name = "eventwatch", about, long_about = None)]
@@ -41,16 +42,23 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 
     // Create a channel for communication between event listener thread and server
     // Use large buffer to handle bursts
-    let (tx, rx) = mpsc::channel::<EventData>(100_000);
+    let (event_sender, event_receiver) = mpsc::channel::<EventData>(100_000);
 
     // Spawn the event listener thread
-    let _listener_handle = event_listener::start_event_listener(event_ring_path, tx);
+    let listener_handle = event_listener::run_event_listener(event_ring_path, event_sender);
 
     // Parse server address
     let addr: SocketAddr = server_addr.parse()?;
 
-    // Run the WebSocket server
-    server::run_websocket_server(addr, rx).await?;
+    // Run both tasks and exit when either completes
+    tokio::select! {
+        result = server::run_websocket_server(addr, event_receiver) => {
+            warn!("WebSocket server stopped: {:?}", result);
+        }
+        _ = tokio::task::spawn_blocking(move || listener_handle.join()) => {
+            warn!("Event listener thread stopped");
+        }
+    }
 
     Ok(())
 }

backend/src/bin/backend.rs

@@ -1,6 +1,8 @@
 use clap::Parser;
+use execution_events_example::event_filter::ClientMessage;
+use execution_events_example::event_listener::EventName;
 use execution_events_example::serializable_event::SerializableEventData;
-use futures_util::StreamExt;
+use futures_util::{SinkExt, StreamExt};
 use tokio_tungstenite::{connect_async, tungstenite::Message};
 
 use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
@@ -14,6 +16,11 @@ struct Cli {
     #[arg(short, long, default_value = "ws://127.0.0.1:3000")]
     url: String,
 
+    /// Filter events by type (comma-separated).
+    /// If not specified, all events are received.
+    #[arg(short, long, value_delimiter = ',')]
+    events: Option<Vec<String>>,
+
     #[arg(short, long, default_value = "false")]
     dump_events: bool,
 }
@@ -36,7 +43,34 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let (ws_stream, _) = connect_async(&cli.url).await?;
     info!("Connected!");
 
-    let (_, mut read) = ws_stream.split();
+    let (mut write, mut read) = ws_stream.split();
+
+    // Parse event names from strings to EventName enum
+    let event_strings = cli.events.clone().unwrap_or_default();
+    let events: Vec<EventName> = if event_strings.is_empty() {
+        Vec::new()
+    } else {
+        event_strings
+            .iter()
+            .map(|s| {
+                serde_json::from_value(serde_json::Value::String(s.clone()))
+                    .map_err(|_| format!("Invalid event name: {}", s))
+            })
+            .collect::<Result<Vec<_>, _>>()?
+    };
+
+    // Send subscription message
+    let subscribe_msg = ClientMessage::Subscribe {
+        events: events.clone(),
+    };
+    let subscribe_json = serde_json::to_string(&subscribe_msg)?;
+    write.send(Message::Text(subscribe_json)).await?;
+
+    if events.is_empty() {
+        info!("Subscribed to all events");
+    } else {
+        info!("Subscribed to events: {:?}", events);
+    }
 
     // Read messages from the server
     let mut events_per_sec_interval = tokio::time::interval(tokio::time::Duration::from_secs(1));

backend/src/bin/client.rs

@@ -1,4 +1,10 @@
 // Re-export modules from lib/ subdirectory
+pub mod timestamp {
+    pub use super::lib::timestamp::*;
+}
+pub mod event_filter {
+    pub use super::lib::event_filter::*;
+}
 pub mod event_listener {
     pub use super::lib::event_listener::*;
 }
@@ -11,7 +17,9 @@ pub mod server {
 
 // Internal module containing implementations
 mod lib {
+    pub mod event_filter;
     pub mod event_listener;
     pub mod serializable_event;
     pub mod server;
+    pub mod timestamp;
 }

backend/src/lib.rs

@@ -0,0 +1,41 @@
+use serde::{Deserialize, Serialize};
+use std::collections::HashSet;
+
+use super::event_listener::EventName;
+
+/// Message sent by client to subscribe to specific event types
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(tag = "type")]
+pub enum ClientMessage {
+    /// Subscribe to specific event types. Empty list means subscribe to all events.
+    /// If any event name fails to deserialize, the connection will be rejected.
+    #[serde(rename = "subscribe")]
+    Subscribe { events: Vec<EventName> },
+}
+
+/// Filter for event types
+#[derive(Clone, Debug, Default)]
+pub struct EventFilter {
+    /// Set of event types to include. If empty, all events pass through.
+    event_types: HashSet<EventName>,
+}
+
+impl EventFilter {
+    /// Create a filter from a list of event names
+    pub fn from_event_names(events: Vec<EventName>) -> Self {
+        Self {
+            event_types: events.into_iter().collect(),
+        }
+    }
+
+    /// Check if an event name matches the filter using direct enum comparison
+    /// Returns true if the filter is empty (accept all) or if the event name is in the filter
+    pub fn matches(&self, event_name: &EventName) -> bool {
+        self.event_types.is_empty() || self.event_types.contains(event_name)
+    }
+
+    /// Check if filter accepts all events
+    pub fn accepts_all(&self) -> bool {
+        self.event_types.is_empty()
+    }
+}

backend/src/lib/event_filter.rs

@@ -1,6 +1,4 @@
-use std::{ffi::CStr, time::Duration};
-
-use chrono::{DateTime, Local, TimeZone};
+use chrono::{DateTime, Local};
 use lazy_static::lazy_static;
 use monad_event_ring::{
     DecodedEventRing, EventDescriptor, EventDescriptorInfo, EventNextResult, EventPayloadResult,
@@ -11,8 +9,13 @@ use monad_exec_events::{
     ffi::{g_monad_exec_event_metadata, MONAD_EXEC_EVENT_COUNT},
     ExecEventDecoder, ExecEventDescriptorExt, ExecEventRing, ExecSnapshotEventRing,
 };
+use serde::{Deserialize, Serialize};
+use std::{ffi::CStr, time::Duration};
 use tracing::{debug, error, info, warn};
 
+use super::timestamp::get_unix_time_ns;
+use super::timestamp::NanoTimestamp;
+
 lazy_static! {
     static ref EXEC_EVENT_NAMES: [&'static str; MONAD_EXEC_EVENT_COUNT] =
         std::array::from_fn(|event_type| unsafe {
@@ -22,11 +25,105 @@ lazy_static! {
         });
 }
 
+/// Type-safe enum for event names based on ExecEvent variants
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
+#[serde(rename_all = "PascalCase")]
+pub enum EventName {
+    RecordError,
+    BlockStart,
+    BlockReject,
+    BlockPerfEvmEnter,
+    BlockPerfEvmExit,
+    BlockEnd,
+    BlockQC,
+    BlockFinalized,
+    BlockVerified,
+    TxnHeaderStart,
+    TxnAccessListEntry,
+    TxnAuthListEntry,
+    TxnHeaderEnd,
+    TxnReject,
+    TxnPerfEvmEnter,
+    TxnPerfEvmExit,
+    TxnEvmOutput,
+    TxnLog,
+    TxnCallFrame,
+    TxnEnd,
+    AccountAccessListHeader,
+    AccountAccess,
+    StorageAccess,
+    EvmError,
+}
+
+impl EventName {
+    /// Convert EventName to string representation
+    pub fn as_str(&self) -> &'static str {
+        match self {
+            EventName::RecordError => "RecordError",
+            EventName::BlockStart => "BlockStart",
+            EventName::BlockReject => "BlockReject",
+            EventName::BlockPerfEvmEnter => "BlockPerfEvmEnter",
+            EventName::BlockPerfEvmExit => "BlockPerfEvmExit",
+            EventName::BlockEnd => "BlockEnd",
+            EventName::BlockQC => "BlockQC",
+            EventName::BlockFinalized => "BlockFinalized",
+            EventName::BlockVerified => "BlockVerified",
+            EventName::TxnHeaderStart => "TxnHeaderStart",
+            EventName::TxnAccessListEntry => "TxnAccessListEntry",
+            EventName::TxnAuthListEntry => "TxnAuthListEntry",
+            EventName::TxnHeaderEnd => "TxnHeaderEnd",
+            EventName::TxnReject => "TxnReject",
+            EventName::TxnPerfEvmEnter => "TxnPerfEvmEnter",
+            EventName::TxnPerfEvmExit => "TxnPerfEvmExit",
+            EventName::TxnEvmOutput => "TxnEvmOutput",
+            EventName::TxnLog => "TxnLog",
+            EventName::TxnCallFrame => "TxnCallFrame",
+            EventName::TxnEnd => "TxnEnd",
+            EventName::AccountAccessListHeader => "AccountAccessListHeader",
+            EventName::AccountAccess => "AccountAccess",
+            EventName::StorageAccess => "StorageAccess",
+            EventName::EvmError => "EvmError",
+        }
+    }
+
+    pub fn from_str(s: &str) -> Option<Self> {
+        match s {
+            "RECORD_ERROR" => Some(EventName::RecordError),
+            "BLOCK_START" => Some(EventName::BlockStart),
+            "BLOCK_REJECT" => Some(EventName::BlockReject),
+            "BLOCK_PERF_EVM_ENTER" => Some(EventName::BlockPerfEvmEnter),
+            "BLOCK_PERF_EVM_EXIT" => Some(EventName::BlockPerfEvmExit),
+            "BLOCK_END" => Some(EventName::BlockEnd),
+            "BLOCK_QC" => Some(EventName::BlockQC),
+            "BLOCK_FINALIZED" => Some(EventName::BlockFinalized),
+            "BLOCK_VERIFIED" => Some(EventName::BlockVerified),
+            "TXN_HEADER_START" => Some(EventName::TxnHeaderStart),
+            "TXN_ACCESS_LIST_ENTRY" => Some(EventName::TxnAccessListEntry),
+            "TXN_AUTH_LIST_ENTRY" => Some(EventName::TxnAuthListEntry),
+            "TXN_HEADER_END" => Some(EventName::TxnHeaderEnd),
+            "TXN_REJECT" => Some(EventName::TxnReject),
+            "TXN_PERF_EVM_ENTER" => Some(EventName::TxnPerfEvmEnter),
+            "TXN_PERF_EVM_EXIT" => Some(EventName::TxnPerfEvmExit),
+            "TXN_EVM_OUTPUT" => Some(EventName::TxnEvmOutput),
+            "TXN_LOG" => Some(EventName::TxnLog),
+            "TXN_CALL_FRAME" => Some(EventName::TxnCallFrame),
+            "TXN_END" => Some(EventName::TxnEnd),
+            "ACCOUNT_ACCESS_LIST_HEADER" => Some(EventName::AccountAccessListHeader),
+            "ACCOUNT_ACCESS" => Some(EventName::AccountAccess),
+            "STORAGE_ACCESS" => Some(EventName::StorageAccess),
+            "EVM_ERROR" => Some(EventName::EvmError),
+            _ => {
+                warn!("Unknown event name: {}", s);
+                None
+            }
+        }
+    }
+}
+
 #[derive(Debug, Clone)]
 pub struct EventData {
-    pub timestamp: String,
-    pub event_name: String,
-    pub event_type: u16,
+    pub timestamp_ns: NanoTimestamp,
+    pub event_name: EventName,
     pub seqno: u64,
     pub block_number: Option<u64>,
     pub txn_idx: Option<usize>,
@@ -55,16 +152,14 @@ fn event_to_data(event: &EventDescriptor<ExecEventDecoder>) -> Option<EventData>
     let EventDescriptorInfo {
         seqno,
         event_type,
-        record_epoch_nanos,
+        record_epoch_nanos: _,
         flow_info,
     } = event.info();
 
-    let timestamp = Local
-        .timestamp_nanos(record_epoch_nanos as i64)
-        .format("%H:%M:%S.%9f")
-        .to_string();
+    let timestamp_ns = get_unix_time_ns();
 
-    let event_name = EXEC_EVENT_NAMES[event_type as usize].to_string();
+    // Convert event_type to EventName enum for type safety
+    let event_name = EventName::from_str(EXEC_EVENT_NAMES[event_type as usize])?;
 
     // Get block number if present
     let block_number = if flow_info.block_seqno != 0 {
@@ -86,19 +181,18 @@ fn event_to_data(event: &EventDescriptor<ExecEventDecoder>) -> Option<EventData>
     };
 
     Some(EventData {
-        timestamp,
+        timestamp_ns,
         event_name,
-        event_type,
         seqno,
         block_number,
         txn_idx,
         payload,
     })
 }
 
-pub fn start_event_listener(
+pub fn run_event_listener(
     event_ring_path: EventRingPath,
-    tx: tokio::sync::mpsc::Sender<EventData>,
+    event_sender: tokio::sync::mpsc::Sender<EventData>,
 ) -> std::thread::JoinHandle<()> {
     std::thread::spawn(move || {
         info!("Starting event listener thread");
@@ -196,7 +290,7 @@ pub fn start_event_listener(
                     if let Some(event_data) = event_to_data(&event) {
                         // Send to channel; if receiver is dropped, exit thread
                         // Use blocking_send since we're in a blocking thread
-                        if tx.blocking_send(event_data).is_err() {
+                        if event_sender.blocking_send(event_data).is_err() {
                             warn!("Channel receiver dropped, exiting listener thread");
                             return;
                         }