added lossy option to events_observers

rdeioris · rdeioris · commit 99ab769e257a · 2025-02-13T11:12:12.000+01:00
diff --git a/stackslib/src/config/mod.rs b/stackslib/src/config/mod.rs
@@ -915,6 +915,7 @@ impl Config {
                         endpoint: observer.endpoint,
                         events_keys,
                         timeout_ms: observer.timeout_ms.unwrap_or(1_000),
+                        lossy: observer.lossy.unwrap_or(false),
                     });
                 }
                 observers
@@ -928,6 +929,7 @@ impl Config {
                 endpoint: val,
                 events_keys: vec![EventKeyType::AnyEvent],
                 timeout_ms: 1_000,
+                lossy: false,
             });
         };
 
@@ -2824,13 +2826,15 @@ pub struct EventObserverConfigFile {
     pub endpoint: String,
     pub events_keys: Vec<String>,
     pub timeout_ms: Option<u64>,
+    pub lossy: Option<bool>,
 }
 
 #[derive(Clone, Default, Debug, Hash, PartialEq, Eq, PartialOrd)]
 pub struct EventObserverConfig {
     pub endpoint: String,
     pub events_keys: Vec<EventKeyType>,
     pub timeout_ms: u64,
+    pub lossy: bool,
 }
 
 #[derive(Clone, Debug, Hash, PartialEq, Eq, PartialOrd)]
diff --git a/testnet/stacks-node/src/event_dispatcher.rs b/testnet/stacks-node/src/event_dispatcher.rs
@@ -90,6 +90,8 @@ struct EventObserver {
     endpoint: String,
     /// Timeout for sending events to this observer
     timeout: Duration,
+    /// Lossy observers do not retry on error
+    lossy: bool,
 }
 
 struct ReceiptPayloadInfo<'a> {
