store: take the max of source and destination history_blocks during graft

The graft finalization previously overwrote the destination's
history_blocks with the source's value unconditionally. As a
result, a destination subgraph manifesting `prune: never`
(history_blocks = BLOCK_NUMBER_MAX) grafted from a `prune: auto`
source silently inherited the source's much smaller retention
window, contradicting the destination's manifest.

Use max(src, dst) instead. The destination's retention is at
least as long as the source's (otherwise the inherited copied
data would be immediately eligible for pruning) while a
destination that requests longer retention is no longer silently
downgraded.

Also expose `SubgraphStore::history_blocks` mirroring
`set_history_blocks`, so the test can verify the post-graft
retention value.

Author: rslowinski

store/postgres/src/deployment_store.rs

@@ -1591,11 +1591,11 @@ impl DeploymentStore {
             }
 
             let src_manifest_idx_and_name = src_deployment.manifest.template_idx_and_name()?;
-            let dst_manifest_idx_and_name = self
-                .load_deployment(dst.site.clone())
-                .await?
-                .manifest
-                .template_idx_and_name()?;
+            // Keep `dst_deployment` bound so we can read its
+            // `history_blocks` below; otherwise the chained
+            // `.manifest.template_idx_and_name()?` would consume it.
+            let dst_deployment = self.load_deployment(dst.site.clone()).await?;
+            let dst_manifest_idx_and_name = dst_deployment.manifest.template_idx_and_name()?;
 
             // Copy subgraph data
             // We allow both not copying tables at all from the source, as well
@@ -1672,10 +1672,25 @@ impl DeploymentStore {
                 info!(logger, "Rewound subgraph to block {}", block.number;
                   "time_ms" => start.elapsed().as_millis());
 
+                // Use the *maximum* of the source's and destination's
+                // `history_blocks` rather than overwriting the destination
+                // with the source's value. The destination has just
+                // received the source's full retained history, so its
+                // retention must be at least as long as the source's
+                // (otherwise the inherited data would be immediately
+                // eligible for pruning); but if the destination's
+                // manifest requests longer retention (notably
+                // `prune: never`, which yields `BLOCK_NUMBER_MAX`), that
+                // intent must be honoured. The previous unconditional
+                // overwrite silently downgraded `prune: never` children
+                // to whatever retention the parent used.
                 deployment::set_history_blocks(
                     conn,
                     &dst.site,
-                    src_deployment.manifest.history_blocks,
+                    src_deployment
+                        .manifest
+                        .history_blocks
+                        .max(dst_deployment.manifest.history_blocks),
                 )
                 .await?;
 

store/postgres/src/subgraph_store.rs

@@ -1520,6 +1520,18 @@ impl Inner {
             .await
     }
 
+    /// Return the current `history_blocks` retention setting recorded in
+    /// the deployment's manifest. Mirrors [`Self::set_history_blocks`] and
+    /// is useful for tooling and tests that need to inspect the effective
+    /// retention after operations such as grafting.
+    pub async fn history_blocks(
+        &self,
+        deployment: &DeploymentLocator,
+    ) -> Result<BlockNumber, StoreError> {
+        let site = self.find_site(deployment.id.into()).await?;
+        Ok(self.load_deployment(site).await?.manifest.history_blocks)
+    }
+
     pub async fn load_deployment(
         &self,
         site: Arc<Site>,

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

@@ -816,3 +816,52 @@ fn graft_store_path_rejects_below_prune_floor() {
         Ok(())
     })
 }
+
+/// Grafting must not silently shorten the destination's `history_blocks`
+/// retention.
+///
+/// Before the fix, `start_subgraph` unconditionally wrote the source
+/// subgraph's `history_blocks` into the destination, so a `prune: never`
+/// child grafted from a `prune: auto` parent silently inherited the
+/// parent's aggressive retention window. The fix takes the *maximum* of
+/// the two, so the destination keeps at least as much history as it has
+/// (the inherited copied data) but is never downgraded from a longer
+/// retention it explicitly requested.
+///
+/// The test setup matches that scenario: the destination is built by
+/// `test_store::create_subgraph`, whose manifest has `indexer_hints: None`
+/// and therefore reports `history_blocks = BLOCK_NUMBER_MAX` (the
+/// `prune: never` equivalent), while we manually lower the source's
+/// retention so old / new behaviour are observably different.
+#[test]
+fn graft_history_blocks_takes_max_of_parent_and_child() {
+    run_test(|store, src| async move {
+        // Lower the source's `history_blocks` to a small value so that
+        // the "max vs overwrite" outcomes diverge measurably.
+        let src_hb = 2;
+        let reorg_threshold = 1;
+        store
+            .set_history_blocks(&src, src_hb, reorg_threshold)
+            .await
+            .expect("lowering source history_blocks");
+
+        let graft_id = DeploymentHash::new("grafted_history_blocks").unwrap();
+        let dst =
+            create_grafted_subgraph(&graft_id, GRAFT_GQL, src.hash.as_str(), BLOCKS[1].clone())
+                .await
+                .expect("graft succeeds");
+
+        let dst_hb = store
+            .history_blocks(&dst)
+            .await
+            .expect("read dst history_blocks");
+        assert_eq!(
+            dst_hb, BLOCK_NUMBER_MAX,
+            "graft must keep the destination's longer retention \
+             (BLOCK_NUMBER_MAX = {BLOCK_NUMBER_MAX}) rather than overwriting it \
+             with the source's smaller value ({src_hb})"
+        );
+
+        Ok(())
+    })
+}