Merge branch 'graphprotocol:master' into shuaib/handle-max-endblock

Author: shuaibbapputty

graph/src/data/subgraph/schema.rs

@@ -1,6 +1,7 @@
 //! Entity types that contain the graph-node state.
 
 use anyhow::{anyhow, bail, Error};
+use chrono::{DateTime, Utc};
 use hex;
 use rand::rngs::OsRng;
 use rand::Rng;
@@ -159,7 +160,7 @@ pub struct SubgraphDeploymentEntity {
     pub manifest: SubgraphManifestEntity,
     pub failed: bool,
     pub health: SubgraphHealth,
-    pub synced: bool,
+    pub synced_at: Option<DateTime<Utc>>,
     pub fatal_error: Option<SubgraphError>,
     pub non_fatal_errors: Vec<SubgraphError>,
     /// The earliest block for which we have data

graph/src/data/value.rs

@@ -276,7 +276,7 @@ impl<'a> IntoIterator for &'a Object {
 impl std::fmt::Debug for Object {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         f.debug_map()
-            .entries(self.0.into_iter().map(|e| {
+            .entries(self.0.iter().map(|e| {
                 (
                     e.key.as_ref().map(|w| w.as_str()).unwrap_or("---"),
                     &e.value,

graph/src/schema/input/mod.rs

@@ -929,7 +929,7 @@ impl Aggregation {
 
     pub fn dimensions(&self) -> impl Iterator<Item = &Field> {
         self.fields
-            .into_iter()
+            .iter()
             .filter(|field| &field.name != &*ID && field.name != kw::TIMESTAMP)
     }
 
@@ -1240,7 +1240,7 @@ impl InputSchema {
         };
         Ok(obj_type
             .shared_interfaces
-            .into_iter()
+            .iter()
             .map(|atom| EntityType::new(self.cheap_clone(), *atom))
             .collect())
     }

graph/src/util/intern.rs

@@ -33,6 +33,7 @@ pub struct Atom(AtomInt);
 
 /// An atom and the underlying pool. A `FatAtom` can be used in place of a
 /// `String` or `Word`
+#[allow(dead_code)]
 pub struct FatAtom {
     pool: Arc<AtomPool>,
     atom: Atom,

server/index-node/src/schema.graphql

@@ -70,8 +70,12 @@ type SubgraphIndexingStatus {
   nonFatalErrors: [SubgraphError!]!
   chains: [ChainIndexingStatus!]!
   entityCount: BigInt!
+
+  "null if deployment is not assigned to an indexing node"
   node: String
-  paused: Boolean!
+  "null if deployment is not assigned to an indexing node"
+  paused: Boolean
+
   historyBlocks: Int!
 }
 

server/json-rpc/src/lib.rs

@@ -204,7 +204,7 @@ impl<R: SubgraphRegistrar> ServerState<R> {
 
     /// Handler for the `subgraph_resume` endpoint.
     async fn resume_handler(&self, params: SubgraphPauseParams) -> JsonRpcResult<GraphValue> {
-        info!(&self.logger, "Received subgraph_pause request"; "params" => format!("{:?}", params));
+        info!(&self.logger, "Received subgraph_resume request"; "params" => format!("{:?}", params));
 
         match self.registrar.resume_subgraph(&params.deployment).await {
             Ok(_) => Ok(Value::Null),

store/postgres/migrations/2024-07-22-140930_track_synced_date/down.sql

@@ -0,0 +1,32 @@
+DROP VIEW info.subgraph_info;
+
+ALTER TABLE subgraphs.subgraph_deployment ADD COLUMN synced BOOLEAN NOT NULL DEFAULT false;
+ALTER TABLE unused_deployments            ADD COLUMN synced BOOLEAN NOT NULL DEFAULT false;
+
+UPDATE subgraphs.subgraph_deployment SET synced = synced_at IS NOT NULL;
+UPDATE unused_deployments            SET synced = synced_at IS NOT NULL;
+
+-- NB: We keep the default on unused_deployment, as it was there before.
+ALTER TABLE subgraphs.subgraph_deployment ALTER COLUMN synced DROP DEFAULT;
+
+ALTER TABLE subgraphs.subgraph_deployment DROP COLUMN synced_at;
+ALTER TABLE unused_deployments            DROP COLUMN synced_at;
+
+CREATE VIEW info.subgraph_info AS
+SELECT ds.id AS schema_id,
+    ds.name AS schema_name,
+    ds.subgraph,
+    ds.version,
+    s.name,
+        CASE
+            WHEN s.pending_version = v.id THEN 'pending'::text
+            WHEN s.current_version = v.id THEN 'current'::text
+            ELSE 'unused'::text
+        END AS status,
+    d.failed,
+    d.synced
+   FROM deployment_schemas ds,
+    subgraphs.subgraph_deployment d,
+    subgraphs.subgraph_version v,
+    subgraphs.subgraph s
+  WHERE d.deployment = ds.subgraph::text AND v.deployment = d.deployment AND v.subgraph = s.id;

store/postgres/migrations/2024-07-22-140930_track_synced_date/up.sql

@@ -0,0 +1,29 @@
+DROP VIEW info.subgraph_info;
+
+ALTER TABLE subgraphs.subgraph_deployment ADD COLUMN synced_at TIMESTAMPTZ;
+ALTER TABLE unused_deployments            ADD COLUMN synced_at TIMESTAMPTZ;
+
+UPDATE subgraphs.subgraph_deployment SET synced_at = '1970-01-01 00:00:00 UTC' WHERE synced;
+UPDATE unused_deployments            SET synced_at = '1970-01-01 00:00:00 UTC' WHERE synced;
+
+ALTER TABLE subgraphs.subgraph_deployment DROP COLUMN synced;
+ALTER TABLE unused_deployments            DROP COLUMN synced;
+
+CREATE VIEW info.subgraph_info AS
+SELECT ds.id AS schema_id,
+    ds.name AS schema_name,
+    ds.subgraph,
+    ds.version,
+    s.name,
+        CASE
+            WHEN s.pending_version = v.id THEN 'pending'::text
+            WHEN s.current_version = v.id THEN 'current'::text
+            ELSE 'unused'::text
+        END AS status,
+    d.failed,
+    d.synced_at
+   FROM deployment_schemas ds,
+    subgraphs.subgraph_deployment d,
+    subgraphs.subgraph_version v,
+    subgraphs.subgraph s
+  WHERE d.deployment = ds.subgraph::text AND v.deployment = d.deployment AND v.subgraph = s.id;

store/postgres/src/deployment.rs

@@ -4,7 +4,7 @@
 use crate::{advisory_lock, detail::GraphNodeVersion, primary::DeploymentId};
 use diesel::{
     connection::SimpleConnection,
-    dsl::{count, delete, insert_into, select, sql, update},
+    dsl::{count, delete, insert_into, now, select, sql, update},
     sql_types::{Bool, Integer},
 };
 use diesel::{expression::SqlLiteral, pg::PgConnection, sql_types::Numeric};
@@ -132,7 +132,7 @@ table! {
         deployment -> Text,
         failed -> Bool,
         health -> crate::deployment::SubgraphHealthMapping,
-        synced -> Bool,
+        synced_at -> Nullable<Timestamptz>,
         fatal_error -> Nullable<Text>,
         non_fatal_errors -> Array<Text>,
         earliest_block_number -> Integer,
@@ -737,9 +737,9 @@ pub fn set_synced(conn: &mut PgConnection, id: &DeploymentHash) -> Result<(), St
     update(
         d::table
             .filter(d::deployment.eq(id.as_str()))
-            .filter(d::synced.eq(false)),
+            .filter(d::synced_at.is_null()),
     )
-    .set(d::synced.eq(true))
+    .set(d::synced_at.eq(now))
     .execute(conn)?;
     Ok(())
 }
@@ -762,7 +762,7 @@ pub fn exists_and_synced(conn: &mut PgConnection, id: &str) -> Result<bool, Stor
 
     let synced = d::table
         .filter(d::deployment.eq(id))
-        .select(d::synced)
+        .select(d::synced_at.is_not_null())
         .first(conn)
         .optional()?
         .unwrap_or(false);
@@ -1142,7 +1142,6 @@ pub fn create_deployment(
         d::id.eq(site.id),
         d::deployment.eq(site.deployment.as_str()),
         d::failed.eq(false),
-        d::synced.eq(false),
         d::health.eq(SubgraphHealth::Healthy),
         d::fatal_error.eq::<Option<String>>(None),
         d::non_fatal_errors.eq::<Vec<String>>(vec![]),

store/postgres/src/detail.rs

@@ -12,7 +12,10 @@ use git_testament::{git_testament, git_testament_macros};
 use graph::blockchain::BlockHash;
 use graph::data::store::scalar::ToPrimitive;
 use graph::data::subgraph::schema::{SubgraphError, SubgraphManifestEntity};
-use graph::prelude::{BigDecimal, BlockPtr, DeploymentHash, StoreError, SubgraphDeploymentEntity};
+use graph::prelude::{
+    chrono::{DateTime, Utc},
+    BigDecimal, BlockPtr, DeploymentHash, StoreError, SubgraphDeploymentEntity,
+};
 use graph::schema::InputSchema;
 use graph::{constraint_violation, data::subgraph::status, prelude::web3::types::H256};
 use itertools::Itertools;
@@ -46,7 +49,7 @@ pub struct DeploymentDetail {
     pub deployment: String,
     pub failed: bool,
     health: HealthType,
-    pub synced: bool,
+    pub synced_at: Option<DateTime<Utc>>,
     fatal_error: Option<String>,
     non_fatal_errors: Vec<String>,
     /// The earliest block for which we have history
@@ -188,7 +191,7 @@ pub(crate) fn info_from_details(
         deployment,
         failed: _,
         health,
-        synced,
+        synced_at,
         fatal_error: _,
         non_fatal_errors: _,
         earliest_block_number,
@@ -238,7 +241,7 @@ pub(crate) fn info_from_details(
     Ok(status::Info {
         id: id.into(),
         subgraph: deployment,
-        synced,
+        synced: synced_at.is_some(),
         health,
         paused: None,
         fatal_error,
@@ -446,7 +449,7 @@ impl StoredDeploymentEntity {
             manifest: manifest.as_manifest(schema),
             failed: detail.failed,
             health: detail.health.into(),
-            synced: detail.synced,
+            synced_at: detail.synced_at,
             fatal_error: None,
             non_fatal_errors: vec![],
             earliest_block_number: detail.earliest_block_number,