Ensure earliest_block on copies is set correctly

store/postgres/migrations/2023-03-28-004152_trigger_remap/down.sql

@@ -0,0 +1,2 @@
+-- No schema changes, migration is only there to trigger remapping of
+-- foreign metadata

store/postgres/migrations/2023-03-28-004152_trigger_remap/up.sql

@@ -0,0 +1,3 @@
+-- No schema changes, migration is only there to trigger remapping of
+-- foreign metadata
+select 1;

store/postgres/src/catalog.rs

@@ -441,11 +441,7 @@ pub fn set_account_like(
 }
 
 pub fn copy_account_like(conn: &PgConnection, src: &Site, dst: &Site) -> Result<usize, StoreError> {
-    let src_nsp = if src.shard == dst.shard {
-        "subgraphs".to_string()
-    } else {
-        ForeignServer::metadata_schema(&src.shard)
-    };
+    let src_nsp = ForeignServer::metadata_schema_in(&src.shard, &dst.shard);
     let query = format!(
         "insert into subgraphs.table_stats(deployment, table_name, is_account_like, last_pruned_block)
          select $2 as deployment, ts.table_name, ts.is_account_like, ts.last_pruned_block

store/postgres/src/connection_pool.rs

@@ -54,12 +54,24 @@ impl ForeignServer {
         format!("shard_{}", shard.as_str())
     }
 
-    /// The name of the schema under which the `subgraphs` schema for `shard`
-    /// is accessible in shards that are not `shard`
+    /// The name of the schema under which the `subgraphs` schema for
+    /// `shard` is accessible in shards that are not `shard`. In most cases
+    /// you actually want to use `metadata_schema_in`
     pub fn metadata_schema(shard: &Shard) -> String {
         format!("{}_subgraphs", Self::name(shard))
     }
 
+    /// The name of the schema under which the `subgraphs` schema for
+    /// `shard` is accessible in the shard `current`. It is permissible for
+    /// `shard` and `current` to be the same.
+    pub fn metadata_schema_in(shard: &Shard, current: &Shard) -> String {
+        if shard == current {
+            "subgraphs".to_string()
+        } else {
+            Self::metadata_schema(&shard)
+        }
+    }
+
     pub fn new_from_raw(shard: String, postgres_url: &str) -> Result<Self, anyhow::Error> {
         Self::new(Shard::new(shard)?, postgres_url)
     }
@@ -202,6 +214,8 @@ impl ForeignServer {
             "subgraph_deployment_assignment",
             "subgraph",
             "subgraph_version",
+            "subgraph_deployment",
+            "subgraph_manifest",
         ] {
             let create_stmt =
                 catalog::create_foreign_table(conn, "subgraphs", table_name, &nsp, &self.name)?;

store/postgres/src/deployment.rs

@@ -944,11 +944,7 @@ pub(crate) fn copy_errors(
 ) -> Result<usize, StoreError> {
     use subgraph_error as e;
 
-    let src_nsp = if src.shard == dst.shard {
-        "subgraphs".to_string()
-    } else {
-        ForeignServer::metadata_schema(&src.shard)
-    };
+    let src_nsp = ForeignServer::metadata_schema_in(&src.shard, &dst.shard);
 
     // Check whether there are any errors for dst which indicates we already
     // did copy
@@ -1167,6 +1163,26 @@ pub fn set_earliest_block(
     Ok(())
 }
 
+/// Copy the `earliest_block` attribute from `src` to `dst`. The copy might
+/// go across shards and use the metadata tables mapped into the shard for
+/// `conn` which must be the shard for `dst`
+pub fn copy_earliest_block(conn: &PgConnection, src: &Site, dst: &Site) -> Result<(), StoreError> {
+    use subgraph_deployment as d;
+
+    let src_nsp = ForeignServer::metadata_schema_in(&src.shard, &dst.shard);
+
+    let query = format!(
+        "(select earliest_block_number from {src_nsp}.subgraph_deployment where id = {})",
+        src.id
+    );
+
+    update(d::table.filter(d::id.eq(dst.id)))
+        .set(d::earliest_block_number.eq(sql(&query)))
+        .execute(conn)?;
+
+    Ok(())
+}
+
 pub fn on_sync(conn: &PgConnection, id: impl Into<DeploymentId>) -> Result<OnSync, StoreError> {
     use subgraph_manifest as m;
 

store/postgres/src/deployment_store.rs

@@ -1577,12 +1577,6 @@ impl DeploymentStore {
                 info!(logger, "Counted the entities";
                       "time_ms" => start.elapsed().as_millis());
 
-                deployment::set_earliest_block(
-                    &conn,
-                    &dst.site,
-                    src_deployment.earliest_block_number,
-                )?;
-
                 deployment::set_history_blocks(
                     &conn,
                     &dst.site,
@@ -1594,6 +1588,15 @@ impl DeploymentStore {
                     self.analyze_with_conn(site.cheap_clone(), entity_name.as_str(), &conn)?;
                 }
 
+                // The `earliest_block` for `src` might have changed while
+                // we did the copy if `src` was pruned while we copied;
+                // adjusting it very late in the copy process ensures that
+                // we truly do have all the data starting at
+                // `earliest_block` and do not inadvertently expose data
+                // that might be incomplete because a prune on the source
+                // removed data just before we copied it
+                deployment::copy_earliest_block(&conn, &src.site, &dst.site)?;
+
                 // Set the block ptr to the graft point to signal that we successfully
                 // performed the graft
                 crate::deployment::forward_block_ptr(&conn, &dst.site.deployment, &block)?;

store/postgres/src/dynds/shared.rs

@@ -183,11 +183,7 @@ pub(crate) fn copy(
         return Ok(0);
     }
 
-    let src_nsp = if src.shard == dst.shard {
-        "subgraphs".to_string()
-    } else {
-        ForeignServer::metadata_schema(&src.shard)
-    };
+    let src_nsp = ForeignServer::metadata_schema_in(&src.shard, &dst.shard);
 
     // Check whether there are any dynamic data sources for dst which
     // indicates we already did copy