mononoke_api: add changeset history

Summary:
Add a method to get the history of a changeset.  This differs from the history
of a changeset path, even the history of the root directory, in that all
changesets are included, even empty ones.

Differential Revision: D21179877

fbshipit-source-id: e19aac75fc40d8e9a3beb134e16a8cdfe882b791

eden/mononoke/mononoke_api/src/changeset.rs

@@ -5,7 +5,7 @@
  * GNU General Public License version 2.
  */
 
-use std::collections::{BTreeMap, HashMap, HashSet};
+use std::collections::{BTreeMap, HashMap, HashSet, VecDeque};
 use std::convert::TryInto;
 use std::fmt;
 use std::future::Future;
@@ -20,10 +20,10 @@ use context::CoreContext;
 use derived_data::BonsaiDerived;
 use fsnodes::RootFsnodeId;
 use futures::compat::{Future01CompatExt, Stream01CompatExt};
-use futures::stream::Stream;
-use futures_util::future::{self, try_join, FutureExt, Shared};
-use futures_util::stream::{StreamExt, TryStreamExt};
+use futures::future::{self, try_join, FutureExt, Shared};
+use futures::stream::{self, Stream, StreamExt, TryStreamExt};
 use manifest::{Diff as ManifestDiff, Entry as ManifestEntry, ManifestOps, PathOrPrefix};
+use maplit::hashset;
 use mercurial_types::Globalrev;
 pub use mononoke_types::Generation;
 use mononoke_types::{BonsaiChangeset, FileChange, MPath, MPathElement};
@@ -573,4 +573,65 @@ impl ChangesetContext {
             .map_ok(|mpath| MononokePath::new(Some(mpath)))
             .map_err(MononokeError::from))
     }
+
+    /// Returns a stream of `ChangesetContext` for the history of the repository from this commit.
+    pub async fn history(
+        &self,
+        until_timestamp: Option<i64>,
+    ) -> impl Stream<Item = Result<ChangesetContext, MononokeError>> + '_ {
+        let cs_info_enabled = self.repo.derive_changeset_info_enabled();
+
+        let terminate = until_timestamp.map(move |until_timestamp| {
+            move |changeset_id| async move {
+                let info = if cs_info_enabled {
+                    ChangesetInfo::derive(
+                        self.ctx().clone(),
+                        self.repo().blob_repo().clone(),
+                        changeset_id,
+                    )
+                    .compat()
+                    .await?
+                } else {
+                    let bonsai = changeset_id
+                        .load(self.ctx().clone(), self.repo().blob_repo().blobstore())
+                        .compat()
+                        .await?;
+                    ChangesetInfo::new(changeset_id, bonsai)
+                };
+                let date = info.author_date().as_chrono().clone();
+                Ok::<_, MononokeError>(date.timestamp() < until_timestamp)
+            }
+        });
+
+        stream::try_unfold(
+            // starting state
+            (hashset! { self.id() }, VecDeque::from(vec![self.id()])),
+            // unfold
+            move |(mut visited, mut queue)| async move {
+                if let Some(changeset_id) = queue.pop_front() {
+                    if let Some(terminate) = terminate {
+                        if terminate(changeset_id).await? {
+                            return Ok(Some((None, (visited, queue))));
+                        }
+                    }
+                    let parents = self
+                        .repo()
+                        .blob_repo()
+                        .get_changeset_parents_by_bonsai(self.ctx().clone(), changeset_id)
+                        .compat()
+                        .await?;
+                    queue.extend(parents.into_iter().filter(|parent| visited.insert(*parent)));
+                    Ok(Some((Some(changeset_id), (visited, queue))))
+                } else {
+                    Ok::<_, MononokeError>(None)
+                }
+            },
+        )
+        .try_filter_map(move |changeset_id| {
+            let changeset = changeset_id
+                .map(|changeset_id| ChangesetContext::new(self.repo().clone(), changeset_id));
+            async move { Ok::<_, MononokeError>(changeset) }
+        })
+        .boxed()
+    }
 }

eden/mononoke/mononoke_api/src/changeset_path.rs

@@ -274,10 +274,10 @@ impl ChangesetPathContext {
 
     /// Returns a list of `ChangesetContext` for the file at this path that represents
     /// a history of the path.
-    pub async fn history<'a>(
-        &'a self,
+    pub async fn history(
+        &self,
         until_timestamp: Option<i64>,
-    ) -> Result<impl Stream<Item = Result<ChangesetContext, MononokeError>> + 'a, MononokeError>
+    ) -> Result<impl Stream<Item = Result<ChangesetContext, MononokeError>> + '_, MononokeError>
     {
         let ctx = self.changeset.ctx().clone();
         let repo = self.repo().blob_repo().clone();

eden/mononoke/mononoke_api/src/test/test_history.rs

@@ -285,3 +285,84 @@ async fn commit_path_history(fb: FacebookInit) -> Result<()> {
 
     Ok(())
 }
+
+#[fbinit::compat_test]
+async fn commit_history(fb: FacebookInit) -> Result<()> {
+    let ctx = CoreContext::test_mock(fb);
+    let (repo, changesets) = init_repo(&ctx).await?;
+
+    let cs = repo
+        .changeset(ChangesetSpecifier::Bonsai(changesets["c2"]))
+        .await?
+        .expect("changeset exists");
+
+    // The commit history includes all commits, including empty ones.
+    let history: Vec<_> = cs
+        .history(None)
+        .await
+        .and_then(|cs| async move { Ok(cs.id()) })
+        .try_collect()
+        .await?;
+    assert_eq!(
+        history,
+        vec![
+            changesets["c2"],
+            changesets["m2"],
+            changesets["e2"],
+            changesets["e3"],
+            changesets["a4"],
+            changesets["b3"],
+            changesets["c1"],
+            changesets["e1"],
+            changesets["m1"],
+            changesets["b2"],
+            changesets["a3"],
+            changesets["b1"],
+            changesets["a2"],
+            changesets["a1"],
+        ]
+    );
+
+    // The commit history of an empty commit starts with itself.
+    let cs = repo
+        .changeset(ChangesetSpecifier::Bonsai(changesets["e1"]))
+        .await?
+        .expect("changeset exists");
+    let history: Vec<_> = cs
+        .history(None)
+        .await
+        .and_then(|cs| async move { Ok(cs.id()) })
+        .try_collect()
+        .await?;
+    assert_eq!(
+        history,
+        vec![
+            changesets["e1"],
+            changesets["m1"],
+            changesets["b2"],
+            changesets["a3"],
+            changesets["b1"],
+            changesets["a2"],
+            changesets["a1"],
+        ]
+    );
+
+    // Setting until_timestamp omits some commits.
+    let history: Vec<_> = cs
+        .history(Some(2500))
+        .await
+        .and_then(|cs| async move { Ok(cs.id()) })
+        .try_collect()
+        .await?;
+    assert_eq!(
+        history,
+        vec![
+            changesets["e1"],
+            changesets["m1"],
+            changesets["b2"],
+            changesets["a3"],
+        ]
+    );
+
+    Ok(())
+}