store: Do not crash the node on invalid StoreEvent

lutter · lutter · commit 121e2688255e · 2021-03-04T15:23:24.000-08:00
It is possible to receive invalid store events when a node is starting up
because Postgres still has events sent by a previous version of the
code. In that case, the best we can do is ignore these events.

Once we've seen enough valid events, go back to panicing on invalid events.
diff --git a/store/postgres/src/store_events.rs b/store/postgres/src/store_events.rs
@@ -1,6 +1,6 @@
 use futures::sync::mpsc::{channel, Sender};
-use std::collections::HashMap;
-use std::sync::{Arc, Mutex, RwLock};
+use std::sync::{atomic::Ordering, Arc, Mutex, RwLock};
+use std::{collections::HashMap, sync::atomic::AtomicUsize};
 use uuid::Uuid;
 
 use crate::notification_listener::{NotificationListener, SafeChannelName};
@@ -9,12 +9,14 @@ use graph::prelude::serde_json;
 use graph::prelude::*;
 
 pub struct StoreEventListener {
+    logger: Logger,
     notification_listener: NotificationListener,
 }
 
 impl StoreEventListener {
     pub fn new(logger: &Logger, postgres_url: String) -> Self {
         StoreEventListener {
+            logger: logger.clone(),
             notification_listener: NotificationListener::new(
                 logger,
                 postgres_url,
@@ -34,17 +36,36 @@ impl EventProducer<StoreEvent> for StoreEventListener {
     ) -> Option<Box<dyn Stream<Item = StoreEvent, Error = ()> + Send>> {
         self.notification_listener.take_event_stream().map(
             |stream| -> Box<dyn Stream<Item = _, Error = _> + Send> {
-                Box::new(stream.map(|notification| {
-                    // Create StoreEvent from JSON
-                    let change: StoreEvent = serde_json::from_value(notification.payload.clone())
-                        .unwrap_or_else(|_| {
-                            panic!(
+                let logger = self.logger.clone();
+                Box::new(stream.filter_map(move |notification| {
+                    // When graph-node is starting up, it is possible that
+                    // Postgres still has old messages queued up that we
+                    // can't decode anymore. It is safe to skip them; once
+                    // We've seen 10 valid messages, we can assume that
+                    // whatever old messages Postgres had queued have been
+                    // cleared. Seeing an invalid message after that
+                    // definitely indicates trouble.
+                    let num_valid = AtomicUsize::new(0);
+                    serde_json::from_value(notification.payload.clone()).map_or_else(
+                        |_err| {
+                            error!(
+                                &logger,
                                 "invalid store event received from database: {:?}",
                                 notification.payload
-                            )
-                        });
-
-                    change
+                            );
+                            if num_valid.load(Ordering::SeqCst) > 10 {
+                                panic!(
+                                    "invalid store event received from database: {:?}",
+                                    notification.payload
+                                );
+                            }
+                            None
+                        },
+                        |change| {
+                            num_valid.fetch_add(1, Ordering::SeqCst);
+                            Some(change)
+                        },
+                    )
                 }))
             },
         )
