Skip to content

Bugfix: Ensure can_copy_from failures fully roll back deployment creation #6228

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
dimitrovmaksim wants to merge 2 commits into
base: master
Choose a base branch
from
+51 −44
Open

Bugfix: Ensure can_copy_from failures fully roll back deployment creation #6228

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
f915084
store: Fix histogram_bounds query
dimitrovmaksim Dec 16, 2025
8926e4b
store: Reject incompatible graft schemas during site allocation
dimitrovmaksim Dec 16, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion store/postgres/src/catalog.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1110,7 +1110,7 @@ pub(crate) async fn histogram_bounds(
table: &SqlName,
column: &str,
) -> Result<Vec<i64>, StoreError> {
const QUERY: &str = "select histogram_bounds::text::int8[] bounds \
const QUERY: &str = "select coalesce(histogram_bounds::text::int8[], '{}'::int8[]) as bounds \
Copy link
Collaborator

@lutter lutter Jan 12, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

woops .. nice!

from pg_stats \
where schemaname = $1 \
and tablename = $2 \
Expand Down
16 changes: 1 addition & 15 deletions store/postgres/src/deployment_store.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -184,7 +184,6 @@ impl DeploymentStore {
schema: &InputSchema,
deployment: DeploymentCreate,
site: Arc<Site>,
graft_base: Option<Arc<Layout>>,
replace: bool,
on_sync: OnSync,
index_def: Option<IndexList>,
Expand Down Expand Up @@ -217,27 +216,14 @@ impl DeploymentStore {
let query = format!("create schema {}", &site.namespace);
conn.batch_execute(&query).await?;

let layout = Layout::create_relational_schema(
let _ = Layout::create_relational_schema(
conn,
site.clone(),
schema,
entities_with_causality_region.into_iter().collect(),
index_def,
)
.await?;
// See if we are grafting and check that the graft is permissible
if let Some(base) = graft_base {
let errors = layout.can_copy_from(&base);
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.catalog.site.namespace,
&layout.catalog.site.namespace,
errors.join("\n - ")
)));
}
}

// Create data sources table
if site.schema_version.private_data_sources() {
Expand Down
77 changes: 49 additions & 28 deletions store/postgres/src/subgraph_store.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,7 @@ use graph::{
use graph::{derive::CheapClone, futures03::future::join_all};

use crate::{
catalog::Catalog,
deployment::{OnSync, SubgraphHealth},
primary::{self, DeploymentId, Mirror as PrimaryMirror, Primary, Site},
relational::{
Expand Down Expand Up @@ -88,7 +89,7 @@ impl Shard {
.all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '_')
{
return Err(StoreError::InvalidIdentifier(format!(
"shard name `{}` is invalid: shard names must only contain lowercase alphanumeric characters or '_'", name
"shard name `{name}` is invalid: shard names must only contain lowercase alphanumeric characters or '_'"
)));
}
Ok(Shard(name))
Expand Down Expand Up @@ -351,33 +352,60 @@ impl SubgraphStore {
// assignment that we used last time to avoid creating
// the same deployment in another shard
let (shard, node_id) = self.place(&name, &network_name, node_id).await?;

let mut conn = self.primary_conn().await?;
let (site, site_was_created) = conn
.allocate_site(shard, schema.id(), network_name, graft_base)
.await?;
let node_id = conn.assigned_node(&site).await?.unwrap_or(node_id);
(site, !site_was_created, node_id)
conn.transaction(|conn| {
async {
let (site, site_was_created) = conn
.allocate_site(shard, schema.id(), network_name, graft_base)
.await?;
let node_id = conn.assigned_node(&site).await?.unwrap_or(node_id);
let site = Arc::new(site);

if let Some(graft_base) = graft_base {
// Ensure that the graft base exists
let base_layout = self.layout(graft_base).await?;
let entities_with_causality_region =
deployment.manifest.entities_with_causality_region.clone();
let catalog = Catalog::for_tests(
Copy link
Collaborator

@lutter lutter Jan 12, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is wrong - for_tests doesn't actually check the database for its capabilities (and the method should be marked as #[cfg(debug_assertions)]). Instead, this needs to use Catalog::for_creation where the connection is a connection to the shard in which we will create the subgraph. It's best to create the catalog object outside of this transaction so that we don't hold multiple db connections at once.

Copy link
Member Author

@dimitrovmaksim dimitrovmaksim Jan 13, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

for_tests doesn't actually check the database for its capabilities

For this specific case the catalog is needed just to create the layout object, so the can_copy_from helpers can be used to compare the src and dst schemas. Later in the code, when creating the deployment relational schema, we construct the catalog with the proper checks. Although I agree using a for_tests function does not look good.

Instead, this needs to use Catalog::for_creation where the connection is a connection to the shard

At this point the only option (i see) to get a shard conn is using deployment_store.get_replica_conn(ReplicaId::Main) because the pool field is private.

It's best to create the catalog object outside of this transaction so that we don't hold multiple db connections at once.

Catalog creation requires to have the site already created, which happens in the transaction, and the point of adding the transaction is to revert the creation of the site if the can_copy_from fails (which is the cause of the original issue) so I'm not sure if holding a primary and a shard conn can be avoided. Unless, instead of using a transaction, we don't just execute drop_site if the can_copy_from check fails. But I may be missing something.

I wanted to make this fix with minimal changes to the current workflow, but maybe I should rething the whole workflow instead of just patching in.

site.cheap_clone(),
entities_with_causality_region.into_iter().collect(),
)?;
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 - ")
)));
}
}

Ok((site, !site_was_created, node_id))
}
.scope_boxed()
})
.await?
};
let site = Arc::new(site);

