use commit_graph in find_draft_ancestors

Summary:
This method is really slow for pushes of thousands of linear commits (chromium/src).

For push of 1k commits to chromium/src we spent ~26 minutes in this function:
 {F1973662376}

See the more details about the experiment here: https://docs.google.com/document/d/1FojOm-eherq0RtCw2gB435SuW1_Xc031K7m0mEt1Y9k/edit?tab=t.0#heading=h.17zk7p1g8o1d

Differential Revision: D66977548

fbshipit-source-id: 71ee1991de496557ec54610e8857345882b66a12

eden/mononoke/features/repo_update_logger/Cargo.toml

@@ -15,11 +15,13 @@ bonsai_git_mapping = { version = "0.1.0", path = "../../bonsai_git_mapping" }
 bonsai_globalrev_mapping = { version = "0.1.0", path = "../../bonsai_globalrev_mapping" }
 bookmarks = { version = "0.1.0", path = "../../bookmarks" }
 bookmarks_types = { version = "0.1.0", path = "../../bookmarks/bookmarks_types" }
+borrowed = { version = "0.1.0", git = "https://github.com/facebookexperimental/rust-shed.git", branch = "main" }
 chrono = { version = "0.4", features = ["clock", "serde", "std"], default-features = false }
 commit_graph = { version = "0.1.0", path = "../../repo_attributes/commit_graph/commit_graph" }
 context = { version = "0.1.0", path = "../../server/context" }
 ephemeral_blobstore = { version = "0.1.0", path = "../../blobstore/ephemeral_blobstore" }
 futures = { version = "0.3.30", features = ["async-await", "compat"] }
+futures_ext = { version = "0.1.0", git = "https://github.com/facebookexperimental/rust-shed.git", branch = "main" }
 futures_stats = { version = "0.1.0", git = "https://github.com/facebookexperimental/rust-shed.git", branch = "main" }
 git_source_of_truth = { version = "0.1.0", path = "../../git_source_of_truth" }
 gix-hash = "0.15.1"

eden/mononoke/features/repo_update_logger/src/commit_logger.rs

@@ -5,8 +5,6 @@
  * GNU General Public License version 2.
  */
 
-use std::collections::HashSet;
-use std::collections::VecDeque;
 use std::num::NonZeroU64;
 
 use anyhow::anyhow;
@@ -17,6 +15,7 @@ use blobstore::Loadable;
 use bonsai_globalrev_mapping::BonsaiGlobalrevMappingRef;
 use bookmarks_types::BookmarkKey;
 use bookmarks_types::BookmarkKind;
+use borrowed::borrowed;
 use chrono::DateTime;
 use chrono::Utc;
 use commit_graph::CommitGraphRef;
@@ -25,6 +24,7 @@ use ephemeral_blobstore::BubbleId;
 use futures::stream;
 use futures::StreamExt;
 use futures::TryStreamExt;
+use futures_ext::stream::FbStreamExt;
 use futures_stats::TimedTryFutureExt;
 use logger_ext::Loggable;
 use metaconfig_types::RepoConfigRef;
@@ -353,43 +353,42 @@ pub async fn find_draft_ancestors(
          + BonsaiGlobalrevMappingRef
          + CommitGraphRef
          + RepoBlobstoreRef
+         + std::marker::Sync
      ),
     to_cs_id: ChangesetId,
 ) -> Result<Vec<BonsaiChangeset>, Error> {
     ctx.scuba()
         .clone()
         .log_with_msg("Started finding draft ancestors", None);
-
     let (stats, drafts) = async move {
-        let phases = repo.phases();
-        let mut queue = VecDeque::new();
-        let mut visited = HashSet::new();
-        let mut drafts = vec![];
-        queue.push_back(to_cs_id);
-        visited.insert(to_cs_id);
-
-        while let Some(cs_id) = queue.pop_front() {
-            let public = phases
-                .get_public(ctx, vec![cs_id], false /*ephemeral_derive*/)
-                .await?;
-
-            if public.contains(&cs_id) {
-                continue;
-            }
-            drafts.push(cs_id);
-
-            let parents = repo.commit_graph().changeset_parents(ctx, cs_id).await?;
-            for p in parents {
-                if visited.insert(p) {
-                    queue.push_back(p);
+        let public_frontier: Vec<ChangesetId> = repo
+            .commit_graph()
+            .ancestors_frontier_with(ctx, vec![to_cs_id], |csid| {
+                borrowed!(ctx, repo);
+                async move {
+                    Ok(repo
+                        .phases()
+                        .get_public(ctx, vec![csid], false)
+                        .await?
+                        .contains(&csid))
                 }
-            }
-        }
+            })
+            .await?
+            .into_iter()
+            .collect();
 
-        stream::iter(drafts)
-            .map(Ok)
-            .map_ok(|cs_id| async move { cs_id.load(ctx, repo.repo_blobstore()).await })
-            .try_buffer_unordered(100)
+        repo.commit_graph()
+            .ancestors_difference_stream(ctx, vec![to_cs_id], public_frontier)
+            .await?
+            .yield_periodically()
+            .map(move |res| async move {
+                match res {
+                    Ok(bcs_id) => Ok(bcs_id.load(ctx, repo.repo_blobstore()).await?),
+                    Err(e) => Err(e),
+                }
+            })
+            .buffered(1000)
+            .boxed()
             .try_collect::<Vec<_>>()
             .await
     }