index: make has_id() async

lib/src/commit_builder.rs

@@ -405,6 +405,7 @@ impl DetachedCommitBuilder {
             && mut_repo
                 .index()
                 .has_id(commit.id())
+                .await
                 // TODO: indexing error shouldn't be a "BackendError"
                 .map_err(|err| BackendError::Other(err.into()))?
         {

lib/src/default_index/composite.rs

@@ -610,7 +610,7 @@ impl Index for CompositeIndex {
         Ok(self.commits().resolve_commit_id_prefix(prefix))
     }
 
-    fn has_id(&self, commit_id: &CommitId) -> IndexResult<bool> {
+    async fn has_id(&self, commit_id: &CommitId) -> IndexResult<bool> {
         Ok(self.commits().has_id(commit_id))
     }
 

lib/src/default_index/mutable.rs

@@ -562,8 +562,8 @@ impl Index for DefaultMutableIndex {
         self.0.resolve_commit_id_prefix(prefix)
     }
 
-    fn has_id(&self, commit_id: &CommitId) -> IndexResult<bool> {
-        self.0.has_id(commit_id)
+    async fn has_id(&self, commit_id: &CommitId) -> IndexResult<bool> {
+        self.0.has_id(commit_id).await
     }
 
     async fn is_ancestor(

lib/src/default_index/readonly.rs

@@ -729,7 +729,7 @@ impl Index for DefaultReadonlyIndex {
         self.0.resolve_commit_id_prefix(prefix)
     }
 
-    fn has_id(&self, commit_id: &CommitId) -> IndexResult<bool> {
+    async fn has_id(&self, commit_id: &CommitId) -> IndexResult<bool> {
         Ok(self.has_id_impl(commit_id))
     }
 

lib/src/git.rs

@@ -30,6 +30,7 @@ use bstr::BStr;
 use bstr::BString;
 use futures::StreamExt as _;
 use futures::TryStreamExt as _;
+use futures::stream;
 use gix::refspec::Instruction;
 use itertools::Itertools as _;
 use thiserror::Error;
@@ -650,14 +651,16 @@ async fn import_refs_inner(
     // changed_git_refs aren't respected because changed_remote_bookmarks/tags
     // should include all heads that will become reachable in jj.
     let index = mut_repo.index();
-    let missing_head_ids: Vec<&CommitId> = new_referenced_heads
-        .iter()
-        .filter_map(|id| match index.has_id(id) {
-            Ok(false) => Some(Ok(id)),
-            Ok(true) => None,
-            Err(e) => Some(Err(e)),
+    let missing_head_ids: Vec<&CommitId> = stream::iter(&new_referenced_heads)
+        .map(async move |id| (id, index.has_id(id).await))
+        .buffered(mut_repo.store().concurrency())
+        .filter_map(async move |m| match m {
+            (id, Ok(false)) => Some(Ok(id)),
+            (_, Ok(true)) => None,
+            (_, Err(err)) => Some(Err(GitImportError::Index(err))),
         })
-        .try_collect()?;
+        .try_collect()
+        .await?;
     let heads_imported = git_backend.import_head_commits(missing_head_ids).is_ok();
 
     // Import new remote heads
@@ -668,7 +671,7 @@ async fn import_refs_inner(
             err,
         };
         // If bulk-import failed, try again to find bad head or ref.
-        if !heads_imported && !index.has_id(id)? {
+        if !heads_imported && !index.has_id(id).await? {
             git_backend
                 .import_head_commits([id])
                 .map_err(missing_ref_err)?;
@@ -1184,7 +1187,7 @@ pub async fn import_head(mut_repo: &mut MutableRepo) -> Result<(), GitImportErro
     // Import new head
     if let Some(head_id) = &new_git_head_id {
         let index = mut_repo.index();
-        if !index.has_id(head_id)? {
+        if !index.has_id(head_id).await? {
             git_backend.import_head_commits([head_id]).map_err(|err| {
                 GitImportError::MissingHeadTarget {
                     id: head_id.clone(),

lib/src/id_prefix.rs

@@ -174,7 +174,7 @@ impl IdPrefixIndex<'_> {
                 PrefixResolution::SingleMatch(id) => {
                     // The disambiguation set may be loaded from a different repo,
                     // and contain a commit that doesn't exist in the current repo.
-                    if repo.index().has_id(&id)? {
+                    if repo.index().has_id(&id).block_on()? {
                         return Ok(PrefixResolution::SingleMatch(id));
                     } else {
                         return Ok(PrefixResolution::NoMatch);

lib/src/index.rs

@@ -116,7 +116,7 @@ pub trait Index: Send + Sync {
     ) -> IndexResult<PrefixResolution<CommitId>>;
 
     /// Returns true if `commit_id` is present in the index.
-    fn has_id(&self, commit_id: &CommitId) -> IndexResult<bool>;
+    async fn has_id(&self, commit_id: &CommitId) -> IndexResult<bool>;
 
     /// Returns true if `ancestor_id` commit is an ancestor of the
     /// `descendant_id` commit, or if `ancestor_id` equals `descendant_id`.

lib/src/repo.rs

@@ -1781,19 +1781,23 @@ impl MutableRepo {
     /// Adds the given `heads` and ancestor commits to the index without making
     /// them visible. Returns newly-indexed commits.
     pub async fn index_commits(&mut self, heads: &[Commit]) -> BackendResult<Vec<Commit>> {
+        let index = self.index();
+        let missing_heads: Vec<_> = stream::iter(heads)
+            .map(async move |commit| (commit, index.has_id(commit.id()).await))
+            .buffered(self.store().concurrency())
+            .filter_map(async |m| match m {
+                (commit, Ok(false)) => Some(Ok(CommitByCommitterTimestamp(commit.clone()))),
+                (_, Ok(true)) => None,
+                (_, Err(err)) => Some(Err(BackendError::Other(err.into()))),
+            })
+            .collect()
+            .await;
         let missing_commits = dag_walk_async::topo_order_reverse_ord(
-            heads
-                .iter()
-                .filter_map(|commit| match self.index().has_id(commit.id()) {
-                    Ok(false) => Some(Ok(CommitByCommitterTimestamp(commit.clone()))),
-                    Ok(true) => None,
-                    // TODO: indexing error shouldn't be a "BackendError"
-                    Err(err) => Some(Err(BackendError::Other(err.into()))),
-                }),
+            missing_heads,
             |CommitByCommitterTimestamp(commit)| commit.id().clone(),
             async |CommitByCommitterTimestamp(commit)| {
                 stream::iter(commit.parent_ids())
-                    .filter_map(async |id| match self.index().has_id(id) {
+                    .filter_map(async |id| match index.has_id(id).await {
                         Ok(false) => Some(
                             self.store()
                                 .get_commit_async(id)

lib/tests/test_git.rs

@@ -6274,7 +6274,7 @@ fn test_concurrent_write_commit() -> TestResult {
 
     // The index should be consistent with the store.
     for commit_id in commit_change_ids.keys() {
-        assert!(repo.index().has_id(commit_id)?);
+        assert!(repo.index().has_id(commit_id).block_on()?);
         let commit = repo.store().get_commit(commit_id)?;
         assert_eq!(
             repo.resolve_change_id(commit.change_id())?
@@ -6401,7 +6401,7 @@ fn test_concurrent_read_write_commit() -> TestResult {
     // The index should be consistent with the store.
     let repo = repo.reload_at_head().block_on()?;
     for commit_id in &commit_ids {
-        assert!(repo.index().has_id(commit_id)?);
+        assert!(repo.index().has_id(commit_id).block_on()?);
         let commit = repo.store().get_commit(commit_id)?;
         assert_eq!(
             repo.resolve_change_id(commit.change_id())?
@@ -6525,7 +6525,7 @@ fn test_shallow_commits_lack_parents() -> TestResult {
         "unshallowed commits have correct parents"
     );
     // FIXME: new ancestors should be indexed
-    assert!(!repo.index().has_id(&jj_id(a))?);
+    assert!(!repo.index().has_id(&jj_id(a)).block_on()?);
     Ok(())
 }
 

lib/tests/test_index.rs

@@ -85,7 +85,7 @@ fn collect_changed_paths(repo: &ReadonlyRepo, commit_id: &CommitId) -> Option<Ve
 }
 
 fn index_has_id(index: &dyn Index, commit_id: &CommitId) -> bool {
-    index.has_id(commit_id).unwrap()
+    index.has_id(commit_id).block_on().unwrap()
 }
 
 fn is_ancestor(

lib/tests/test_mut_repo.rs

@@ -50,7 +50,7 @@ where
 }
 
 fn index_has_id(index: &dyn Index, commit_id: &CommitId) -> bool {
-    index.has_id(commit_id).unwrap()
+    index.has_id(commit_id).block_on().unwrap()
 }
 
 #[test]

lib/tests/test_operations.rs

@@ -61,7 +61,7 @@ fn list_dir(dir: &Path) -> Vec<String> {
 }
 
 fn index_has_id(index: &dyn Index, commit_id: &CommitId) -> bool {
-    index.has_id(commit_id).unwrap()
+    index.has_id(commit_id).block_on().unwrap()
 }
 
 #[test]