mononoke: basic infinitepush implementation

Summary:
Support b2x:infinitepush part. It contains changegroup v2, so just reuse the
normal push path for it.
Note that pushbackup still fails because we don't support
b2x:infinitepushscratchbookmarks part. Also all wireproto params are ignored
for b2x:infinitepush.

Reviewed By: farnz

Differential Revision: D7086120

fbshipit-source-id: 2f98e5d59059ca3c2b82842c98e6dc771c70c6f0

bundle2-resolver/src/resolver.rs

@@ -122,7 +122,8 @@ impl Bundle2Resolver {
 
         next_item(bundle2)
             .and_then(move |(changegroup, bundle2)| match changegroup {
-                Some(Bundle2Item::Changegroup(header, parts)) => {
+                Some(Bundle2Item::Changegroup(header, parts))
+                | Some(Bundle2Item::B2xInfinitepush(header, parts)) => {
                     let part_id = header.part_id();
                     let (c, f) = split_changegroup(parts);
                     convert_to_revlog_changesets(c)

mercurial-bundles/src/lib.rs

@@ -83,6 +83,7 @@ pub use types::StreamHeader;
 pub enum Bundle2Item {
     Start(StreamHeader),
     Changegroup(PartHeader, BoxStream<changegroup::Part, Error>),
+    B2xInfinitepush(PartHeader, BoxStream<changegroup::Part, Error>),
     B2xTreegroup2(PartHeader, BoxStream<wirepack::Part, Error>),
     Replycaps(PartHeader, BoxFuture<capabilities::Capabilities, Error>),
 }
@@ -110,6 +111,9 @@ impl fmt::Debug for Bundle2Item {
         match self {
             &Start(ref header) => write!(f, "Bundle2Item::Start({:?})", header),
             &Changegroup(ref header, _) => write!(f, "Bundle2Item::Changegroup({:?}, ...)", header),
+            &B2xInfinitepush(ref header, _) => {
+                write!(f, "Bundle2Item::B2xInfinitepush({:?}, ...)", header)
+            }
             &B2xTreegroup2(ref header, _) => {
                 write!(f, "Bundle2Item::B2xTreegroup2({:?}, ...)", header)
             }

mercurial-bundles/src/part_header.rs

@@ -34,6 +34,8 @@ pub enum PartHeaderType {
     /// Mononoke this parameter will be ignored, because it does not provide transaction during
     /// push
     CheckHeads,
+    /// Contains changegroup for infinitepush commits
+    B2xInfinitepush,
     // RemoteChangegroup,       // We don't wish to support this functionality
     // CheckBookmarks,          // TODO Do we want to support this?
     // CheckHeads,              // TODO Do we want to support this?
@@ -63,6 +65,7 @@ impl PartHeaderType {
             "replycaps" => Ok(Replycaps),
             "listkeys" => Ok(Listkeys),
             "b2x:treegroup2" => Ok(B2xTreegroup2),
+            "b2x:infinitepush" => Ok(B2xInfinitepush),
             "check:heads" => Ok(CheckHeads),
             bad => bail_msg!("unknown header type {}", bad),
         }
@@ -76,6 +79,7 @@ impl PartHeaderType {
             Replycaps => "replycaps",
             Listkeys => "listkeys",
             B2xTreegroup2 => "b2x:treegroup2",
+            B2xInfinitepush => "b2x:infinitepush",
             CheckHeads => "check:heads",
         }
     }

mercurial-bundles/src/part_inner.rs

@@ -33,6 +33,12 @@ lazy_static! {
     static ref KNOWN_PARAMS: HashMap<PartHeaderType, HashSet<&'static str>> = {
         let mut m: HashMap<PartHeaderType, HashSet<&'static str>> = HashMap::new();
         m.insert(PartHeaderType::Changegroup, hashset!{"version", "nbchanges", "treemanifest"});
+        // TODO(stash): currently ignore all the parameters. Later we'll
+        // support 'bookmark' parameter, and maybe 'create' and 'force' (although 'force' will
+        // probably) be renamed T26385545. 'bookprevnode' and 'pushbackbookmarks' will be
+        // removed T26384190.
+        m.insert(PartHeaderType::B2xInfinitepush, hashset!{
+            "pushbackbookmarks", "cgversion", "bookmark", "bookprevnode", "create", "force"});
         m.insert(PartHeaderType::B2xTreegroup2, hashset!{"version", "cache", "category"});
         m.insert(PartHeaderType::Replycaps, hashset!{});
         m
@@ -84,6 +90,12 @@ pub fn inner_stream<R: AsyncRead + BufRead + 'static + Send>(
             ));
             Bundle2Item::Changegroup(header, Box::new(cg2_stream))
         }
+        &PartHeaderType::B2xInfinitepush => {
+            let cg2_stream = wrapped_stream.decode(changegroup::unpacker::Cg2Unpacker::new(
+                logger.new(o!("stream" => "cg2")),
+            ));
+            Bundle2Item::B2xInfinitepush(header, Box::new(cg2_stream))
+        }
         &PartHeaderType::B2xTreegroup2 => {
             let wirepack_stream = wrapped_stream.decode(wirepack::unpacker::new(
                 logger.new(o!("stream" => "wirepack")),

tests/integration/test-infinitepush.t

@@ -1,51 +1,21 @@
   $ . $TESTDIR/library.sh
 
 setup configuration
-  $ setup_config_repo
-  $ cd $TESTTMP
-
-setup common configuration
-  $ cat >> $HGRCPATH <<EOF
-  > [ui]
-  > ssh="$DUMMYSSH"
-  > [extensions]
-  > remotefilelog=
-  > [remotefilelog]
-  > cachepath=$TESTTMP/cachepath
-  > EOF
-
+  $ setup_common_config
 
 setup repo
 
-  $ hg init repo-hg
-
-Init treemanifest and remotefilelog
+  $ hginit_treemanifest repo-hg
   $ cd repo-hg
-  $ cat >> .hg/hgrc <<EOF
-  > [extensions]
-  > treemanifest=
-  > remotefilelog=
-  > [treemanifest]
-  > server=True
-  > [remotefilelog]
-  > server=True
-  > shallowtrees=True
-  > EOF
-
-  $ touch a
-  $ hg add a
-  $ hg ci -ma
-  $ hg log
-  changeset:   0:3903775176ed
-  tag:         tip
-  user:        test
-  date:        Thu Jan 01 00:00:00 1970 +0000
-  summary:     a
-  
+  $ touch a && hg addremove && hg ci -q -ma
+  adding a
+  $ hg log -T '{node}\n'
+  3903775176ed42b1458a6281db4a0ccf4d9f287a
   $ cd $TESTTMP
 
-setup repo2
-  $ hgclone_treemanifest ssh://user@dummy/repo-hg repo2 --noupdate
+setup repo-push and repo-pull
+  $ hgclone_treemanifest ssh://user@dummy/repo-hg repo-push --noupdate
+  $ hgclone_treemanifest ssh://user@dummy/repo-hg repo-pull --noupdate
 
   $ blobimport --blobstore files --linknodes repo-hg repo
 
@@ -54,7 +24,9 @@ start mononoke
   $ mononoke -P $TESTTMP/mononoke-config -B test-config
   $ wait_for_mononoke $TESTTMP/repo
 
-  $ cd repo2
+
+Do infinitepush (aka commit cloud) push
+  $ cd repo-push
   $ cat >> .hg/hgrc <<EOF
   > [extensions]
   > infinitepush=
@@ -87,11 +59,26 @@ start mononoke
   bundle2-output-part: "replycaps" 250 bytes payload
   bundle2-output-part: "B2X:INFINITEPUSH" (params: 0 advisory) streamed payload
   bundle2-output-part: "b2x:treegroup2" (params: 3 mandatory) streamed payload
-  * unknown header type b2x:infinitepush, backtrace:* (glob)
-  abort: stream ended unexpectedly (got 0 bytes, expected 4)
-  [255]
+  bundle2-input-bundle: 1 params no-transaction
+  bundle2-input-part: "reply:changegroup" (params: 2 mandatory) supported
+  bundle2-input-bundle: 0 parts total
+
+  $ cd ../repo-pull
+  $ hgmn pull
+  pulling from ssh://user@dummy/repo
+  searching for changes
+  adding changesets
+  adding manifests
+  adding file changes
+  added 1 changesets with 0 changes to 0 files
+  new changesets 47da8b81097c
+  (run 'hg update' to get a working copy)
+  $ hgmn up -q 47da8b81097c
+  $ cat newfile
+  new
 
 Pushbackup fails too
+  $ cd ../repo-push
   $ hgmn pushbackup ssh://user@dummy/repo --debug
   starting backup* (glob)
   running * (glob)
@@ -104,16 +91,11 @@ Pushbackup fails too
   sending batch command
   searching for changes
   all remote heads known locally
-  1 changesets found
-  list of changesets:
-  47da8b81097c5534f3eb7947a8764dd323cffe3d
   sending unbundle command
-  bundle2-output-bundle: "HG20", (1 params) 4 parts total
+  bundle2-output-bundle: "HG20", (1 params) 2 parts total
   bundle2-output-part: "replycaps" 250 bytes payload
-  bundle2-output-part: "B2X:INFINITEPUSH" (params: 0 advisory) streamed payload
-  bundle2-output-part: "b2x:treegroup2" (params: 3 mandatory) streamed payload
-  bundle2-output-part: "B2X:INFINITEPUSHSCRATCHBOOKMARKS" 459 bytes payload
-  * unknown header type b2x:infinitepush, backtrace:* (glob)
+  bundle2-output-part: "B2X:INFINITEPUSHSCRATCHBOOKMARKS" * bytes payload (glob)
+  * unknown header type b2x:infinitepushscratchbookmarks* (glob)
   finished in * seconds (glob)
   abort: stream ended unexpectedly (got 0 bytes, expected 4)
   [255]