graph, store: Reject grafts below the base's earliest block

graph/src/components/store/traits.rs

@@ -219,6 +219,12 @@ pub trait SubgraphStore: Send + Sync + 'static {
     /// being set up
     async fn least_block_ptr(&self, id: &DeploymentHash) -> Result<Option<BlockPtr>, StoreError>;
 
+    /// Return the earliest block for which the deployment with the given
+    /// `id` still has entity data. Blocks below this are either pre-`startBlock`
+    /// or have been removed by pruning, and cannot serve as a graft or copy
+    /// source.
+    async fn earliest_block_number(&self, id: &DeploymentHash) -> Result<BlockNumber, StoreError>;
+
     async fn is_healthy(&self, id: &DeploymentHash) -> Result<bool, StoreError>;
 
     /// Find all deployment locators for the subgraph with the given hash.

graph/src/data/subgraph/mod.rs

@@ -600,7 +600,24 @@ impl Graft {
                 self.block,
                 ptr.number - 1
             ))),
-            (Some(_), _) => Ok(()),
+            (Some(_), _) => {
+                // The graft block must be at or above the base's earliest
+                // available block. Below that the base store no longer has
+                // the entity versions the copy step would need: either the
+                // base started above the graft block, or pruning has
+                // removed the history.
+                let earliest_block = store
+                    .earliest_block_number(&self.base)
+                    .await
+                    .map_err(|e| GraftBaseInvalid(e.to_string()))?;
+                if self.block < earliest_block {
+                    return Err(GraftBaseInvalid(format!(
+                        "failed to graft onto `{}` at block {} since its earliest available block is {}",
+                        self.base, self.block, earliest_block
+                    )));
+                }
+                Ok(())
+            }
         }
     }
 }

store/postgres/src/subgraph_store.rs

@@ -1888,6 +1888,12 @@ impl SubgraphStoreTrait for SubgraphStore {
         store.block_ptr(site.cheap_clone()).await
     }
 
+    async fn earliest_block_number(&self, id: &DeploymentHash) -> Result<BlockNumber, StoreError> {
+        let (store, site) = self.store(id).await?;
+        let state = store.deployment_state(site.cheap_clone()).await?;
+        Ok(state.earliest_block_number)
+    }
+
     async fn is_healthy(&self, id: &DeploymentHash) -> Result<bool, StoreError> {
         let (store, site) = self.store(id).await?;
         let health = store.health(&site).await?;