@@ -439,7 +441,7 @@ impl EventObserver {
 
         for (id, url, payload, timeout_ms) in pending_payloads {
             let timeout = Duration::from_millis(timeout_ms);
-            Self::send_payload_directly(&payload, &url, timeout);
+            Self::send_payload_directly(&payload, &url, timeout, false);
 
             #[cfg(test)]
             if TEST_EVENT_OBSERVER_SKIP_RETRY.get() {
@@ -456,7 +458,12 @@ impl EventObserver {
         }
     }
 
-    fn send_payload_directly(payload: &serde_json::Value, full_url: &str, timeout: Duration) {
+    fn send_payload_directly(
+        payload: &serde_json::Value,
+        full_url: &str,
+        timeout: Duration,
+        lossy: bool,
+    ) {
         debug!(
             "Event dispatcher: Sending payload"; "url" => %full_url, "payload" => ?payload
         );
@@ -506,6 +513,11 @@ impl EventObserver {
                 }
             }
 
+            if lossy {
+                warn!("Observer is configured in lossy mode: skipping retry of payload");
+                return;
+            }
+
             #[cfg(test)]
             if TEST_EVENT_OBSERVER_SKIP_RETRY.get() {
                 warn!("Fault injection: skipping retry of payload");
@@ -522,7 +534,7 @@ impl EventObserver {
         }
     }
 
-    fn new(working_dir: Option<PathBuf>, endpoint: String, timeout: Duration) -> Self {
+    fn new(working_dir: Option<PathBuf>, endpoint: String, timeout: Duration, lossy: bool) -> Self {
         let db_path = if let Some(mut db_path) = working_dir {
             db_path.push("event_observers.sqlite");
 
@@ -541,6 +553,7 @@ impl EventObserver {
             db_path,
             endpoint,
             timeout,
+            lossy,
         }
     }
 
@@ -555,18 +568,23 @@ impl EventObserver {
         };
         let full_url = format!("http://{url_str}");
 
-        if let Some(db_path) = &self.db_path {
-            let conn =
-                Connection::open(db_path).expect("Failed to open database for event observer");
+        // if the observer is in lossy mode quickly send the payload without checking for the db
+        if self.lossy {
+            Self::send_payload_directly(payload, &full_url, self.timeout, true);
+        } else {
+            if let Some(db_path) = &self.db_path {
+                let conn =
+                    Connection::open(db_path).expect("Failed to open database for event observer");
 
-            // Insert the new payload into the database
-            Self::insert_payload_with_retry(&conn, &full_url, payload, self.timeout);
+                // Insert the new payload into the database
+                Self::insert_payload_with_retry(&conn, &full_url, payload, self.timeout);
 
-            // Process all pending payloads
-            Self::process_pending_payloads(&conn);
-        } else {
-            // No database, just send the payload
-            Self::send_payload_directly(payload, &full_url, self.timeout);
+                // Process all pending payloads
+                Self::process_pending_payloads(&conn);
+            } else {
+                // No database, just send the payload
+                Self::send_payload_directly(payload, &full_url, self.timeout, false);
+            }
         }
     }
 
@@ -1666,6 +1684,7 @@ impl EventDispatcher {
             Some(working_dir),
             conf.endpoint.clone(),
             Duration::from_millis(conf.timeout_ms),
+            conf.lossy,
         );
 
         let observer_index = self.registered_observers.len() as u16;
@@ -1770,7 +1789,8 @@ mod test {
 
     #[test]
     fn build_block_processed_event() {
-        let observer = EventObserver::new(None, "nowhere".to_string(), Duration::from_secs(3));
+        let observer =
+            EventObserver::new(None, "nowhere".to_string(), Duration::from_secs(3), false);
 
         let filtered_events = vec![];
         let block = StacksBlock::genesis_block();
@@ -1830,7 +1850,8 @@ mod test {
 
     #[test]
     fn test_block_processed_event_nakamoto() {
-        let observer = EventObserver::new(None, "nowhere".to_string(), Duration::from_secs(3));
+        let observer =
+            EventObserver::new(None, "nowhere".to_string(), Duration::from_secs(3), false);
 
         let filtered_events = vec![];
         let mut block_header = NakamotoBlockHeader::empty();
@@ -2087,7 +2108,8 @@ mod test {
         let endpoint = "http://example.com".to_string();
         let timeout = Duration::from_secs(5);
 
-        let observer = EventObserver::new(Some(working_dir.clone()), endpoint.clone(), timeout);
+        let observer =
+            EventObserver::new(Some(working_dir.clone()), endpoint.clone(), timeout, false);
 
         // Verify fields
         assert_eq!(observer.endpoint, endpoint);
@@ -2104,7 +2126,7 @@ mod test {
         let endpoint = "http://example.com".to_string();
         let timeout = Duration::from_secs(5);
 
-        let observer = EventObserver::new(None, endpoint.clone(), timeout);
+        let observer = EventObserver::new(None, endpoint.clone(), timeout, false);
 
         // Verify fields
         assert_eq!(observer.endpoint, endpoint);
@@ -2133,7 +2155,7 @@ mod test {
         let endpoint = server.url().strip_prefix("http://").unwrap().to_string();
         let timeout = Duration::from_secs(5);
 
-        let observer = EventObserver::new(Some(working_dir), endpoint, timeout);
+        let observer = EventObserver::new(Some(working_dir), endpoint, timeout, false);
 
         TEST_EVENT_OBSERVER_SKIP_RETRY.set(false);
 
@@ -2170,7 +2192,7 @@ mod test {
 
         let endpoint = server.url().strip_prefix("http://").unwrap().to_string();
 
-        let observer = EventObserver::new(None, endpoint, timeout);
+        let observer = EventObserver::new(None, endpoint, timeout, false);
 
         // Call send_payload
         observer.send_payload(&payload, "/test");
@@ -2201,8 +2223,12 @@ mod test {
             tx.send(()).unwrap();
         });
 
-        let observer =
-            EventObserver::new(None, format!("127.0.0.1:{port}"), Duration::from_secs(3));
+        let observer = EventObserver::new(
+            None,
+            format!("127.0.0.1:{port}"),
+            Duration::from_secs(3),
+            false,
+        );
 
         let payload = json!({"key": "value"});
 
@@ -2250,8 +2276,12 @@ mod test {
             }
         });
 
-        let observer =
-            EventObserver::new(None, format!("127.0.0.1:{port}"), Duration::from_secs(3));
+        let observer = EventObserver::new(
+            None,
+            format!("127.0.0.1:{port}"),
+            Duration::from_secs(3),
+            false,
+        );
 
         let payload = json!({"key": "value"});
 
@@ -2298,7 +2328,7 @@ mod test {
             }
         });
 
-        let observer = EventObserver::new(None, format!("127.0.0.1:{port}"), timeout);
+        let observer = EventObserver::new(None, format!("127.0.0.1:{port}"), timeout, false);
 
         let payload = json!({"key": "value"});
 
@@ -2391,7 +2421,12 @@ mod test {
             }
         });
 
-        let observer = EventObserver::new(Some(working_dir), format!("127.0.0.1:{port}"), timeout);
+        let observer = EventObserver::new(
+            Some(working_dir),
+            format!("127.0.0.1:{port}"),
+            timeout,
+            false,
+        );
 
         let payload = json!({"key": "value"});
         let payload2 = json!({"key": "value2"});
diff --git a/testnet/stacks-node/src/tests/epoch_22.rs b/testnet/stacks-node/src/tests/epoch_22.rs
@@ -644,6 +644,7 @@ fn pox_2_unlock_all() {
         endpoint: format!("localhost:{}", test_observer::EVENT_OBSERVER_PORT),
         events_keys: vec![EventKeyType::AnyEvent],
         timeout_ms: 1000,
+        lossy: false,
     });
     conf.initial_balances.append(&mut initial_balances);
 
diff --git a/testnet/stacks-node/src/tests/neon_integrations.rs b/testnet/stacks-node/src/tests/neon_integrations.rs
@@ -657,6 +657,7 @@ pub mod test_observer {
             endpoint: format!("localhost:{EVENT_OBSERVER_PORT}"),
             events_keys: event_keys.to_vec(),
             timeout_ms: 1000,
+            lossy: false,
         });
     }
 
@@ -7490,6 +7491,7 @@ fn atlas_integration_test() {
             endpoint: format!("localhost:{}", test_observer::EVENT_OBSERVER_PORT),
             events_keys: vec![EventKeyType::AnyEvent],
             timeout_ms: 1000,
+            lossy: false,
         });
 
     conf_follower_node.node.always_use_affirmation_maps = false;
@@ -8023,6 +8025,7 @@ fn antientropy_integration_test() {
             endpoint: format!("localhost:{}", test_observer::EVENT_OBSERVER_PORT),
             events_keys: vec![EventKeyType::AnyEvent],
             timeout_ms: 1000,
+            lossy: false,
         });
 
     conf_follower_node.node.mine_microblocks = true;
diff --git a/testnet/stacks-node/src/tests/signer/mod.rs b/testnet/stacks-node/src/tests/signer/mod.rs
@@ -845,6 +845,7 @@ fn setup_stx_btc_node<G: FnMut(&mut NeonConfig)>(
                 EventKeyType::BurnchainBlocks,
             ],
             timeout_ms: 1000,
+            lossy: false,
         });
     }
 
@@ -860,6 +861,7 @@ fn setup_stx_btc_node<G: FnMut(&mut NeonConfig)>(
             EventKeyType::BurnchainBlocks,
         ],
         timeout_ms: 1000,
+        lossy: false,
     });
 
     // The signers need some initial balances in order to pay for epoch 2.5 transaction votes
diff --git a/testnet/stacks-node/src/tests/signer/v0.rs b/testnet/stacks-node/src/tests/signer/v0.rs
@@ -4238,6 +4238,7 @@ fn signer_set_rollover() {
                         EventKeyType::BurnchainBlocks,
                     ],
                     timeout_ms: 1000,
+                    lossy: false,
                 });
             }
             naka_conf.node.rpc_bind = rpc_bind.clone();
@@ -11114,6 +11115,7 @@ fn injected_signatures_are_ignored_across_boundaries() {
                     EventKeyType::BurnchainBlocks,
                 ],
                 timeout_ms: 1000,
+                lossy: false,
             });
             naka_conf.node.rpc_bind = rpc_bind.clone();
         },
