From a80548bcc850fd06f1d101008c3aeafcfb7a29c2 Mon Sep 17 00:00:00 2001 From: Liang Xu <24490353+lx3133584@users.noreply.github.com> Date: Sat, 7 Nov 2026 10:09:26 +0800 Subject: [PATCH] store: keep postponed attribute indexes pending when a graft copies from an unsynced base A graft or `graphman copy` made while the source has not reached the chain head yet produced a destination marked `postponed_indexes_created = true` without the postponed attribute indexes. `copy_data` recreates on the destination only the indexes that exist on the source, so a source that has not built them yet hands over none of them, and `create_indexes` then set the flag unconditionally. `create_postponed_indexes` returns early once that flag is set, so the destination never built them, not even at the chain head, and queries filtering on those attributes ran without an index. Set the flag on the destination only when the source has already created its own postponed indexes. That keeps the reason the flag was added to the copy path -- not recreating indexes that were dropped on the source on purpose -- while leaving it unset for a source that is simply still syncing, so the destination builds the indexes when it gets near the head. The regression test grafts at block 1 before the base builds anything, then lets both deployments run `create_postponed_indexes` and checks that the destination ends up with the attribute indexes for the columns it shares with the base. Signed-off-by: Liang Xu <24490353+lx3133584@users.noreply.github.com> --- store/postgres/src/copy.rs | 19 +++- store/test-store/tests/postgres/graft.rs | 123 +++++++++++++++++++++++ 2 files changed, 138 insertions(+), 4 deletions(-) diff --git a/store/postgres/src/copy.rs b/store/postgres/src/copy.rs index d35e585a653..ff1c8b5a95d 100644 --- a/store/postgres/src/copy.rs +++ b/store/postgres/src/copy.rs @@ -336,10 +336,21 @@ impl CopyState { creat.execute_many(conn, &new_idxs).await?; } - conn.transaction(async |conn| { - deployment::set_postponed_indexes_created(conn, &self.dst.site).await - }) - .await?; + // Mark the postponed indexes as created only if the source had + // already created its own. When the source is still syncing it has + // not built them yet, so `index_list` does not contain them and we + // just created none of them; claiming otherwise would make + // `create_postponed_indexes` return early once the destination gets + // near the chain head and leave those attribute indexes unbuilt for + // good. A source that created them and then had some dropped + // intentionally does have the flag set, and we keep honouring that. + let src_created = crate::deployment::postponed_indexes_created(conn, &self.src.site).await?; + if src_created { + conn.transaction(async |conn| { + deployment::set_postponed_indexes_created(conn, &self.dst.site).await + }) + .await?; + } Ok(()) } diff --git a/store/test-store/tests/postgres/graft.rs b/store/test-store/tests/postgres/graft.rs index 6536fc3e06c..359b722f800 100644 --- a/store/test-store/tests/postgres/graft.rs +++ b/store/test-store/tests/postgres/graft.rs @@ -670,3 +670,126 @@ fn prune() { }) } } + +/// Names of the indexes on `entity` that `CreateIndex::to_postpone()` +/// classifies as postponed (BTree attribute indexes, not on `id` alone) +async fn postponed_attr_indexes( + store: &DieselSubgraphStore, + loc: &DeploymentLocator, + entity: &str, +) -> Vec { + let mut names: Vec = store + .indexes_for_entity(loc, entity) + .await + .unwrap() + .into_iter() + .filter(|idx| idx.to_postpone()) + .filter_map(|idx| idx.name().map(str::to_string)) + .collect(); + names.sort(); + names +} + +/// `subgraphs.deployment.postponed_indexes_created` for `hash` +async fn postponed_flag(hash: &DeploymentHash) -> bool { + use diesel_async::RunQueryDsl; + + #[derive(diesel::QueryableByName)] + struct Flag { + #[diesel(sql_type = diesel::sql_types::Bool)] + postponed_indexes_created: bool, + } + + let mut conn = PRIMARY_POOL.get().await.unwrap(); + diesel::sql_query( + "select postponed_indexes_created from subgraphs.deployment where subgraph = $1", + ) + .bind::(hash.as_str()) + .get_result::(&mut conn) + .await + .unwrap() + .postponed_indexes_created +} + +/// `create_postponed_indexes` runs in a background task; wait for the flag +async fn wait_for_flag(hash: &DeploymentHash) { + for _ in 0..60 { + if postponed_flag(hash).await { + return; + } + tokio::time::sleep(std::time::Duration::from_millis(500)).await; + } + panic!("postponed indexes for {hash} were not created within 30s"); +} + +/// A graft made while the base has not built its postponed indexes yet must +/// not claim they were created: the destination still has to build them once +/// it gets near the chain head. +#[test] +fn graft_before_postponed_indexes() { + if !graph::env::ENV_VARS.postpone_attribute_index_creation { + println!("skipping: needs GRAPH_POSTPONE_ATTRIBUTE_INDEX_CREATION=true"); + return; + } + run_test(|store, src| async move { + // The source was created with postponement on and has not been near + // the chain head yet: no postponed index, flag false + assert!(!postponed_flag(&src.hash).await); + assert_eq!( + Vec::::new(), + postponed_attr_indexes(&store, &src, USER).await + ); + + // Graft onto the source at block 1, as `graph deploy` with a `graft:` + // section would; `graphman copy` goes through the same + // `copy::Connection::copy_data` + let dst_id = DeploymentHash::new("grafted_postponed").unwrap(); + let dst = create_grafted_subgraph( + &dst_id, + GRAFT_GQL, + TEST_SUBGRAPH_ID.as_str(), + BLOCKS[1].clone(), + ) + .await?; + + let dst_flag = postponed_flag(&dst.hash).await; + let dst_before = postponed_attr_indexes(&store, &dst, USER).await; + println!("dst after graft: flag={dst_flag} postponed={dst_before:?}"); + + // The source gets near the head: the runner calls + // `create_postponed_indexes`, which creates them + store + .cheap_clone() + .writable(LOGGER.clone(), src.id, Arc::new(Vec::new())) + .await? + .create_postponed_indexes() + .await?; + wait_for_flag(&src.hash).await; + let src_after = postponed_attr_indexes(&store, &src, USER).await; + println!("src after create_postponed_indexes: {src_after:?}"); + assert!(src_after.iter().any(|n| n.ends_with("_name"))); + assert!(src_after.iter().any(|n| n.ends_with("_email"))); + + // The destination gets near the head: same call + store + .cheap_clone() + .writable(LOGGER.clone(), dst.id, Arc::new(Vec::new())) + .await? + .create_postponed_indexes() + .await?; + wait_for_flag(&dst.hash).await; + // Leave the background task time to run if it were going to + tokio::time::sleep(std::time::Duration::from_secs(2)).await; + let dst_after = postponed_attr_indexes(&store, &dst, USER).await; + println!("dst after create_postponed_indexes: {dst_after:?}"); + + // The destination ends up with the postponed indexes for the columns + // it shares with the source + assert!( + dst_after.iter().any(|n| n.ends_with("_name")) + && dst_after.iter().any(|n| n.ends_with("_email")), + "graft marked postponed indexes as created but they are missing: {dst_after:?}" + ); + Ok(()) + }) +}