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(()) + }) +}