Implement opportunistic boundary completion

Summary:
## Context
Currently, we have a problem with backfills for repos with many commits, and with large working copies near the top of the repo:

Our strategy consists of splitting the repo into "slices" and deriving "boundaries" (the parents of the base commits for a set of slices) from a "predecessor" derived data types.
Deriving a boundary this way does work that's `O(working copy size)`, but it only needs the predecessor type to exist for the commit, instead of the target type to exist for its parent.
This enables us to take advantage of the number of workers we have available as we can parallelize the work, which is necessary for a repo like `configerator` (130M public commits) or `fbsource` (70M public commits).

There is one hiccup with this strategy, though: near the top of a repo like fbsource, the working copy is so large that a single "derive boundaries request" can take weeks to compute.
By that time, it's quite likely that some previous slices caught up, so we're waiting for boundary derivation for nothing and just wasting time.

In practice, it has meant that we have had to fiddle with the backfilling parameters: do a first pass with "slice size" 100k or so, which parallelizes the work optimally,
then do a second pass with "slice size" 5M, which allows to avoid having boundary commits too near the top of the repo where computation is unacceptably expensive.

## Solution
Here, we introduce the concept of "opportunistic boundary completion": instead of always waiting for the "derive from predecessors" calls to finish, we will move on if the commits we are trying to derive
get derived by some other job before we have time to complete them.

This should be strictly more performant, and hopefully will remove a lot of the fiddling work in backfilling large repos.

The entire feature is gated by a JK, because I only trust the implementation so much, and

safe_jk2

