Store synced_at_block_number when a deployment syncs (#5610)

This is a follow-up to #5566

Author: encalypto

graph/src/components/store/traits.rs

@@ -350,7 +350,7 @@ pub trait WritableStore: ReadStore + DeploymentCursorTracker {
     ) -> Result<(), StoreError>;
 
     /// Force synced status, used for testing.
-    fn deployment_synced(&self) -> Result<(), StoreError>;
+    fn deployment_synced(&self, block_ptr: BlockPtr) -> Result<(), StoreError>;
 
     /// Return true if the deployment with the given id is fully synced, and return false otherwise.
     /// Cheap, cached operation.

store/postgres/migrations/2024-08-14-205601_store_synced_at_block/down.sql

@@ -0,0 +1,2 @@
+ALTER TABLE subgraphs.subgraph_deployment DROP COLUMN synced_at_block_number;
+ALTER TABLE unused_deployments            DROP COLUMN synced_at_block_number;

store/postgres/migrations/2024-08-14-205601_store_synced_at_block/up.sql

@@ -0,0 +1,2 @@
+ALTER TABLE subgraphs.subgraph_deployment ADD COLUMN synced_at_block_number INT4;
+ALTER TABLE unused_deployments            ADD COLUMN synced_at_block_number INT4;

store/postgres/src/deployment.rs

@@ -133,6 +133,7 @@ table! {
         failed -> Bool,
         health -> crate::deployment::SubgraphHealthMapping,
         synced_at -> Nullable<Timestamptz>,
+        synced_at_block_number -> Nullable<Int4>,
         fatal_error -> Nullable<Text>,
         non_fatal_errors -> Array<Text>,
         earliest_block_number -> Integer,
@@ -731,15 +732,22 @@ pub fn state(conn: &mut PgConnection, id: DeploymentHash) -> Result<DeploymentSt
 }
 
 /// Mark the deployment `id` as synced
-pub fn set_synced(conn: &mut PgConnection, id: &DeploymentHash) -> Result<(), StoreError> {
+pub fn set_synced(
+    conn: &mut PgConnection,
+    id: &DeploymentHash,
+    block_ptr: BlockPtr,
+) -> Result<(), StoreError> {
     use subgraph_deployment as d;
 
     update(
         d::table
             .filter(d::deployment.eq(id.as_str()))
             .filter(d::synced_at.is_null()),
     )
-    .set(d::synced_at.eq(now))
+    .set((
+        d::synced_at.eq(now),
+        d::synced_at_block_number.eq(block_ptr.number),
+    ))
     .execute(conn)?;
     Ok(())
 }

store/postgres/src/deployment_store.rs

@@ -563,9 +563,13 @@ impl DeploymentStore {
         deployment::exists_and_synced(&mut conn, id.as_str())
     }
 
-    pub(crate) fn deployment_synced(&self, id: &DeploymentHash) -> Result<(), StoreError> {
+    pub(crate) fn deployment_synced(
+        &self,
+        id: &DeploymentHash,
+        block_ptr: BlockPtr,
+    ) -> Result<(), StoreError> {
         let mut conn = self.get_conn()?;
-        conn.transaction(|conn| deployment::set_synced(conn, id))
+        conn.transaction(|conn| deployment::set_synced(conn, id, block_ptr))
     }
 
     /// Look up the on_sync action for this deployment

store/postgres/src/detail.rs

@@ -50,6 +50,7 @@ pub struct DeploymentDetail {
     pub failed: bool,
     health: HealthType,
     pub synced_at: Option<DateTime<Utc>>,
+    pub synced_at_block_number: Option<i32>,
     fatal_error: Option<String>,
     non_fatal_errors: Vec<String>,
     /// The earliest block for which we have history

store/postgres/src/primary.rs

@@ -180,6 +180,7 @@ table! {
         latest_ethereum_block_number -> Nullable<Integer>,
         failed -> Bool,
         synced_at -> Nullable<Timestamptz>,
+        synced_at_block_number -> Nullable<Int4>,
     }
 }
 
@@ -233,6 +234,7 @@ pub struct UnusedDeployment {
     pub latest_ethereum_block_number: Option<i32>,
     pub failed: bool,
     pub synced_at: Option<DateTime<Utc>>,
+    pub synced_at_block_number: Option<i32>,
 }
 
 #[derive(Clone, Debug, PartialEq, Eq, Hash, AsExpression, FromSqlRow)]
@@ -1681,6 +1683,7 @@ impl<'a> Connection<'a> {
                     u::latest_ethereum_block_number.eq(latest_number),
                     u::failed.eq(detail.failed),
                     u::synced_at.eq(detail.synced_at),
+                    u::synced_at_block_number.eq(detail.synced_at_block_number.clone()),
                 ))
                 .execute(self.conn.as_mut())?;
         }

store/postgres/src/writable.rs

@@ -420,7 +420,7 @@ impl SyncStore {
         }
     }
 
-    fn deployment_synced(&self) -> Result<(), StoreError> {
+    fn deployment_synced(&self, block_ptr: BlockPtr) -> Result<(), StoreError> {
         retry::forever(&self.logger, "deployment_synced", || {
             let event = {
                 // Make sure we drop `pconn` before we call into the deployment
@@ -452,7 +452,8 @@ impl SyncStore {
                 }
             }
 
-            self.writable.deployment_synced(&self.site.deployment)?;
+            self.writable
+                .deployment_synced(&self.site.deployment, block_ptr.clone())?;
 
             self.store.send_store_event(&event)
         })
@@ -1659,7 +1660,7 @@ impl WritableStoreTrait for WritableStore {
         is_caught_up_with_chain_head: bool,
     ) -> Result<(), StoreError> {
         if is_caught_up_with_chain_head {
-            self.deployment_synced()?;
+            self.deployment_synced(block_ptr_to.clone())?;
         } else {
             self.writer.start_batching();
         }
@@ -1696,10 +1697,10 @@ impl WritableStoreTrait for WritableStore {
     /// - Disable the time-to-sync metrics gathering.
     /// - Stop batching writes.
     /// - Promote it to 'synced' status in the DB, if that hasn't been done already.
-    fn deployment_synced(&self) -> Result<(), StoreError> {
+    fn deployment_synced(&self, block_ptr: BlockPtr) -> Result<(), StoreError> {
         self.writer.deployment_synced();
         if !self.is_deployment_synced.load(Ordering::SeqCst) {
-            self.store.deployment_synced()?;
+            self.store.deployment_synced(block_ptr)?;
             self.is_deployment_synced.store(true, Ordering::SeqCst);
         }
         Ok(())

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

@@ -156,7 +156,7 @@ impl WritableStore for MockStore {
         unimplemented!()
     }
 
-    fn deployment_synced(&self) -> Result<(), StoreError> {
+    fn deployment_synced(&self, _block_ptr: BlockPtr) -> Result<(), StoreError> {
         unimplemented!()
     }
 

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

@@ -482,7 +482,7 @@ fn on_sync() {
                     .await?;
 
                 writable.start_subgraph_deployment(&LOGGER).await?;
-                writable.deployment_synced()?;
+                writable.deployment_synced(BLOCKS[0].clone())?;
 
                 let mut primary = primary_connection();
                 let src_site = primary.locate_site(src)?.unwrap();
@@ -539,7 +539,7 @@ fn on_sync() {
             store.activate(&dst)?;
             store.remove_deployment(src.id.into())?;
 
-            let res = writable.deployment_synced();
+            let res = writable.deployment_synced(BLOCKS[2].clone());
             assert!(res.is_ok());
         }
         Ok(())