// if the deployment already exists, we don't need to perform any copying
// so we can set graft_base to None
// if it doesn't exist, we need to copy the graft base to the new deployment
let graft_base_layout = if !exists {
let graft_base = match deployment.graft_base.as_ref() {
Some(base) => Some(self.layout(&base).await?),
// If the deployment already exists, we don't need to perform any copying
// If it doesn't exist, we need to copy the graft base to the new deployment
if !exists {
let graft_base_layout = match graft_base {
Some(base) => Some(self.layout(base).await?),
None => None,
};

if let Some(graft_base) = &graft_base {
if let Some(graft_base_layout) = &graft_base_layout {
self.primary_conn()
.await?
.record_active_copy(graft_base.site.as_ref(), site.as_ref())
.record_active_copy(graft_base_layout.site.as_ref(), site.as_ref())
Copy link
Collaborator

@lutter lutter Jan 12, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Shouldn't this also happen in the transaction above? Otherwise, can't we end up in a situation where we set up everything except recording the active copy if graph-node gets killed atthe wrong moment?

Copy link
Member Author

@dimitrovmaksim dimitrovmaksim Jan 13, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good catch, I was only focusing on the incompatible schemas case

.await?;
}
graft_base
} else {
None
};

// Create the actual databases schema and metadata entries
Expand All @@ -386,7 +414,7 @@ impl SubgraphStore {
.get(&site.shard)
.ok_or_else(|| StoreError::UnknownShard(site.shard.to_string()))?;

let index_def = if let Some(graft) = &graft_base.clone() {
let index_def = if let Some(graft) = graft_base {
if let Some(site) = self.sites.get(graft) {
let store = self
.stores
Expand All @@ -406,7 +434,6 @@ impl SubgraphStore {
schema,
deployment,
site.clone(),
graft_base_layout,
replace,
OnSync::None,
index_def,
Expand Down Expand Up @@ -731,18 +758,15 @@ impl Inner {

if src.id == dst.id {
return Err(StoreError::Unknown(anyhow!(
"can not copy deployment {} onto itself",
src_loc
"can not copy deployment {src_loc} onto itself"
)));
}
// The very last thing we do when we set up a copy here is assign it
// to a node. Therefore, if `dst` is already assigned, this function
// should not have been called.
if let Some(node) = self.mirror.assigned_node(dst.as_ref()).await? {
return Err(StoreError::Unknown(anyhow!(
"can not copy into deployment {} since it is already assigned to node `{}`",
dst_loc,
node
"can not copy into deployment {dst_loc} since it is already assigned to node `{node}`"
)));
}
let deployment = src_store.load_deployment(src.clone()).await?;
Expand All @@ -758,8 +782,6 @@ impl Inner {
history_blocks_override: None,
};

let graft_base = self.layout(&src.deployment).await?;

self.primary_conn()
.await?
.record_active_copy(src.as_ref(), dst.as_ref())
Expand All @@ -776,7 +798,6 @@ impl Inner {
&src_layout.input_schema,
deployment,
dst.clone(),
Some(graft_base),
false,
on_sync,
Some(index_def),
Expand Down
Loading