fix(store): handle orphaned sites in graft compatibility check

Move graft compatibility validation out of primary transaction to avoid
holding primary and shard connections simultaneously, which could exhaust
connection pools when they share the same database.

- Detect orphaned sites (site exists but deployment doesn't) and re-run
  the graft `can_copy_from` check on redeploy
- Only insert into `active_copies` when a copy is actually needed, avoiding
  spurious records for already-copied deployments
- Maintain idempotency: failed deployments leave state that will be properly
  validated on the next attempt

Signed-off-by: Maksim Dimitrov <dimitrov.maksim@gmail.com>

Author: dimitrovmaksim

store/postgres/src/subgraph_store.rs

@@ -354,57 +354,63 @@ impl SubgraphStore {
             let (shard, node_id) = self.place(&name, &network_name, node_id).await?;
 
             let mut pconn = self.primary_conn().await?;
-            pconn
-                .transaction(|pconn| {
-                    async {
-                        let (site, site_was_created) = pconn
-                            .allocate_site(shard, schema.id(), network_name, graft_base)
-                            .await?;
-                        let node_id = pconn.assigned_node(&site).await?.unwrap_or(node_id);
-                        let site = Arc::new(site);
-                        let deployment_store = self
-                            .stores
-                            .get(&site.shard)
-                            .ok_or_else(|| StoreError::UnknownShard(site.shard.to_string()))?;
-
-                        if site_was_created {
-                            if let Some(graft_base) = graft_base {
-                                // Ensure that the graft base exists
-                                let base_layout = self.layout(graft_base).await?;
-                                let mut shard_conn =
-                                    deployment_store.get_replica_conn(ReplicaId::Main).await?;
-                                let entities_with_causality_region =
-                                    deployment.manifest.entities_with_causality_region.clone();
-                                let catalog = Catalog::for_creation(
-                                    &mut shard_conn,
-                                    site.cheap_clone(),
-                                    entities_with_causality_region.into_iter().collect(),
-                                )
-                                .await?;
-                                let layout = Layout::new(site.cheap_clone(), schema, catalog)?;
-
-                                let errors = layout.can_copy_from(&base_layout);
-                                if !errors.is_empty() {
-                                    return Err(StoreError::Unknown(anyhow!(
-                                        "The subgraph `{}` cannot be used as the graft base \
+
+            let (site, site_was_created) = pconn
+                .allocate_site(shard, schema.id(), network_name, graft_base)
+                .await?;
+            let node_id = pconn.assigned_node(&site).await?.unwrap_or(node_id);
+            let site = Arc::new(site);
+            let deployment_store = self
+                .stores
+                .get(&site.shard)
+                .ok_or_else(|| StoreError::UnknownShard(site.shard.to_string()))?;
+
+            let mut shard_conn = deployment_store.get_replica_conn(ReplicaId::Main).await?;
+            let needs_check = if site_was_created {
+                true
+            } else {
+                // If deployment does not exist, but site exists it means
+                // that we are recovering from a failed deployment creation with an orphaned site.
+                // In that case, we should check graft compatibility again.
+                let exists = crate::deployment::exists(&mut shard_conn, &site).await?;
+                !exists
+            };
+
+            if let Some(graft_base) = graft_base {
+                let base_layout = self.layout(graft_base).await?;
+
+                if needs_check {
+                    let entities_with_causality_region =
+                        deployment.manifest.entities_with_causality_region.clone();
+                    let catalog = Catalog::for_creation(
+                        &mut shard_conn,
+                        site.cheap_clone(),
+                        entities_with_causality_region.into_iter().collect(),
+                    )
+                    .await?;
+                    let layout = Layout::new(site.cheap_clone(), schema, catalog)?;
+
+                    let errors = layout.can_copy_from(&base_layout);
+                    if !errors.is_empty() {
+                        return Err(StoreError::Unknown(anyhow!(
+                            "The subgraph `{}` cannot be used as the graft base \
                                 for `{}` because the schemas are incompatible:\n    - {}",
-                                        &base_layout.catalog.site.namespace,
-                                        &layout.catalog.site.namespace,
-                                        errors.join("\n    - ")
-                                    )));
-                                }
-
-                                pconn
-                                    .record_active_copy(base_layout.site.as_ref(), site.as_ref())
-                                    .await?;
-                            }
-                        }
-
-                        Ok((site, deployment_store, node_id))
+                            &base_layout.catalog.site.namespace,
+                            &layout.catalog.site.namespace,
+                            errors.join("\n    - ")
+                        )));
                     }
-                    .scope_boxed()
-                })
-                .await?
+
+                    // Only record active copy when the graft check passes and a copy is needed.
+                    // If deployment already exists, the copy has either completed (no active_copies
+                    // record) or is in progress (active_copies record already exists).
+                    pconn
+                        .record_active_copy(base_layout.site.as_ref(), site.as_ref())
+                        .await?;
+                }
+            }
+
+            (site, deployment_store, node_id)
         };
 
         // Create the actual databases schema and metadata entries