core, graph, store: Reset runner state only when needed

We only need to reset the state of the runner when the `WritableStore`
actually had to be restarted because of an error; if it had an error, we
have to reset the state.

Use `SubgraphRunner.revert_state` to properly reset the runner state.

Author: lutter

core/src/subgraph/runner.rs

@@ -183,13 +183,13 @@ where
                         // Restart the store to clear any errors that it
                         // might have encountered and use that from now on
                         let store = self.inputs.store.cheap_clone();
-                        let store = store.restart().await?;
-                        self.inputs = Arc::new(self.inputs.with_store(store));
-                        // Also clear the entity cache since we might have
-                        // entries in there that never made it to the
-                        // database
-                        self.state.entity_lfu_cache = LfuCache::new();
-                        self.state.synced = self.inputs.store.is_deployment_synced().await?;
+                        if let Some(store) = store.restart().await? {
+                            let last_good_block =
+                                store.block_ptr().map(|ptr| ptr.number).unwrap_or(0);
+                            self.revert_state(last_good_block)?;
+                            self.inputs = Arc::new(self.inputs.with_store(store));
+                            self.state.synced = self.inputs.store.is_deployment_synced().await?;
+                        }
                         break;
                     }
                 };

graph/src/components/store/traits.rs

@@ -331,10 +331,11 @@ pub trait WritableStore: ReadStore + DeploymentCursorTracker {
     /// contain unprocessed write requests that will be discarded by this
     /// call.
     ///
-    /// After this call, `self` should not be used anymore, as it will
-    /// continue to produce errors for any write requests, and instead, the
-    /// returned `WritableStore` should be used.
-    async fn restart(self: Arc<Self>) -> Result<Arc<dyn WritableStore>, StoreError>;
+    /// This call returns `None` if a restart was not needed because `self`
+    /// had no errors. If it returns `Some`, `self` should not be used
+    /// anymore, as it will continue to produce errors for any write
+    /// requests, and instead, the returned `WritableStore` should be used.
+    async fn restart(self: Arc<Self>) -> Result<Option<Arc<dyn WritableStore>>, StoreError>;
 }
 
 #[async_trait]

store/postgres/src/writable.rs

@@ -1316,16 +1316,19 @@ impl WritableStoreTrait for WritableStore {
         self.writer.flush().await
     }
 
-    async fn restart(self: Arc<Self>) -> Result<Arc<dyn WritableStoreTrait>, StoreError> {
+    async fn restart(self: Arc<Self>) -> Result<Option<Arc<dyn WritableStoreTrait>>, StoreError> {
         if self.poisoned() {
             let logger = self.store.logger.clone();
             if let Err(e) = self.stop().await {
                 warn!(logger, "Writable had error when stopping, it is safe to ignore this error"; "error" => e.to_string());
             }
             let store = Arc::new(self.store.store.0.clone());
-            store.writable(logger, self.store.site.id.into()).await
+            store
+                .writable(logger, self.store.site.id.into())
+                .await
+                .map(|store| Some(store))
         } else {
-            Ok(self)
+            Ok(None)
         }
     }
 }

store/test-store/tests/graph/entity_cache.rs

@@ -170,7 +170,7 @@ impl WritableStore for MockStore {
         unimplemented!()
     }
 
-    async fn restart(self: Arc<Self>) -> Result<Arc<dyn WritableStore>, StoreError> {
+    async fn restart(self: Arc<Self>) -> Result<Option<Arc<dyn WritableStore>>, StoreError> {
         unimplemented!()
     }
 }

store/test-store/tests/postgres/writable.rs

@@ -192,7 +192,7 @@ fn restart() {
 
         // We now have a poisoned store. Restarting it gives us a new store
         // that works again
-        let writable = writable.restart().await.unwrap();
+        let writable = writable.restart().await.unwrap().unwrap();
         writable.flush().await.unwrap();
 
         // Retry our write with correct data