`scm/mononoke:derived_data_backfill_opportunistic_boundary_completion`: the jk was created [here](https://www.internalfb.com/intern/justknobs/?name=scm%2Fmononoke#derived_data_backfill_opportunistic_boundary_completion) and set to false for now

Reviewed By: lmvasquezg

Differential Revision: D109853257

fbshipit-source-id: d9008c5889e4b0958a86aff0b31d2ff857be7202

Author: Pierre Chevalier

eden/mononoke/common/mononoke_macros/test_just_knobs/just_knobs.json

@@ -3,6 +3,7 @@
     "scm/mononoke:cross_repo_skip_backsyncing_ordinary_empty_commits": false,
     "scs/mononoke:defer_to_backsyncer_for_backsync": true,
     "scm/mononoke:dbcm_read_from_manifest": true,
+    "scm/mononoke:derived_data_backfill_opportunistic_boundary_completion": true,
     "scm/mononoke:derived_data_disable_derivation_workers": false,
     "scm/mononoke:derived_data_disable_remote_derivation": false,
     "scm/mononoke:derived_data_use_content_manifests": true,

eden/mononoke/features/async_requests/worker_lib/BUCK

@@ -7,11 +7,20 @@ rust_library(
     name = "worker_lib",
     srcs = glob(["src/**/*.rs"]),
     test_deps = [
+        "//common/rust/shed/facet:facet",
         "//common/rust/shed/fbinit:fbinit",
         "//common/rust/shed/fbinit:fbinit-tokio",
         "//eden/mononoke/common/mononoke_macros:mononoke_macros",
+        "//eden/mononoke/derived_data:fsnodes",
         "//eden/mononoke/features/async_requests:requests_table",
+        "//eden/mononoke/repo_attributes/bonsai_hg_mapping:bonsai_hg_mapping",
+        "//eden/mononoke/repo_attributes/bookmarks:bookmarks",
+        "//eden/mononoke/repo_attributes/filestore:filestore",
+        "//eden/mononoke/repo_attributes/repo_derivation_queues/if:derivation_queue_thrift-rust",
+        "//eden/mononoke/repo_attributes/repo_identity:repo_identity",
+        "//eden/mononoke/repo_factory:test_repo_factory",
         "//eden/mononoke/scs/if:source_control-rust",
+        "//eden/mononoke/tests/utils:tests_utils",
     ],
     deps = [
         "fbsource//third-party/rust:anyhow",

eden/mononoke/features/async_requests/worker_lib/Cargo.toml

@@ -46,5 +46,14 @@ tokio = { version = "1.52.3", features = ["full", "test-util", "tracing"] }
 tracing = { version = "0.1.41", features = ["attributes", "valuable"] }
 
 [dev-dependencies]
+bonsai_hg_mapping = { version = "0.1.0", path = "../../../repo_attributes/bonsai_hg_mapping" }
+bookmarks = { version = "0.1.0", path = "../../../repo_attributes/bookmarks" }
+derivation_queue_thrift = { version = "0.1.0", path = "../../../repo_attributes/repo_derivation_queues/if" }
+facet = { version = "0.1.0", git = "https://github.com/facebookexperimental/rust-shed.git", branch = "main", features = ["impl_never_type"] }
 fbinit = { version = "0.2.0", git = "https://github.com/facebookexperimental/rust-shed.git", branch = "main" }
 fbinit-tokio = { version = "0.1.2", git = "https://github.com/facebookexperimental/rust-shed.git", branch = "main" }
+filestore = { version = "0.1.0", path = "../../../repo_attributes/filestore" }
+fsnodes = { version = "0.1.0", path = "../../../derived_data/fsnodes" }
+repo_identity = { version = "0.1.0", path = "../../../repo_attributes/repo_identity" }
+test_repo_factory = { version = "0.1.0", path = "../../../repo_factory/test_repo_factory" }
+tests_utils = { version = "0.1.0", path = "../../../tests/utils" }

eden/mononoke/features/async_requests/worker_lib/src/backfill.rs

@@ -64,6 +64,18 @@ const JK_BACKFILL_CHUNK_RETRY_LIMIT: &str = "scm/mononoke:derived_data_backfill_
 const CHUNK_RETRY_BASE_BACKOFF: Duration = Duration::from_secs(1);
 const CHUNK_RETRY_MAX_BACKOFF: Duration = Duration::from_secs(30);
 
+/// JustKnob gating opportunistic boundary completion: when enabled, each
+/// boundary's predecessor derivation races against the boundary being derived
+/// by another task (e.g. a concurrent slice), so a slow predecessor derivation
+/// is abandoned as soon as it becomes redundant.
+const JK_BACKFILL_OPPORTUNISTIC_BOUNDARIES: &str =
+    "scm/mononoke:derived_data_backfill_opportunistic_boundary_completion";
+
+/// Poll interval for noticing that a boundary changeset has been derived by
+/// another task (e.g. a concurrently-running slice), so we can abandon its
+/// now-redundant predecessor derivation.
+const BOUNDARY_DERIVED_POLL_INTERVAL: Duration = Duration::from_secs(1);
+
 /// Get a DerivedDataManager with optional read/write throttling applied.
 /// If JustKnobs are not set or are zero, returns the manager unchanged.
 fn get_throttled_manager(manager: &DerivedDataManager) -> Result<DerivedDataManager> {
@@ -137,6 +149,27 @@ fn requires_serial_slice_processing(derived_data_type: DerivableType) -> bool {
         .is_err()
 }
 
+/// Resolves once `csid` has `derived_data_type` derived by any means, polling
+/// periodically. Used to race against predecessor derivation: if a concurrent
+/// slice derives the boundary first, we can abandon the slower
+/// `derive_from_predecessor` call. Transient lookup errors are ignored and
+/// retried, since predecessor derivation remains the guaranteed backstop.
+async fn wait_until_derived(
+    manager: &DerivedDataManager,
+    ctx: &CoreContext,
+    csid: ChangesetId,
+    derived_data_type: DerivableType,
+) {
+    loop {
+        tokio::time::sleep(BOUNDARY_DERIVED_POLL_INTERVAL).await;
+        if let Ok(true) =
+            BulkDerivation::is_derived(manager, ctx, csid, None, derived_data_type).await
+        {
+            return;
+        }
+    }
+}
+
 /// Compute derive_boundaries request - derives boundary changesets using predecessor derivation
 pub(crate) async fn compute_derive_boundaries(
     ctx: &CoreContext,
@@ -189,6 +222,7 @@ pub(crate) async fn compute_derive_boundaries(
     let manager = with_type_enabled(&manager, derived_data_type);
     let concurrency = params.concurrency.max(1) as usize;
     let use_predecessor = params.use_predecessor_derivation;
+    let opportunistic = justknobs::eval(JK_BACKFILL_OPPORTUNISTIC_BOUNDARIES, None, None);
 
     stream::iter(boundary_cs_ids)
         .map(Ok::<_, Error>)
@@ -198,14 +232,29 @@ pub(crate) async fn compute_derive_boundaries(
             let derived_count = derived_count.clone();
             async move {
                 if use_predecessor {
-                    BulkDerivation::unsafe_derive_untopologically(
+                    let derive = BulkDerivation::unsafe_derive_untopologically(
                         &manager,
                         &ctx,
                         csid,
                         None, // rederivation
                         derived_data_type,
-                    )
-                    .await?;
+                    );
+                    if opportunistic {
+                        // Race predecessor derivation against the boundary becoming
+                        // derived by another task (e.g. a concurrently-running slice
+                        // whose head is this boundary). Whichever happens first
+                        // satisfies the boundary; if it's derived elsewhere we drop
+                        // the now-redundant (and possibly slow) predecessor future.
+                        tokio::select! {
+                            biased;
+                            res = derive => {
+                                res?;
+                            }
+                            _ = wait_until_derived(&manager, &ctx, csid, derived_data_type) => {}
+                        }
+                    } else {
+                        derive.await?;
+                    }
                 } else {
                     manager
                         .derive_bulk_locally(
@@ -808,3 +857,78 @@ async fn process_repo_backfill(
     )
     .await
 }
+
+#[cfg(test)]
+mod tests {
+    use bonsai_hg_mapping::BonsaiHgMapping;
+    use bookmarks::Bookmarks;
+    use commit_graph::CommitGraph;
+    use commit_graph::CommitGraphWriter;
+    use context::CoreContext;
+    use derivation_queue_thrift::DerivationPriority;
+    use fbinit::FacebookInit;
+    use filestore::FilestoreConfig;
+    use fsnodes::RootFsnodeId;
+    use mononoke_macros::mononoke;
+    use repo_blobstore::RepoBlobstore;
+    use repo_derived_data::RepoDerivedData;
+    use repo_derived_data::RepoDerivedDataRef;
+    use repo_identity::RepoIdentity;
+    use tests_utils::CreateCommitContext;
+
+    use super::*;
+
+    #[facet::container]
+    struct TestRepo(
+        dyn BonsaiHgMapping,
+        dyn Bookmarks,
+        CommitGraph,
+        dyn CommitGraphWriter,
+        RepoDerivedData,
+        RepoBlobstore,
+        FilestoreConfig,
+        RepoIdentity,
+    );
+
+    /// `wait_until_derived` must observe a changeset transitioning to derived
+    /// (by any path) and resolve. This is the only non-trivial bit of the
+    /// opportunistic-boundary race: the `select!` itself is plumbing, but this
+    /// poll has to actually detect real derivation against a real manager.
+    #[mononoke::fbinit_test]
+    async fn wait_until_derived_resolves_after_derivation(fb: FacebookInit) -> Result<()> {
+        let ctx = CoreContext::test_mock(fb);
+        let repo: TestRepo = test_repo_factory::build_empty(fb)
+            .await
+            .map_err(|e| anyhow::anyhow!("{e}"))?;
+        let cs_id = CreateCommitContext::new_root(&ctx, &repo)
+            .add_file("a", "a")
+            .commit()
+            .await?;
+        let manager = repo.repo_derived_data().manager();
+
+        // Underived to begin with.
+        assert!(
+            !BulkDerivation::is_derived(manager, &ctx, cs_id, None, DerivableType::Fsnodes).await?,
+            "changeset should start underived",
+        );
+
+        // Derive concurrently with the waiter. The waiter must observe the
+        // transition and resolve; if it never did, this join would hang and the
+        // test would time out.
+        tokio::join!(
+            async {
+                repo.repo_derived_data()
+                    .derive::<RootFsnodeId>(&ctx, cs_id, DerivationPriority::LOW)
+                    .await
+                    .expect("derivation should succeed");
+            },
+            wait_until_derived(manager, &ctx, cs_id, DerivableType::Fsnodes),
+        );
+
+        assert!(
+            BulkDerivation::is_derived(manager, &ctx, cs_id, None, DerivableType::Fsnodes).await?,
+            "changeset should be derived after the waiter resolved",
+        );
+        Ok(())
+    }
+}