getbundle: add support for 'bookmarks' boolean argument

This new argument requests a 'bookmarks' part from the server. It is meant to
be used instead of the "listkeys" request.

mercurial/exchange.py

@@ -1,2175 +1,2188 b''
 # exchange.py - utility to exchange data between repos.
 #
 # Copyright 2005-2007 Matt Mackall <[e-mail]>
 #
 # This software may be used and distributed according to the terms of the
 # GNU General Public License version 2 or any later version.
 
 from __future__ import absolute_import
 
 import collections
 import errno
 import hashlib
 
 from .i18n import _
 from .node import (
     bin,
     hex,
     nullid,
 )
 from . import (
     bookmarks as bookmod,
     bundle2,
     changegroup,
     discovery,
     error,
     lock as lockmod,
     obsolete,
     phases,
     pushkey,
     pycompat,
     remotenames,
     scmutil,
     sslutil,
     streamclone,
     url as urlmod,
     util,
 )
 
 urlerr = util.urlerr
 urlreq = util.urlreq
 
 # Maps bundle version human names to changegroup versions.
 _bundlespeccgversions = {'v1': '01',
                          'v2': '02',
                          'packed1': 's1',
                          'bundle2': '02', #legacy
                         }
 
 # Compression engines allowed in version 1. THIS SHOULD NEVER CHANGE.
 _bundlespecv1compengines = {'gzip', 'bzip2', 'none'}
 
 def parsebundlespec(repo, spec, strict=True, externalnames=False):
     """Parse a bundle string specification into parts.
 
     Bundle specifications denote a well-defined bundle/exchange format.
     The content of a given specification should not change over time in
     order to ensure that bundles produced by a newer version of Mercurial are
     readable from an older version.
 
     The string currently has the form:
 
        <compression>-<type>[;<parameter0>[;<parameter1>]]
 
     Where <compression> is one of the supported compression formats
     and <type> is (currently) a version string. A ";" can follow the type and
     all text afterwards is interpreted as URI encoded, ";" delimited key=value
     pairs.
 
     If ``strict`` is True (the default) <compression> is required. Otherwise,
     it is optional.
 
     If ``externalnames`` is False (the default), the human-centric names will
     be converted to their internal representation.
 
     Returns a 3-tuple of (compression, version, parameters). Compression will
     be ``None`` if not in strict mode and a compression isn't defined.
 
     An ``InvalidBundleSpecification`` is raised when the specification is
     not syntactically well formed.
 
     An ``UnsupportedBundleSpecification`` is raised when the compression or
     bundle type/version is not recognized.
 
     Note: this function will likely eventually return a more complex data
     structure, including bundle2 part information.
     """
     def parseparams(s):
         if ';' not in s:
             return s, {}
 
         params = {}
         version, paramstr = s.split(';', 1)
 
         for p in paramstr.split(';'):
             if '=' not in p:
                 raise error.InvalidBundleSpecification(
                     _('invalid bundle specification: '
                       'missing "=" in parameter: %s') % p)
 
             key, value = p.split('=', 1)
             key = urlreq.unquote(key)
             value = urlreq.unquote(value)
             params[key] = value
 
         return version, params
 
 
     if strict and '-' not in spec:
         raise error.InvalidBundleSpecification(
                 _('invalid bundle specification; '
                   'must be prefixed with compression: %s') % spec)
 
     if '-' in spec:
         compression, version = spec.split('-', 1)
 
         if compression not in util.compengines.supportedbundlenames:
             raise error.UnsupportedBundleSpecification(
                     _('%s compression is not supported') % compression)
 
         version, params = parseparams(version)
 
         if version not in _bundlespeccgversions:
             raise error.UnsupportedBundleSpecification(
                     _('%s is not a recognized bundle version') % version)
     else:
         # Value could be just the compression or just the version, in which
         # case some defaults are assumed (but only when not in strict mode).
         assert not strict
 
         spec, params = parseparams(spec)
 
         if spec in util.compengines.supportedbundlenames:
             compression = spec
             version = 'v1'
             # Generaldelta repos require v2.
             if 'generaldelta' in repo.requirements:
                 version = 'v2'
             # Modern compression engines require v2.
             if compression not in _bundlespecv1compengines:
                 version = 'v2'
         elif spec in _bundlespeccgversions:
             if spec == 'packed1':
                 compression = 'none'
             else:
                 compression = 'bzip2'
             version = spec
         else:
             raise error.UnsupportedBundleSpecification(
                     _('%s is not a recognized bundle specification') % spec)
 
     # Bundle version 1 only supports a known set of compression engines.
     if version == 'v1' and compression not in _bundlespecv1compengines:
         raise error.UnsupportedBundleSpecification(
             _('compression engine %s is not supported on v1 bundles') %
             compression)
 
     # The specification for packed1 can optionally declare the data formats
     # required to apply it. If we see this metadata, compare against what the
     # repo supports and error if the bundle isn't compatible.
     if version == 'packed1' and 'requirements' in params:
         requirements = set(params['requirements'].split(','))
         missingreqs = requirements - repo.supportedformats
         if missingreqs:
             raise error.UnsupportedBundleSpecification(
                     _('missing support for repository features: %s') %
                       ', '.join(sorted(missingreqs)))
 
     if not externalnames:
         engine = util.compengines.forbundlename(compression)
         compression = engine.bundletype()[1]
         version = _bundlespeccgversions[version]
     return compression, version, params
 
 def readbundle(ui, fh, fname, vfs=None):
     header = changegroup.readexactly(fh, 4)
 
     alg = None
     if not fname:
         fname = "stream"
         if not header.startswith('HG') and header.startswith('\0'):
             fh = changegroup.headerlessfixup(fh, header)
             header = "HG10"
             alg = 'UN'
     elif vfs:
         fname = vfs.join(fname)
 
     magic, version = header[0:2], header[2:4]
 
     if magic != 'HG':
         raise error.Abort(_('%s: not a Mercurial bundle') % fname)
     if version == '10':
         if alg is None:
             alg = changegroup.readexactly(fh, 2)
         return changegroup.cg1unpacker(fh, alg)
     elif version.startswith('2'):
         return bundle2.getunbundler(ui, fh, magicstring=magic + version)
     elif version == 'S1':
         return streamclone.streamcloneapplier(fh)
     else:
         raise error.Abort(_('%s: unknown bundle version %s') % (fname, version))
 
 def getbundlespec(ui, fh):
     """Infer the bundlespec from a bundle file handle.
 
     The input file handle is seeked and the original seek position is not
     restored.
     """
     def speccompression(alg):
         try:
             return util.compengines.forbundletype(alg).bundletype()[0]
         except KeyError:
             return None
 
     b = readbundle(ui, fh, None)
     if isinstance(b, changegroup.cg1unpacker):
         alg = b._type
         if alg == '_truncatedBZ':
             alg = 'BZ'
         comp = speccompression(alg)
         if not comp:
             raise error.Abort(_('unknown compression algorithm: %s') % alg)
         return '%s-v1' % comp
     elif isinstance(b, bundle2.unbundle20):
         if 'Compression' in b.params:
             comp = speccompression(b.params['Compression'])
             if not comp:
                 raise error.Abort(_('unknown compression algorithm: %s') % comp)
         else:
             comp = 'none'
 
         version = None
         for part in b.iterparts():
             if part.type == 'changegroup':
                 version = part.params['version']
                 if version in ('01', '02'):
                     version = 'v2'
                 else:
                     raise error.Abort(_('changegroup version %s does not have '
                                         'a known bundlespec') % version,
                                       hint=_('try upgrading your Mercurial '
                                               'client'))
 
         if not version:
             raise error.Abort(_('could not identify changegroup version in '
                                 'bundle'))
 
         return '%s-%s' % (comp, version)
     elif isinstance(b, streamclone.streamcloneapplier):
         requirements = streamclone.readbundle1header(fh)[2]
         params = 'requirements=%s' % ','.join(sorted(requirements))
         return 'none-packed1;%s' % urlreq.quote(params)
     else:
         raise error.Abort(_('unknown bundle type: %s') % b)
 
 def _computeoutgoing(repo, heads, common):
     """Computes which revs are outgoing given a set of common
     and a set of heads.
 
     This is a separate function so extensions can have access to
     the logic.
 
     Returns a discovery.outgoing object.
     """
     cl = repo.changelog
     if common:
         hasnode = cl.hasnode
         common = [n for n in common if hasnode(n)]
     else:
         common = [nullid]
     if not heads:
         heads = cl.heads()
     return discovery.outgoing(repo, common, heads)
 
 def _forcebundle1(op):
     """return true if a pull/push must use bundle1
 
     This function is used to allow testing of the older bundle version"""
     ui = op.repo.ui
     forcebundle1 = False
     # The goal is this config is to allow developer to choose the bundle
     # version used during exchanged. This is especially handy during test.
     # Value is a list of bundle version to be picked from, highest version
     # should be used.
     #
     # developer config: devel.legacy.exchange
     exchange = ui.configlist('devel', 'legacy.exchange')
     forcebundle1 = 'bundle2' not in exchange and 'bundle1' in exchange
     return forcebundle1 or not op.remote.capable('bundle2')
 
 class pushoperation(object):
     """A object that represent a single push operation
 
     Its purpose is to carry push related state and very common operations.
 
     A new pushoperation should be created at the beginning of each push and
     discarded afterward.
     """
 
     def __init__(self, repo, remote, force=False, revs=None, newbranch=False,
                  bookmarks=(), pushvars=None):
         # repo we push from
         self.repo = repo
         self.ui = repo.ui
         # repo we push to
         self.remote = remote
         # force option provided
         self.force = force
         # revs to be pushed (None is "all")
         self.revs = revs
         # bookmark explicitly pushed
         self.bookmarks = bookmarks
         # allow push of new branch
         self.newbranch = newbranch
         # step already performed
         # (used to check what steps have been already performed through bundle2)
         self.stepsdone = set()
         # Integer version of the changegroup push result
         # - None means nothing to push
         # - 0 means HTTP error
         # - 1 means we pushed and remote head count is unchanged *or*
         #   we have outgoing changesets but refused to push
         # - other values as described by addchangegroup()
         self.cgresult = None
         # Boolean value for the bookmark push
         self.bkresult = None
         # discover.outgoing object (contains common and outgoing data)
         self.outgoing = None
         # all remote topological heads before the push
         self.remoteheads = None
         # Details of the remote branch pre and post push
         #
         # mapping: {'branch': ([remoteheads],
         #                      [newheads],
         #                      [unsyncedheads],
         #                      [discardedheads])}
         # - branch: the branch name
         # - remoteheads: the list of remote heads known locally
         #                None if the branch is new
         # - newheads: the new remote heads (known locally) with outgoing pushed
         # - unsyncedheads: the list of remote heads unknown locally.
         # - discardedheads: the list of remote heads made obsolete by the push
         self.pushbranchmap = None
         # testable as a boolean indicating if any nodes are missing locally.
         self.incoming = None
         # summary of the remote phase situation
         self.remotephases = None
         # phases changes that must be pushed along side the changesets
         self.outdatedphases = None
         # phases changes that must be pushed if changeset push fails
         self.fallbackoutdatedphases = None
         # outgoing obsmarkers
         self.outobsmarkers = set()
         # outgoing bookmarks
         self.outbookmarks = []
         # transaction manager
         self.trmanager = None
         # map { pushkey partid -> callback handling failure}
         # used to handle exception from mandatory pushkey part failure
         self.pkfailcb = {}
         # an iterable of pushvars or None
         self.pushvars = pushvars
 
     @util.propertycache
     def futureheads(self):
         """future remote heads if the changeset push succeeds"""
         return self.outgoing.missingheads
 
     @util.propertycache
     def fallbackheads(self):
         """future remote heads if the changeset push fails"""
         if self.revs is None:
             # not target to push, all common are relevant
             return self.outgoing.commonheads
         unfi = self.repo.unfiltered()
         # I want cheads = heads(::missingheads and ::commonheads)
         # (missingheads is revs with secret changeset filtered out)
         #
         # This can be expressed as:
         #     cheads = ( (missingheads and ::commonheads)
         #              + (commonheads and ::missingheads))"
         #              )
         #
         # while trying to push we already computed the following:
         #     common = (::commonheads)
         #     missing = ((commonheads::missingheads) - commonheads)
         #
         # We can pick:
         # * missingheads part of common (::commonheads)
         common = self.outgoing.common
         nm = self.repo.changelog.nodemap
         cheads = [node for node in self.revs if nm[node] in common]
         # and
         # * commonheads parents on missing
         revset = unfi.set('%ln and parents(roots(%ln))',
                          self.outgoing.commonheads,
                          self.outgoing.missing)
         cheads.extend(c.node() for c in revset)
         return cheads
 
     @property
     def commonheads(self):
         """set of all common heads after changeset bundle push"""
         if self.cgresult:
             return self.futureheads
         else:
             return self.fallbackheads
 
 # mapping of message used when pushing bookmark
 bookmsgmap = {'update': (_("updating bookmark %s\n"),
                          _('updating bookmark %s failed!\n')),
               'export': (_("exporting bookmark %s\n"),
                          _('exporting bookmark %s failed!\n')),
               'delete': (_("deleting remote bookmark %s\n"),
                          _('deleting remote bookmark %s failed!\n')),
               }
 
 
 def push(repo, remote, force=False, revs=None, newbranch=False, bookmarks=(),
          opargs=None):
     '''Push outgoing changesets (limited by revs) from a local
     repository to remote. Return an integer:
       - None means nothing to push
       - 0 means HTTP error
       - 1 means we pushed and remote head count is unchanged *or*
         we have outgoing changesets but refused to push
       - other values as described by addchangegroup()
     '''
     if opargs is None:
         opargs = {}
     pushop = pushoperation(repo, remote, force, revs, newbranch, bookmarks,
                            **pycompat.strkwargs(opargs))
     if pushop.remote.local():
         missing = (set(pushop.repo.requirements)
                    - pushop.remote.local().supported)
         if missing:
             msg = _("required features are not"
                     " supported in the destination:"
                     " %s") % (', '.join(sorted(missing)))
             raise error.Abort(msg)
 
     if not pushop.remote.canpush():
         raise error.Abort(_("destination does not support push"))
 
     if not pushop.remote.capable('unbundle'):
         raise error.Abort(_('cannot push: destination does not support the '
                             'unbundle wire protocol command'))
 
     # get lock as we might write phase data
     wlock = lock = None
     try:
         # bundle2 push may receive a reply bundle touching bookmarks or other
         # things requiring the wlock. Take it now to ensure proper ordering.
         maypushback = pushop.ui.configbool('experimental', 'bundle2.pushback')
         if (not _forcebundle1(pushop)) and maypushback:
             wlock = pushop.repo.wlock()
         lock = pushop.repo.lock()
         pushop.trmanager = transactionmanager(pushop.repo,
                                               'push-response',
                                               pushop.remote.url())
     except IOError as err:
         if err.errno != errno.EACCES:
             raise
         # source repo cannot be locked.
         # We do not abort the push, but just disable the local phase
         # synchronisation.
         msg = 'cannot lock source repository: %s\n' % err
         pushop.ui.debug(msg)
 
     with wlock or util.nullcontextmanager(), \
             lock or util.nullcontextmanager(), \
             pushop.trmanager or util.nullcontextmanager():
         pushop.repo.checkpush(pushop)
         _pushdiscovery(pushop)
         if not _forcebundle1(pushop):
             _pushbundle2(pushop)
         _pushchangeset(pushop)
         _pushsyncphase(pushop)
         _pushobsolete(pushop)
         _pushbookmark(pushop)
 
     return pushop
 
 # list of steps to perform discovery before push
 pushdiscoveryorder = []
 
 # Mapping between step name and function
 #
 # This exists to help extensions wrap steps if necessary
 pushdiscoverymapping = {}
 
 def pushdiscovery(stepname):
     """decorator for function performing discovery before push
 
     The function is added to the step -> function mapping and appended to the
     list of steps.  Beware that decorated function will be added in order (this
     may matter).
 
     You can only use this decorator for a new step, if you want to wrap a step
     from an extension, change the pushdiscovery dictionary directly."""
     def dec(func):
         assert stepname not in pushdiscoverymapping
         pushdiscoverymapping[stepname] = func
         pushdiscoveryorder.append(stepname)
         return func
     return dec
 
 def _pushdiscovery(pushop):
     """Run all discovery steps"""
     for stepname in pushdiscoveryorder:
         step = pushdiscoverymapping[stepname]
         step(pushop)
 
 @pushdiscovery('changeset')
 def _pushdiscoverychangeset(pushop):
     """discover the changeset that need to be pushed"""
     fci = discovery.findcommonincoming
     commoninc = fci(pushop.repo, pushop.remote, force=pushop.force)
     common, inc, remoteheads = commoninc
     fco = discovery.findcommonoutgoing
     outgoing = fco(pushop.repo, pushop.remote, onlyheads=pushop.revs,
                    commoninc=commoninc, force=pushop.force)
     pushop.outgoing = outgoing
     pushop.remoteheads = remoteheads
     pushop.incoming = inc
 
 @pushdiscovery('phase')
 def _pushdiscoveryphase(pushop):
     """discover the phase that needs to be pushed
 
     (computed for both success and failure case for changesets push)"""
     outgoing = pushop.outgoing
     unfi = pushop.repo.unfiltered()
     remotephases = pushop.remote.listkeys('phases')
     if (pushop.ui.configbool('ui', '_usedassubrepo')
         and remotephases    # server supports phases
         and not pushop.outgoing.missing # no changesets to be pushed
         and remotephases.get('publishing', False)):
         # When:
         # - this is a subrepo push
         # - and remote support phase
         # - and no changeset are to be pushed
         # - and remote is publishing
         # We may be in issue 3781 case!
         # We drop the possible phase synchronisation done by
         # courtesy to publish changesets possibly locally draft
         # on the remote.
         pushop.outdatedphases = []
         pushop.fallbackoutdatedphases = []
         return
 
     pushop.remotephases = phases.remotephasessummary(pushop.repo,
                                                      pushop.fallbackheads,
                                                      remotephases)
     droots = pushop.remotephases.draftroots
 
     extracond = ''
     if not pushop.remotephases.publishing:
         extracond = ' and public()'
     revset = 'heads((%%ln::%%ln) %s)' % extracond
     # Get the list of all revs draft on remote by public here.
     # XXX Beware that revset break if droots is not strictly
     # XXX root we may want to ensure it is but it is costly
     fallback = list(unfi.set(revset, droots, pushop.fallbackheads))
     if not outgoing.missing:
         future = fallback
     else:
         # adds changeset we are going to push as draft
         #
         # should not be necessary for publishing server, but because of an
         # issue fixed in xxxxx we have to do it anyway.
         fdroots = list(unfi.set('roots(%ln  + %ln::)',
                        outgoing.missing, droots))
         fdroots = [f.node() for f in fdroots]
         future = list(unfi.set(revset, fdroots, pushop.futureheads))
     pushop.outdatedphases = future
     pushop.fallbackoutdatedphases = fallback
 
 @pushdiscovery('obsmarker')
 def _pushdiscoveryobsmarkers(pushop):
     if (obsolete.isenabled(pushop.repo, obsolete.exchangeopt)
         and pushop.repo.obsstore
         and 'obsolete' in pushop.remote.listkeys('namespaces')):
         repo = pushop.repo
         # very naive computation, that can be quite expensive on big repo.
         # However: evolution is currently slow on them anyway.
         nodes = (c.node() for c in repo.set('::%ln', pushop.futureheads))
         pushop.outobsmarkers = pushop.repo.obsstore.relevantmarkers(nodes)
 
 @pushdiscovery('bookmarks')
 def _pushdiscoverybookmarks(pushop):
     ui = pushop.ui
     repo = pushop.repo.unfiltered()
     remote = pushop.remote
     ui.debug("checking for updated bookmarks\n")
     ancestors = ()
     if pushop.revs:
         revnums = map(repo.changelog.rev, pushop.revs)
         ancestors = repo.changelog.ancestors(revnums, inclusive=True)
     remotebookmark = remote.listkeys('bookmarks')
 
     explicit = set([repo._bookmarks.expandname(bookmark)
                     for bookmark in pushop.bookmarks])
 
     remotebookmark = bookmod.unhexlifybookmarks(remotebookmark)
     comp = bookmod.comparebookmarks(repo, repo._bookmarks, remotebookmark)
 
     def safehex(x):
         if x is None:
             return x
         return hex(x)
 
     def hexifycompbookmarks(bookmarks):
         for b, scid, dcid in bookmarks:
             yield b, safehex(scid), safehex(dcid)
 
     comp = [hexifycompbookmarks(marks) for marks in comp]
     addsrc, adddst, advsrc, advdst, diverge, differ, invalid, same = comp
 
     for b, scid, dcid in advsrc:
         if b in explicit:
             explicit.remove(b)
         if not ancestors or repo[scid].rev() in ancestors:
             pushop.outbookmarks.append((b, dcid, scid))
     # search added bookmark
     for b, scid, dcid in addsrc:
         if b in explicit:
             explicit.remove(b)
             pushop.outbookmarks.append((b, '', scid))
     # search for overwritten bookmark
     for b, scid, dcid in list(advdst) + list(diverge) + list(differ):
         if b in explicit:
             explicit.remove(b)
             pushop.outbookmarks.append((b, dcid, scid))
     # search for bookmark to delete
     for b, scid, dcid in adddst:
         if b in explicit:
             explicit.remove(b)
             # treat as "deleted locally"
             pushop.outbookmarks.append((b, dcid, ''))
     # identical bookmarks shouldn't get reported
     for b, scid, dcid in same:
         if b in explicit:
             explicit.remove(b)
 
     if explicit:
         explicit = sorted(explicit)
         # we should probably list all of them
         ui.warn(_('bookmark %s does not exist on the local '
                   'or remote repository!\n') % explicit[0])
         pushop.bkresult = 2
 
     pushop.outbookmarks.sort()
 
 def _pushcheckoutgoing(pushop):
     outgoing = pushop.outgoing
     unfi = pushop.repo.unfiltered()
     if not outgoing.missing:
         # nothing to push
         scmutil.nochangesfound(unfi.ui, unfi, outgoing.excluded)
         return False
     # something to push
     if not pushop.force:
         # if repo.obsstore == False --> no obsolete
         # then, save the iteration
         if unfi.obsstore:
             # this message are here for 80 char limit reason
             mso = _("push includes obsolete changeset: %s!")
             mspd = _("push includes phase-divergent changeset: %s!")
             mscd = _("push includes content-divergent changeset: %s!")
             mst = {"orphan": _("push includes orphan changeset: %s!"),
                    "phase-divergent": mspd,
                    "content-divergent": mscd}
             # If we are to push if there is at least one
             # obsolete or unstable changeset in missing, at
             # least one of the missinghead will be obsolete or
             # unstable. So checking heads only is ok
             for node in outgoing.missingheads:
                 ctx = unfi[node]
                 if ctx.obsolete():
                     raise error.Abort(mso % ctx)
                 elif ctx.isunstable():
                     # TODO print more than one instability in the abort
                     # message
                     raise error.Abort(mst[ctx.instabilities()[0]] % ctx)
 
         discovery.checkheads(pushop)
     return True
 
 # List of names of steps to perform for an outgoing bundle2, order matters.
 b2partsgenorder = []
 
 # Mapping between step name and function
 #
 # This exists to help extensions wrap steps if necessary
 b2partsgenmapping = {}
 
 def b2partsgenerator(stepname, idx=None):
     """decorator for function generating bundle2 part
 
     The function is added to the step -> function mapping and appended to the
     list of steps.  Beware that decorated functions will be added in order
     (this may matter).
 
     You can only use this decorator for new steps, if you want to wrap a step
     from an extension, attack the b2partsgenmapping dictionary directly."""
     def dec(func):
         assert stepname not in b2partsgenmapping
         b2partsgenmapping[stepname] = func
         if idx is None:
             b2partsgenorder.append(stepname)
         else:
             b2partsgenorder.insert(idx, stepname)
         return func
     return dec
 
 def _pushb2ctxcheckheads(pushop, bundler):
     """Generate race condition checking parts
 
     Exists as an independent function to aid extensions
     """
     # * 'force' do not check for push race,
     # * if we don't push anything, there are nothing to check.
     if not pushop.force and pushop.outgoing.missingheads:
         allowunrelated = 'related' in bundler.capabilities.get('checkheads', ())
         emptyremote = pushop.pushbranchmap is None
         if not allowunrelated or emptyremote:
             bundler.newpart('check:heads', data=iter(pushop.remoteheads))
         else:
             affected = set()
             for branch, heads in pushop.pushbranchmap.iteritems():
                 remoteheads, newheads, unsyncedheads, discardedheads = heads
                 if remoteheads is not None:
                     remote = set(remoteheads)
                     affected |= set(discardedheads) & remote
                     affected |= remote - set(newheads)
             if affected:
                 data = iter(sorted(affected))
                 bundler.newpart('check:updated-heads', data=data)
 
 def _pushing(pushop):
     """return True if we are pushing anything"""
     return bool(pushop.outgoing.missing
                 or pushop.outdatedphases
                 or pushop.outobsmarkers
                 or pushop.outbookmarks)
 
 @b2partsgenerator('check-bookmarks')
 def _pushb2checkbookmarks(pushop, bundler):
     """insert bookmark move checking"""
     if not _pushing(pushop) or pushop.force:
         return
     b2caps = bundle2.bundle2caps(pushop.remote)
     hasbookmarkcheck = 'bookmarks' in b2caps
     if not (pushop.outbookmarks and hasbookmarkcheck):
         return
     data = []
     for book, old, new in pushop.outbookmarks:
         old = bin(old)
         data.append((book, old))
     checkdata = bookmod.binaryencode(data)
     bundler.newpart('check:bookmarks', data=checkdata)
 
 @b2partsgenerator('check-phases')
 def _pushb2checkphases(pushop, bundler):
     """insert phase move checking"""
     if not _pushing(pushop) or pushop.force:
         return
     b2caps = bundle2.bundle2caps(pushop.remote)
     hasphaseheads = 'heads' in b2caps.get('phases', ())
     if pushop.remotephases is not None and hasphaseheads:
         # check that the remote phase has not changed
         checks = [[] for p in phases.allphases]
         checks[phases.public].extend(pushop.remotephases.publicheads)
         checks[phases.draft].extend(pushop.remotephases.draftroots)
         if any(checks):
             for nodes in checks:
                 nodes.sort()
             checkdata = phases.binaryencode(checks)
             bundler.newpart('check:phases', data=checkdata)
 
 @b2partsgenerator('changeset')
 def _pushb2ctx(pushop, bundler):
     """handle changegroup push through bundle2
 
     addchangegroup result is stored in the ``pushop.cgresult`` attribute.
     """
     if 'changesets' in pushop.stepsdone:
         return
     pushop.stepsdone.add('changesets')
     # Send known heads to the server for race detection.
     if not _pushcheckoutgoing(pushop):
         return
     pushop.repo.prepushoutgoinghooks(pushop)
 
     _pushb2ctxcheckheads(pushop, bundler)
 
     b2caps = bundle2.bundle2caps(pushop.remote)
     version = '01'
     cgversions = b2caps.get('changegroup')
     if cgversions:  # 3.1 and 3.2 ship with an empty value
         cgversions = [v for v in cgversions
                       if v in changegroup.supportedoutgoingversions(
                           pushop.repo)]
         if not cgversions:
             raise ValueError(_('no common changegroup version'))
         version = max(cgversions)
     cgstream = changegroup.makestream(pushop.repo, pushop.outgoing, version,
                                       'push')
     cgpart = bundler.newpart('changegroup', data=cgstream)
     if cgversions:
         cgpart.addparam('version', version)
     if 'treemanifest' in pushop.repo.requirements:
         cgpart.addparam('treemanifest', '1')
     def handlereply(op):
         """extract addchangegroup returns from server reply"""
         cgreplies = op.records.getreplies(cgpart.id)
         assert len(cgreplies['changegroup']) == 1
         pushop.cgresult = cgreplies['changegroup'][0]['return']
     return handlereply
 
 @b2partsgenerator('phase')
 def _pushb2phases(pushop, bundler):
     """handle phase push through bundle2"""
     if 'phases' in pushop.stepsdone:
         return
     b2caps = bundle2.bundle2caps(pushop.remote)
     ui = pushop.repo.ui
 
     legacyphase = 'phases' in ui.configlist('devel', 'legacy.exchange')
     haspushkey = 'pushkey' in b2caps
     hasphaseheads = 'heads' in b2caps.get('phases', ())
 
     if hasphaseheads and not legacyphase:
         return _pushb2phaseheads(pushop, bundler)
     elif haspushkey:
         return _pushb2phasespushkey(pushop, bundler)
 
 def _pushb2phaseheads(pushop, bundler):
     """push phase information through a bundle2 - binary part"""
     pushop.stepsdone.add('phases')
     if pushop.outdatedphases:
         updates = [[] for p in phases.allphases]
         updates[0].extend(h.node() for h in pushop.outdatedphases)
         phasedata = phases.binaryencode(updates)
         bundler.newpart('phase-heads', data=phasedata)
 
 def _pushb2phasespushkey(pushop, bundler):
     """push phase information through a bundle2 - pushkey part"""
     pushop.stepsdone.add('phases')
     part2node = []
 
     def handlefailure(pushop, exc):
         targetid = int(exc.partid)
         for partid, node in part2node:
             if partid == targetid:
                 raise error.Abort(_('updating %s to public failed') % node)
 
     enc = pushkey.encode
     for newremotehead in pushop.outdatedphases:
         part = bundler.newpart('pushkey')
         part.addparam('namespace', enc('phases'))
         part.addparam('key', enc(newremotehead.hex()))
         part.addparam('old', enc('%d' % phases.draft))
         part.addparam('new', enc('%d' % phases.public))
         part2node.append((part.id, newremotehead))
         pushop.pkfailcb[part.id] = handlefailure
 
     def handlereply(op):
         for partid, node in part2node:
             partrep = op.records.getreplies(partid)
             results = partrep['pushkey']
             assert len(results) <= 1
             msg = None
             if not results:
                 msg = _('server ignored update of %s to public!\n') % node
             elif not int(results[0]['return']):
                 msg = _('updating %s to public failed!\n') % node
             if msg is not None:
                 pushop.ui.warn(msg)
     return handlereply
 
 @b2partsgenerator('obsmarkers')
 def _pushb2obsmarkers(pushop, bundler):
     if 'obsmarkers' in pushop.stepsdone:
         return
     remoteversions = bundle2.obsmarkersversion(bundler.capabilities)
     if obsolete.commonversion(remoteversions) is None:
         return
     pushop.stepsdone.add('obsmarkers')
     if pushop.outobsmarkers:
         markers = sorted(pushop.outobsmarkers)
         bundle2.buildobsmarkerspart(bundler, markers)
 
 @b2partsgenerator('bookmarks')
 def _pushb2bookmarks(pushop, bundler):
     """handle bookmark push through bundle2"""
     if 'bookmarks' in pushop.stepsdone:
         return
     b2caps = bundle2.bundle2caps(pushop.remote)
 
     legacy = pushop.repo.ui.configlist('devel', 'legacy.exchange')
     legacybooks = 'bookmarks' in legacy
 
     if not legacybooks and 'bookmarks' in b2caps:
         return _pushb2bookmarkspart(pushop, bundler)
     elif 'pushkey' in b2caps:
         return _pushb2bookmarkspushkey(pushop, bundler)
 
 def _bmaction(old, new):
     """small utility for bookmark pushing"""
     if not old:
         return 'export'
     elif not new:
         return 'delete'
     return 'update'
 
 def _pushb2bookmarkspart(pushop, bundler):
     pushop.stepsdone.add('bookmarks')
     if not pushop.outbookmarks:
         return
 
     allactions = []
     data = []
     for book, old, new in pushop.outbookmarks:
         new = bin(new)
         data.append((book, new))
         allactions.append((book, _bmaction(old, new)))
     checkdata = bookmod.binaryencode(data)
     bundler.newpart('bookmarks', data=checkdata)
 
     def handlereply(op):
         ui = pushop.ui
         # if success
         for book, action in allactions:
             ui.status(bookmsgmap[action][0] % book)
 
     return handlereply
 
 def _pushb2bookmarkspushkey(pushop, bundler):
     pushop.stepsdone.add('bookmarks')
     part2book = []
     enc = pushkey.encode
 
     def handlefailure(pushop, exc):
         targetid = int(exc.partid)
         for partid, book, action in part2book:
             if partid == targetid:
                 raise error.Abort(bookmsgmap[action][1].rstrip() % book)
         # we should not be called for part we did not generated
         assert False
 
     for book, old, new in pushop.outbookmarks:
         part = bundler.newpart('pushkey')
         part.addparam('namespace', enc('bookmarks'))
         part.addparam('key', enc(book))
         part.addparam('old', enc(old))
         part.addparam('new', enc(new))
         action = 'update'
         if not old:
             action = 'export'
         elif not new:
             action = 'delete'
         part2book.append((part.id, book, action))
         pushop.pkfailcb[part.id] = handlefailure
 
     def handlereply(op):
         ui = pushop.ui
         for partid, book, action in part2book:
             partrep = op.records.getreplies(partid)
             results = partrep['pushkey']
             assert len(results) <= 1
             if not results:
                 pushop.ui.warn(_('server ignored bookmark %s update\n') % book)
             else:
                 ret = int(results[0]['return'])
                 if ret:
                     ui.status(bookmsgmap[action][0] % book)
                 else:
                     ui.warn(bookmsgmap[action][1] % book)
                     if pushop.bkresult is not None:
                         pushop.bkresult = 1
     return handlereply
 
 @b2partsgenerator('pushvars', idx=0)
 def _getbundlesendvars(pushop, bundler):
     '''send shellvars via bundle2'''
     pushvars = pushop.pushvars
     if pushvars:
         shellvars = {}
         for raw in pushvars:
             if '=' not in raw:
                 msg = ("unable to parse variable '%s', should follow "
                         "'KEY=VALUE' or 'KEY=' format")
                 raise error.Abort(msg % raw)
             k, v = raw.split('=', 1)
             shellvars[k] = v
 
         part = bundler.newpart('pushvars')
 
         for key, value in shellvars.iteritems():
             part.addparam(key, value, mandatory=False)
 
 def _pushbundle2(pushop):
     """push data to the remote using bundle2
 
     The only currently supported type of data is changegroup but this will
     evolve in the future."""
     bundler = bundle2.bundle20(pushop.ui, bundle2.bundle2caps(pushop.remote))
     pushback = (pushop.trmanager
                 and pushop.ui.configbool('experimental', 'bundle2.pushback'))
 
     # create reply capability
     capsblob = bundle2.encodecaps(bundle2.getrepocaps(pushop.repo,
                                                       allowpushback=pushback))
     bundler.newpart('replycaps', data=capsblob)
     replyhandlers = []
     for partgenname in b2partsgenorder:
         partgen = b2partsgenmapping[partgenname]
         ret = partgen(pushop, bundler)
         if callable(ret):
             replyhandlers.append(ret)
     # do not push if nothing to push
     if bundler.nbparts <= 1:
         return
     stream = util.chunkbuffer(bundler.getchunks())
     try:
         try:
             reply = pushop.remote.unbundle(
                 stream, ['force'], pushop.remote.url())
         except error.BundleValueError as exc:
             raise error.Abort(_('missing support for %s') % exc)
         try:
             trgetter = None
             if pushback:
                 trgetter = pushop.trmanager.transaction
             op = bundle2.processbundle(pushop.repo, reply, trgetter)
         except error.BundleValueError as exc:
             raise error.Abort(_('missing support for %s') % exc)
         except bundle2.AbortFromPart as exc:
             pushop.ui.status(_('remote: %s\n') % exc)
             if exc.hint is not None:
                 pushop.ui.status(_('remote: %s\n') % ('(%s)' % exc.hint))
             raise error.Abort(_('push failed on remote'))
     except error.PushkeyFailed as exc:
         partid = int(exc.partid)
         if partid not in pushop.pkfailcb:
             raise
         pushop.pkfailcb[partid](pushop, exc)
     for rephand in replyhandlers:
         rephand(op)
 
 def _pushchangeset(pushop):
     """Make the actual push of changeset bundle to remote repo"""
     if 'changesets' in pushop.stepsdone:
         return
     pushop.stepsdone.add('changesets')
     if not _pushcheckoutgoing(pushop):
         return
 
     # Should have verified this in push().
     assert pushop.remote.capable('unbundle')
 
     pushop.repo.prepushoutgoinghooks(pushop)
     outgoing = pushop.outgoing
     # TODO: get bundlecaps from remote
     bundlecaps = None
     # create a changegroup from local
     if pushop.revs is None and not (outgoing.excluded
                             or pushop.repo.changelog.filteredrevs):
         # push everything,
         # use the fast path, no race possible on push
         cg = changegroup.makechangegroup(pushop.repo, outgoing, '01', 'push',
                 fastpath=True, bundlecaps=bundlecaps)
     else:
         cg = changegroup.makechangegroup(pushop.repo, outgoing, '01',
                                         'push', bundlecaps=bundlecaps)
 
     # apply changegroup to remote
     # local repo finds heads on server, finds out what
     # revs it must push. once revs transferred, if server
     # finds it has different heads (someone else won
     # commit/push race), server aborts.
     if pushop.force:
         remoteheads = ['force']
     else:
         remoteheads = pushop.remoteheads
     # ssh: return remote's addchangegroup()
     # http: return remote's addchangegroup() or 0 for error
     pushop.cgresult = pushop.remote.unbundle(cg, remoteheads,
                                         pushop.repo.url())
 
 def _pushsyncphase(pushop):
     """synchronise phase information locally and remotely"""
     cheads = pushop.commonheads
     # even when we don't push, exchanging phase data is useful
     remotephases = pushop.remote.listkeys('phases')
     if (pushop.ui.configbool('ui', '_usedassubrepo')
         and remotephases    # server supports phases
         and pushop.cgresult is None # nothing was pushed
         and remotephases.get('publishing', False)):
         # When:
         # - this is a subrepo push
         # - and remote support phase
         # - and no changeset was pushed
         # - and remote is publishing
         # We may be in issue 3871 case!
         # We drop the possible phase synchronisation done by
         # courtesy to publish changesets possibly locally draft
         # on the remote.
         remotephases = {'publishing': 'True'}
     if not remotephases: # old server or public only reply from non-publishing
         _localphasemove(pushop, cheads)
         # don't push any phase data as there is nothing to push
     else:
         ana = phases.analyzeremotephases(pushop.repo, cheads,
                                          remotephases)
         pheads, droots = ana
         ### Apply remote phase on local
         if remotephases.get('publishing', False):
             _localphasemove(pushop, cheads)
         else: # publish = False
             _localphasemove(pushop, pheads)
             _localphasemove(pushop, cheads, phases.draft)
         ### Apply local phase on remote
 
         if pushop.cgresult:
             if 'phases' in pushop.stepsdone:
                 # phases already pushed though bundle2
                 return
             outdated = pushop.outdatedphases
         else:
             outdated = pushop.fallbackoutdatedphases
 
         pushop.stepsdone.add('phases')
 
         # filter heads already turned public by the push
         outdated = [c for c in outdated if c.node() not in pheads]
         # fallback to independent pushkey command
         for newremotehead in outdated:
             r = pushop.remote.pushkey('phases',
                                       newremotehead.hex(),
                                       str(phases.draft),
                                       str(phases.public))
             if not r:
                 pushop.ui.warn(_('updating %s to public failed!\n')
                                % newremotehead)
 
 def _localphasemove(pushop, nodes, phase=phases.public):
     """move <nodes> to <phase> in the local source repo"""
     if pushop.trmanager:
         phases.advanceboundary(pushop.repo,
                                pushop.trmanager.transaction(),
                                phase,
                                nodes)
     else:
         # repo is not locked, do not change any phases!
         # Informs the user that phases should have been moved when
         # applicable.
         actualmoves = [n for n in nodes if phase < pushop.repo[n].phase()]
         phasestr = phases.phasenames[phase]
         if actualmoves:
             pushop.ui.status(_('cannot lock source repo, skipping '
                                'local %s phase update\n') % phasestr)
 
 def _pushobsolete(pushop):
     """utility function to push obsolete markers to a remote"""
     if 'obsmarkers' in pushop.stepsdone:
         return
     repo = pushop.repo
     remote = pushop.remote
     pushop.stepsdone.add('obsmarkers')
     if pushop.outobsmarkers:
         pushop.ui.debug('try to push obsolete markers to remote\n')
         rslts = []
         remotedata = obsolete._pushkeyescape(sorted(pushop.outobsmarkers))
         for key in sorted(remotedata, reverse=True):
             # reverse sort to ensure we end with dump0
             data = remotedata[key]
             rslts.append(remote.pushkey('obsolete', key, '', data))
         if [r for r in rslts if not r]:
             msg = _('failed to push some obsolete markers!\n')
             repo.ui.warn(msg)
 
 def _pushbookmark(pushop):
     """Update bookmark position on remote"""
     if pushop.cgresult == 0 or 'bookmarks' in pushop.stepsdone:
         return
     pushop.stepsdone.add('bookmarks')
     ui = pushop.ui
     remote = pushop.remote
 
     for b, old, new in pushop.outbookmarks:
         action = 'update'
         if not old:
             action = 'export'
         elif not new:
             action = 'delete'
         if remote.pushkey('bookmarks', b, old, new):
             ui.status(bookmsgmap[action][0] % b)
         else:
             ui.warn(bookmsgmap[action][1] % b)
             # discovery can have set the value form invalid entry
             if pushop.bkresult is not None:
                 pushop.bkresult = 1
 
 class pulloperation(object):
     """A object that represent a single pull operation
 
     It purpose is to carry pull related state and very common operation.
 
     A new should be created at the beginning of each pull and discarded
     afterward.
     """
 
     def __init__(self, repo, remote, heads=None, force=False, bookmarks=(),
                  remotebookmarks=None, streamclonerequested=None):
         # repo we pull into
         self.repo = repo
         # repo we pull from
         self.remote = remote
         # revision we try to pull (None is "all")
         self.heads = heads
         # bookmark pulled explicitly
         self.explicitbookmarks = [repo._bookmarks.expandname(bookmark)
                                   for bookmark in bookmarks]
         # do we force pull?
         self.force = force
         # whether a streaming clone was requested
         self.streamclonerequested = streamclonerequested
         # transaction manager
         self.trmanager = None
         # set of common changeset between local and remote before pull
         self.common = None
         # set of pulled head
         self.rheads = None
         # list of missing changeset to fetch remotely
         self.fetch = None
         # remote bookmarks data
         self.remotebookmarks = remotebookmarks
         # result of changegroup pulling (used as return code by pull)
         self.cgresult = None
         # list of step already done
         self.stepsdone = set()
         # Whether we attempted a clone from pre-generated bundles.
         self.clonebundleattempted = False
 
     @util.propertycache
     def pulledsubset(self):
         """heads of the set of changeset target by the pull"""
         # compute target subset
         if self.heads is None:
             # We pulled every thing possible
             # sync on everything common
             c = set(self.common)
             ret = list(self.common)
             for n in self.rheads:
                 if n not in c:
                     ret.append(n)
             return ret
         else:
             # We pulled a specific subset
             # sync on this subset
             return self.heads
 
     @util.propertycache
     def canusebundle2(self):
         return not _forcebundle1(self)
 
     @util.propertycache
     def remotebundle2caps(self):
         return bundle2.bundle2caps(self.remote)
 
     def gettransaction(self):
         # deprecated; talk to trmanager directly
         return self.trmanager.transaction()
 
 class transactionmanager(util.transactional):
     """An object to manage the life cycle of a transaction
 
     It creates the transaction on demand and calls the appropriate hooks when
     closing the transaction."""
     def __init__(self, repo, source, url):
         self.repo = repo
         self.source = source
         self.url = url
         self._tr = None
 
     def transaction(self):
         """Return an open transaction object, constructing if necessary"""
         if not self._tr:
             trname = '%s\n%s' % (self.source, util.hidepassword(self.url))
             self._tr = self.repo.transaction(trname)
             self._tr.hookargs['source'] = self.source
             self._tr.hookargs['url'] = self.url
         return self._tr
 
     def close(self):
         """close transaction if created"""
         if self._tr is not None:
             self._tr.close()
 
     def release(self):
         """release transaction if created"""
         if self._tr is not None:
             self._tr.release()
 
 def pull(repo, remote, heads=None, force=False, bookmarks=(), opargs=None,
          streamclonerequested=None):
     """Fetch repository data from a remote.
 
     This is the main function used to retrieve data from a remote repository.
 
     ``repo`` is the local repository to clone into.
     ``remote`` is a peer instance.
     ``heads`` is an iterable of revisions we want to pull. ``None`` (the
     default) means to pull everything from the remote.
     ``bookmarks`` is an iterable of bookmarks requesting to be pulled. By
     default, all remote bookmarks are pulled.
     ``opargs`` are additional keyword arguments to pass to ``pulloperation``
     initialization.
     ``streamclonerequested`` is a boolean indicating whether a "streaming
     clone" is requested. A "streaming clone" is essentially a raw file copy
     of revlogs from the server. This only works when the local repository is
     empty. The default value of ``None`` means to respect the server
     configuration for preferring stream clones.
 
     Returns the ``pulloperation`` created for this pull.
     """
     if opargs is None:
         opargs = {}
     pullop = pulloperation(repo, remote, heads, force, bookmarks=bookmarks,
                            streamclonerequested=streamclonerequested, **opargs)
 
     peerlocal = pullop.remote.local()
     if peerlocal:
         missing = set(peerlocal.requirements) - pullop.repo.supported
         if missing:
             msg = _("required features are not"
                     " supported in the destination:"
                     " %s") % (', '.join(sorted(missing)))
             raise error.Abort(msg)
 
     wlock = lock = None
     try:
         wlock = pullop.repo.wlock()
         lock = pullop.repo.lock()
         pullop.trmanager = transactionmanager(repo, 'pull', remote.url())
         # This should ideally be in _pullbundle2(). However, it needs to run
         # before discovery to avoid extra work.
         _maybeapplyclonebundle(pullop)
         streamclone.maybeperformlegacystreamclone(pullop)
         _pulldiscovery(pullop)
         if pullop.canusebundle2:
             _pullbundle2(pullop)
         _pullchangeset(pullop)
         _pullphase(pullop)
         _pullbookmarks(pullop)
         _pullobsolete(pullop)
         pullop.trmanager.close()
     finally:
         lockmod.release(pullop.trmanager, lock, wlock)
 
     # storing remotenames
     if repo.ui.configbool('experimental', 'remotenames'):
         remotenames.pullremotenames(repo, remote)
 
     return pullop
 
 # list of steps to perform discovery before pull
 pulldiscoveryorder = []
 
 # Mapping between step name and function
 #
 # This exists to help extensions wrap steps if necessary
 pulldiscoverymapping = {}
 
 def pulldiscovery(stepname):
     """decorator for function performing discovery before pull
 
     The function is added to the step -> function mapping and appended to the
     list of steps.  Beware that decorated function will be added in order (this
     may matter).
 
     You can only use this decorator for a new step, if you want to wrap a step
     from an extension, change the pulldiscovery dictionary directly."""
     def dec(func):
         assert stepname not in pulldiscoverymapping
         pulldiscoverymapping[stepname] = func
         pulldiscoveryorder.append(stepname)
         return func
     return dec
 
 def _pulldiscovery(pullop):
     """Run all discovery steps"""
     for stepname in pulldiscoveryorder:
         step = pulldiscoverymapping[stepname]
         step(pullop)
 
 @pulldiscovery('b1:bookmarks')
 def _pullbookmarkbundle1(pullop):
     """fetch bookmark data in bundle1 case
 
     If not using bundle2, we have to fetch bookmarks before changeset
     discovery to reduce the chance and impact of race conditions."""
     if pullop.remotebookmarks is not None:
         return
     if pullop.canusebundle2 and 'listkeys' in pullop.remotebundle2caps:
         # all known bundle2 servers now support listkeys, but lets be nice with
         # new implementation.
         return
     books = pullop.remote.listkeys('bookmarks')
     pullop.remotebookmarks = bookmod.unhexlifybookmarks(books)
 
 
 @pulldiscovery('changegroup')
 def _pulldiscoverychangegroup(pullop):
     """discovery phase for the pull
 
     Current handle changeset discovery only, will change handle all discovery
     at some point."""
     tmp = discovery.findcommonincoming(pullop.repo,
                                        pullop.remote,
                                        heads=pullop.heads,
                                        force=pullop.force)
     common, fetch, rheads = tmp
     nm = pullop.repo.unfiltered().changelog.nodemap
     if fetch and rheads:
         # If a remote heads is filtered locally, put in back in common.
         #
         # This is a hackish solution to catch most of "common but locally
         # hidden situation".  We do not performs discovery on unfiltered
         # repository because it end up doing a pathological amount of round
         # trip for w huge amount of changeset we do not care about.
         #
         # If a set of such "common but filtered" changeset exist on the server
         # but are not including a remote heads, we'll not be able to detect it,
         scommon = set(common)
         for n in rheads:
             if n in nm:
                 if n not in scommon:
                     common.append(n)
         if set(rheads).issubset(set(common)):
             fetch = []
     pullop.common = common
     pullop.fetch = fetch
     pullop.rheads = rheads
 
 def _pullbundle2(pullop):
     """pull data using bundle2
 
     For now, the only supported data are changegroup."""
     kwargs = {'bundlecaps': caps20to10(pullop.repo)}
 
     # At the moment we don't do stream clones over bundle2. If that is
     # implemented then here's where the check for that will go.
     streaming = False
 
     # pulling changegroup
     pullop.stepsdone.add('changegroup')
 
     kwargs['common'] = pullop.common
     kwargs['heads'] = pullop.heads or pullop.rheads
     kwargs['cg'] = pullop.fetch
 
     ui = pullop.repo.ui
     legacyphase = 'phases' in ui.configlist('devel', 'legacy.exchange')
     hasbinaryphase = 'heads' in pullop.remotebundle2caps.get('phases', ())
     if (not legacyphase and hasbinaryphase):
         kwargs['phases'] = True
         pullop.stepsdone.add('phases')
 
     if 'listkeys' in pullop.remotebundle2caps:
         if 'phases' not in pullop.stepsdone:
             kwargs['listkeys'] = ['phases']
         if pullop.remotebookmarks is None:
             # make sure to always includes bookmark data when migrating
             # `hg incoming --bundle` to using this function.
             kwargs.setdefault('listkeys', []).append('bookmarks')
 
     # If this is a full pull / clone and the server supports the clone bundles
     # feature, tell the server whether we attempted a clone bundle. The
     # presence of this flag indicates the client supports clone bundles. This
     # will enable the server to treat clients that support clone bundles
     # differently from those that don't.
     if (pullop.remote.capable('clonebundles')
         and pullop.heads is None and list(pullop.common) == [nullid]):
         kwargs['cbattempted'] = pullop.clonebundleattempted
 
     if streaming:
         pullop.repo.ui.status(_('streaming all changes\n'))
     elif not pullop.fetch:
         pullop.repo.ui.status(_("no changes found\n"))
         pullop.cgresult = 0
     else:
         if pullop.heads is None and list(pullop.common) == [nullid]:
             pullop.repo.ui.status(_("requesting all changes\n"))
     if obsolete.isenabled(pullop.repo, obsolete.exchangeopt):
         remoteversions = bundle2.obsmarkersversion(pullop.remotebundle2caps)
         if obsolete.commonversion(remoteversions) is not None:
             kwargs['obsmarkers'] = True
             pullop.stepsdone.add('obsmarkers')
     _pullbundle2extraprepare(pullop, kwargs)
     bundle = pullop.remote.getbundle('pull', **pycompat.strkwargs(kwargs))
     try:
         op = bundle2.processbundle(pullop.repo, bundle, pullop.gettransaction)
     except bundle2.AbortFromPart as exc:
         pullop.repo.ui.status(_('remote: abort: %s\n') % exc)
         raise error.Abort(_('pull failed on remote'), hint=exc.hint)
     except error.BundleValueError as exc:
         raise error.Abort(_('missing support for %s') % exc)
 
     if pullop.fetch:
         pullop.cgresult = bundle2.combinechangegroupresults(op)
 
     # processing phases change
     for namespace, value in op.records['listkeys']:
         if namespace == 'phases':
             _pullapplyphases(pullop, value)
 
     # processing bookmark update
     for namespace, value in op.records['listkeys']:
         if namespace == 'bookmarks':
             pullop.remotebookmarks = bookmod.unhexlifybookmarks(value)
 
     # bookmark data were either already there or pulled in the bundle
     if pullop.remotebookmarks is not None:
         _pullbookmarks(pullop)
 
 def _pullbundle2extraprepare(pullop, kwargs):
     """hook function so that extensions can extend the getbundle call"""
 
 def _pullchangeset(pullop):
     """pull changeset from unbundle into the local repo"""
     # We delay the open of the transaction as late as possible so we
     # don't open transaction for nothing or you break future useful
     # rollback call
     if 'changegroup' in pullop.stepsdone:
         return
     pullop.stepsdone.add('changegroup')
     if not pullop.fetch:
         pullop.repo.ui.status(_("no changes found\n"))
         pullop.cgresult = 0
         return
     tr = pullop.gettransaction()
     if pullop.heads is None and list(pullop.common) == [nullid]:
         pullop.repo.ui.status(_("requesting all changes\n"))
     elif pullop.heads is None and pullop.remote.capable('changegroupsubset'):
         # issue1320, avoid a race if remote changed after discovery
         pullop.heads = pullop.rheads
 
     if pullop.remote.capable('getbundle'):
         # TODO: get bundlecaps from remote
         cg = pullop.remote.getbundle('pull', common=pullop.common,
                                      heads=pullop.heads or pullop.rheads)
     elif pullop.heads is None:
         cg = pullop.remote.changegroup(pullop.fetch, 'pull')
     elif not pullop.remote.capable('changegroupsubset'):
         raise error.Abort(_("partial pull cannot be done because "
                            "other repository doesn't support "
                            "changegroupsubset."))
     else:
         cg = pullop.remote.changegroupsubset(pullop.fetch, pullop.heads, 'pull')
     bundleop = bundle2.applybundle(pullop.repo, cg, tr, 'pull',
                                    pullop.remote.url())
     pullop.cgresult = bundle2.combinechangegroupresults(bundleop)
 
 def _pullphase(pullop):
     # Get remote phases data from remote
     if 'phases' in pullop.stepsdone:
         return
     remotephases = pullop.remote.listkeys('phases')
     _pullapplyphases(pullop, remotephases)
 
 def _pullapplyphases(pullop, remotephases):
     """apply phase movement from observed remote state"""
     if 'phases' in pullop.stepsdone:
         return
     pullop.stepsdone.add('phases')
     publishing = bool(remotephases.get('publishing', False))
     if remotephases and not publishing:
         # remote is new and non-publishing
         pheads, _dr = phases.analyzeremotephases(pullop.repo,
                                                  pullop.pulledsubset,
                                                  remotephases)
         dheads = pullop.pulledsubset
     else:
         # Remote is old or publishing all common changesets
         # should be seen as public
         pheads = pullop.pulledsubset
         dheads = []
     unfi = pullop.repo.unfiltered()
     phase = unfi._phasecache.phase
     rev = unfi.changelog.nodemap.get
     public = phases.public
     draft = phases.draft
 
     # exclude changesets already public locally and update the others
     pheads = [pn for pn in pheads if phase(unfi, rev(pn)) > public]
     if pheads:
         tr = pullop.gettransaction()
         phases.advanceboundary(pullop.repo, tr, public, pheads)
 
     # exclude changesets already draft locally and update the others
     dheads = [pn for pn in dheads if phase(unfi, rev(pn)) > draft]
     if dheads:
         tr = pullop.gettransaction()
         phases.advanceboundary(pullop.repo, tr, draft, dheads)
 
 def _pullbookmarks(pullop):
     """process the remote bookmark information to update the local one"""
     if 'bookmarks' in pullop.stepsdone:
         return
     pullop.stepsdone.add('bookmarks')
     repo = pullop.repo
     remotebookmarks = pullop.remotebookmarks
     bookmod.updatefromremote(repo.ui, repo, remotebookmarks,
                              pullop.remote.url(),
                              pullop.gettransaction,
                              explicit=pullop.explicitbookmarks)
 
 def _pullobsolete(pullop):
     """utility function to pull obsolete markers from a remote
 
     The `gettransaction` is function that return the pull transaction, creating
     one if necessary. We return the transaction to inform the calling code that
     a new transaction have been created (when applicable).
 
     Exists mostly to allow overriding for experimentation purpose"""
     if 'obsmarkers' in pullop.stepsdone:
         return
     pullop.stepsdone.add('obsmarkers')
     tr = None
     if obsolete.isenabled(pullop.repo, obsolete.exchangeopt):
         pullop.repo.ui.debug('fetching remote obsolete markers\n')
         remoteobs = pullop.remote.listkeys('obsolete')
         if 'dump0' in remoteobs:
             tr = pullop.gettransaction()
             markers = []
             for key in sorted(remoteobs, reverse=True):
                 if key.startswith('dump'):
                     data = util.b85decode(remoteobs[key])
                     version, newmarks = obsolete._readmarkers(data)
                     markers += newmarks
             if markers:
                 pullop.repo.obsstore.add(tr, markers)
             pullop.repo.invalidatevolatilesets()
     return tr
 
 def caps20to10(repo):
     """return a set with appropriate options to use bundle20 during getbundle"""
     caps = {'HG20'}
     capsblob = bundle2.encodecaps(bundle2.getrepocaps(repo))
     caps.add('bundle2=' + urlreq.quote(capsblob))
     return caps
 
 # List of names of steps to perform for a bundle2 for getbundle, order matters.
 getbundle2partsorder = []
 
 # Mapping between step name and function
 #
 # This exists to help extensions wrap steps if necessary
 getbundle2partsmapping = {}
 
 def getbundle2partsgenerator(stepname, idx=None):
     """decorator for function generating bundle2 part for getbundle
 
     The function is added to the step -> function mapping and appended to the
     list of steps.  Beware that decorated functions will be added in order
     (this may matter).
 
     You can only use this decorator for new steps, if you want to wrap a step
     from an extension, attack the getbundle2partsmapping dictionary directly."""
     def dec(func):
         assert stepname not in getbundle2partsmapping
         getbundle2partsmapping[stepname] = func
         if idx is None:
             getbundle2partsorder.append(stepname)
         else:
             getbundle2partsorder.insert(idx, stepname)
         return func
     return dec
 
 def bundle2requested(bundlecaps):
     if bundlecaps is not None:
         return any(cap.startswith('HG2') for cap in bundlecaps)
     return False
 
 def getbundlechunks(repo, source, heads=None, common=None, bundlecaps=None,
                     **kwargs):
     """Return chunks constituting a bundle's raw data.
 
     Could be a bundle HG10 or a bundle HG20 depending on bundlecaps
     passed.
 
     Returns an iterator over raw chunks (of varying sizes).
     """
     kwargs = pycompat.byteskwargs(kwargs)
     usebundle2 = bundle2requested(bundlecaps)
     # bundle10 case
     if not usebundle2:
         if bundlecaps and not kwargs.get('cg', True):
             raise ValueError(_('request for bundle10 must include changegroup'))
 
         if kwargs:
             raise ValueError(_('unsupported getbundle arguments: %s')
                              % ', '.join(sorted(kwargs.keys())))
         outgoing = _computeoutgoing(repo, heads, common)
         return changegroup.makestream(repo, outgoing, '01', source,
                                       bundlecaps=bundlecaps)
 
     # bundle20 case
     b2caps = {}
     for bcaps in bundlecaps:
         if bcaps.startswith('bundle2='):
             blob = urlreq.unquote(bcaps[len('bundle2='):])
             b2caps.update(bundle2.decodecaps(blob))
     bundler = bundle2.bundle20(repo.ui, b2caps)
 
     kwargs['heads'] = heads
     kwargs['common'] = common
 
     for name in getbundle2partsorder:
         func = getbundle2partsmapping[name]
         func(bundler, repo, source, bundlecaps=bundlecaps, b2caps=b2caps,
              **pycompat.strkwargs(kwargs))
 
     return bundler.getchunks()
 
 @getbundle2partsgenerator('changegroup')
 def _getbundlechangegrouppart(bundler, repo, source, bundlecaps=None,
                               b2caps=None, heads=None, common=None, **kwargs):
     """add a changegroup part to the requested bundle"""
     cgstream = None
     if kwargs.get('cg', True):
         # build changegroup bundle here.
         version = '01'
         cgversions = b2caps.get('changegroup')
         if cgversions:  # 3.1 and 3.2 ship with an empty value
             cgversions = [v for v in cgversions
                           if v in changegroup.supportedoutgoingversions(repo)]
             if not cgversions:
                 raise ValueError(_('no common changegroup version'))
             version = max(cgversions)
         outgoing = _computeoutgoing(repo, heads, common)
         if outgoing.missing:
             cgstream = changegroup.makestream(repo, outgoing, version, source,
                                               bundlecaps=bundlecaps)
 
     if cgstream:
         part = bundler.newpart('changegroup', data=cgstream)
         if cgversions:
             part.addparam('version', version)
         part.addparam('nbchanges', '%d' % len(outgoing.missing),
                       mandatory=False)
         if 'treemanifest' in repo.requirements:
             part.addparam('treemanifest', '1')
 
+@getbundle2partsgenerator('bookmarks')
+def _getbundlebookmarkpart(bundler, repo, source, bundlecaps=None,
+                              b2caps=None, **kwargs):
+    """add a bookmark part to the requested bundle"""
+    if not kwargs.get('bookmarks', False):
+        return
+    if 'bookmarks' not in b2caps:
+        raise ValueError(_('no common bookmarks exchange method'))
+    books  = bookmod.listbinbookmarks(repo)
+    data = bookmod.binaryencode(books)
+    if data:
+        bundler.newpart('bookmarks', data=data)
+
 @getbundle2partsgenerator('listkeys')
 def _getbundlelistkeysparts(bundler, repo, source, bundlecaps=None,
                             b2caps=None, **kwargs):
     """add parts containing listkeys namespaces to the requested bundle"""
     listkeys = kwargs.get('listkeys', ())
     for namespace in listkeys:
         part = bundler.newpart('listkeys')
         part.addparam('namespace', namespace)
         keys = repo.listkeys(namespace).items()
         part.data = pushkey.encodekeys(keys)
 
 @getbundle2partsgenerator('obsmarkers')
 def _getbundleobsmarkerpart(bundler, repo, source, bundlecaps=None,
                             b2caps=None, heads=None, **kwargs):
     """add an obsolescence markers part to the requested bundle"""
     if kwargs.get('obsmarkers', False):
         if heads is None:
             heads = repo.heads()
         subset = [c.node() for c in repo.set('::%ln', heads)]
         markers = repo.obsstore.relevantmarkers(subset)
         markers = sorted(markers)
         bundle2.buildobsmarkerspart(bundler, markers)
 
 @getbundle2partsgenerator('phases')
 def _getbundlephasespart(bundler, repo, source, bundlecaps=None,
                             b2caps=None, heads=None, **kwargs):
     """add phase heads part to the requested bundle"""
     if kwargs.get('phases', False):
         if not 'heads' in b2caps.get('phases'):
             raise ValueError(_('no common phases exchange method'))
         if heads is None:
             heads = repo.heads()
 
         headsbyphase = collections.defaultdict(set)
         if repo.publishing():
             headsbyphase[phases.public] = heads
         else:
             # find the appropriate heads to move
 
             phase = repo._phasecache.phase
             node = repo.changelog.node
             rev = repo.changelog.rev
             for h in heads:
                 headsbyphase[phase(repo, rev(h))].add(h)
             seenphases = list(headsbyphase.keys())
 
             # We do not handle anything but public and draft phase for now)
             if seenphases:
                 assert max(seenphases) <= phases.draft
 
             # if client is pulling non-public changesets, we need to find
             # intermediate public heads.
             draftheads = headsbyphase.get(phases.draft, set())
             if draftheads:
                 publicheads = headsbyphase.get(phases.public, set())
 
                 revset = 'heads(only(%ln, %ln) and public())'
                 extraheads = repo.revs(revset, draftheads, publicheads)
                 for r in extraheads:
                     headsbyphase[phases.public].add(node(r))
 
         # transform data in a format used by the encoding function
         phasemapping = []
         for phase in phases.allphases:
             phasemapping.append(sorted(headsbyphase[phase]))
 
         # generate the actual part
         phasedata = phases.binaryencode(phasemapping)
         bundler.newpart('phase-heads', data=phasedata)
 
 @getbundle2partsgenerator('hgtagsfnodes')
 def _getbundletagsfnodes(bundler, repo, source, bundlecaps=None,
                          b2caps=None, heads=None, common=None,
                          **kwargs):
     """Transfer the .hgtags filenodes mapping.
 
     Only values for heads in this bundle will be transferred.
 
     The part data consists of pairs of 20 byte changeset node and .hgtags
     filenodes raw values.
     """
     # Don't send unless:
     # - changeset are being exchanged,
     # - the client supports it.
     if not (kwargs.get('cg', True) and 'hgtagsfnodes' in b2caps):
         return
 
     outgoing = _computeoutgoing(repo, heads, common)
     bundle2.addparttagsfnodescache(repo, bundler, outgoing)
 
 def check_heads(repo, their_heads, context):
     """check if the heads of a repo have been modified
 
     Used by peer for unbundling.
     """
     heads = repo.heads()
     heads_hash = hashlib.sha1(''.join(sorted(heads))).digest()
     if not (their_heads == ['force'] or their_heads == heads or
             their_heads == ['hashed', heads_hash]):
         # someone else committed/pushed/unbundled while we
         # were transferring data
         raise error.PushRaced('repository changed while %s - '
                               'please try again' % context)
 
 def unbundle(repo, cg, heads, source, url):
     """Apply a bundle to a repo.
 
     this function makes sure the repo is locked during the application and have
     mechanism to check that no push race occurred between the creation of the
     bundle and its application.
 
     If the push was raced as PushRaced exception is raised."""
     r = 0
     # need a transaction when processing a bundle2 stream
     # [wlock, lock, tr] - needs to be an array so nested functions can modify it
     lockandtr = [None, None, None]
     recordout = None
     # quick fix for output mismatch with bundle2 in 3.4
     captureoutput = repo.ui.configbool('experimental', 'bundle2-output-capture')
     if url.startswith('remote:http:') or url.startswith('remote:https:'):
         captureoutput = True
     try:
         # note: outside bundle1, 'heads' is expected to be empty and this
         # 'check_heads' call wil be a no-op
         check_heads(repo, heads, 'uploading changes')
         # push can proceed
         if not isinstance(cg, bundle2.unbundle20):
             # legacy case: bundle1 (changegroup 01)
             txnname = "\n".join([source, util.hidepassword(url)])
             with repo.lock(), repo.transaction(txnname) as tr:
                 op = bundle2.applybundle(repo, cg, tr, source, url)
                 r = bundle2.combinechangegroupresults(op)
         else:
             r = None
             try:
                 def gettransaction():
                     if not lockandtr[2]:
                         lockandtr[0] = repo.wlock()
                         lockandtr[1] = repo.lock()
                         lockandtr[2] = repo.transaction(source)
                         lockandtr[2].hookargs['source'] = source
                         lockandtr[2].hookargs['url'] = url
                         lockandtr[2].hookargs['bundle2'] = '1'
                     return lockandtr[2]
 
                 # Do greedy locking by default until we're satisfied with lazy
                 # locking.
                 if not repo.ui.configbool('experimental', 'bundle2lazylocking'):
                     gettransaction()
 
                 op = bundle2.bundleoperation(repo, gettransaction,
                                              captureoutput=captureoutput)
                 try:
                     op = bundle2.processbundle(repo, cg, op=op)
                 finally:
                     r = op.reply
                     if captureoutput and r is not None:
                         repo.ui.pushbuffer(error=True, subproc=True)
                         def recordout(output):
                             r.newpart('output', data=output, mandatory=False)
                 if lockandtr[2] is not None:
                     lockandtr[2].close()
             except BaseException as exc:
                 exc.duringunbundle2 = True
                 if captureoutput and r is not None:
                     parts = exc._bundle2salvagedoutput = r.salvageoutput()
                     def recordout(output):
                         part = bundle2.bundlepart('output', data=output,
                                                   mandatory=False)
                         parts.append(part)
                 raise
     finally:
         lockmod.release(lockandtr[2], lockandtr[1], lockandtr[0])
         if recordout is not None:
             recordout(repo.ui.popbuffer())
     return r
 
 def _maybeapplyclonebundle(pullop):
     """Apply a clone bundle from a remote, if possible."""
 
     repo = pullop.repo
     remote = pullop.remote
 
     if not repo.ui.configbool('ui', 'clonebundles'):
         return
 
     # Only run if local repo is empty.
     if len(repo):
         return
 
     if pullop.heads:
         return
 
     if not remote.capable('clonebundles'):
         return
 
     res = remote._call('clonebundles')
 
     # If we call the wire protocol command, that's good enough to record the
     # attempt.
     pullop.clonebundleattempted = True
 
     entries = parseclonebundlesmanifest(repo, res)
     if not entries:
         repo.ui.note(_('no clone bundles available on remote; '
                        'falling back to regular clone\n'))
         return
 
     entries = filterclonebundleentries(
         repo, entries, streamclonerequested=pullop.streamclonerequested)
 
     if not entries:
         # There is a thundering herd concern here. However, if a server
         # operator doesn't advertise bundles appropriate for its clients,
         # they deserve what's coming. Furthermore, from a client's
         # perspective, no automatic fallback would mean not being able to
         # clone!
         repo.ui.warn(_('no compatible clone bundles available on server; '
                        'falling back to regular clone\n'))
         repo.ui.warn(_('(you may want to report this to the server '
                        'operator)\n'))
         return
 
     entries = sortclonebundleentries(repo.ui, entries)
 
     url = entries[0]['URL']
     repo.ui.status(_('applying clone bundle from %s\n') % url)
     if trypullbundlefromurl(repo.ui, repo, url):
         repo.ui.status(_('finished applying clone bundle\n'))
     # Bundle failed.
     #
     # We abort by default to avoid the thundering herd of
     # clients flooding a server that was expecting expensive
     # clone load to be offloaded.
     elif repo.ui.configbool('ui', 'clonebundlefallback'):
         repo.ui.warn(_('falling back to normal clone\n'))
     else:
         raise error.Abort(_('error applying bundle'),
                           hint=_('if this error persists, consider contacting '
                                  'the server operator or disable clone '
                                  'bundles via '
                                  '"--config ui.clonebundles=false"'))
 
 def parseclonebundlesmanifest(repo, s):
     """Parses the raw text of a clone bundles manifest.
 
     Returns a list of dicts. The dicts have a ``URL`` key corresponding
     to the URL and other keys are the attributes for the entry.
     """
     m = []
     for line in s.splitlines():
         fields = line.split()
         if not fields:
             continue
         attrs = {'URL': fields[0]}
         for rawattr in fields[1:]:
             key, value = rawattr.split('=', 1)
             key = urlreq.unquote(key)
             value = urlreq.unquote(value)
             attrs[key] = value
 
             # Parse BUNDLESPEC into components. This makes client-side
             # preferences easier to specify since you can prefer a single
             # component of the BUNDLESPEC.
             if key == 'BUNDLESPEC':
                 try:
                     comp, version, params = parsebundlespec(repo, value,
                                                             externalnames=True)
                     attrs['COMPRESSION'] = comp
                     attrs['VERSION'] = version
                 except error.InvalidBundleSpecification:
                     pass
                 except error.UnsupportedBundleSpecification:
                     pass
 
         m.append(attrs)
 
     return m
 
 def filterclonebundleentries(repo, entries, streamclonerequested=False):
     """Remove incompatible clone bundle manifest entries.
 
     Accepts a list of entries parsed with ``parseclonebundlesmanifest``
     and returns a new list consisting of only the entries that this client
     should be able to apply.
 
     There is no guarantee we'll be able to apply all returned entries because
     the metadata we use to filter on may be missing or wrong.
     """
     newentries = []
     for entry in entries:
         spec = entry.get('BUNDLESPEC')
         if spec:
             try:
                 comp, version, params = parsebundlespec(repo, spec, strict=True)
 
                 # If a stream clone was requested, filter out non-streamclone
                 # entries.
                 if streamclonerequested and (comp != 'UN' or version != 's1'):
                     repo.ui.debug('filtering %s because not a stream clone\n' %
                                   entry['URL'])
                     continue
 
             except error.InvalidBundleSpecification as e:
                 repo.ui.debug(str(e) + '\n')
                 continue
             except error.UnsupportedBundleSpecification as e:
                 repo.ui.debug('filtering %s because unsupported bundle '
                               'spec: %s\n' % (entry['URL'], str(e)))
                 continue
         # If we don't have a spec and requested a stream clone, we don't know
         # what the entry is so don't attempt to apply it.
         elif streamclonerequested:
             repo.ui.debug('filtering %s because cannot determine if a stream '
                           'clone bundle\n' % entry['URL'])
             continue
 
         if 'REQUIRESNI' in entry and not sslutil.hassni:
             repo.ui.debug('filtering %s because SNI not supported\n' %
                           entry['URL'])
             continue
 
         newentries.append(entry)
 
     return newentries
 
 class clonebundleentry(object):
     """Represents an item in a clone bundles manifest.
 
     This rich class is needed to support sorting since sorted() in Python 3
     doesn't support ``cmp`` and our comparison is complex enough that ``key=``
     won't work.
     """
 
     def __init__(self, value, prefers):
         self.value = value
         self.prefers = prefers
 
     def _cmp(self, other):
         for prefkey, prefvalue in self.prefers:
             avalue = self.value.get(prefkey)
             bvalue = other.value.get(prefkey)
 
             # Special case for b missing attribute and a matches exactly.
             if avalue is not None and bvalue is None and avalue == prefvalue:
                 return -1
 
             # Special case for a missing attribute and b matches exactly.
             if bvalue is not None and avalue is None and bvalue == prefvalue:
                 return 1
 
             # We can't compare unless attribute present on both.
             if avalue is None or bvalue is None:
                 continue
 
             # Same values should fall back to next attribute.
             if avalue == bvalue:
                 continue
 
             # Exact matches come first.
             if avalue == prefvalue:
                 return -1
             if bvalue == prefvalue:
                 return 1
 
             # Fall back to next attribute.
             continue
 
         # If we got here we couldn't sort by attributes and prefers. Fall
         # back to index order.
         return 0
 
     def __lt__(self, other):
         return self._cmp(other) < 0
 
     def __gt__(self, other):
         return self._cmp(other) > 0
 
     def __eq__(self, other):
         return self._cmp(other) == 0
 
     def __le__(self, other):
         return self._cmp(other) <= 0
 
     def __ge__(self, other):
         return self._cmp(other) >= 0
 
     def __ne__(self, other):
         return self._cmp(other) != 0
 
 def sortclonebundleentries(ui, entries):
     prefers = ui.configlist('ui', 'clonebundleprefers')
     if not prefers:
         return list(entries)
 
     prefers = [p.split('=', 1) for p in prefers]
 
     items = sorted(clonebundleentry(v, prefers) for v in entries)
     return [i.value for i in items]
 
 def trypullbundlefromurl(ui, repo, url):
     """Attempt to apply a bundle from a URL."""
     with repo.lock(), repo.transaction('bundleurl') as tr:
         try:
             fh = urlmod.open(ui, url)
             cg = readbundle(ui, fh, 'stream')
 
             if isinstance(cg, streamclone.streamcloneapplier):
                 cg.apply(repo)
             else:
                 bundle2.applybundle(repo, cg, tr, 'clonebundles', url)
             return True
         except urlerr.httperror as e:
             ui.warn(_('HTTP error fetching bundle: %s\n') % str(e))
         except urlerr.urlerror as e:
             ui.warn(_('error fetching bundle: %s\n') % e.reason)
 
         return False

mercurial/help/internals/wireprotocol.txt

@@ -1,923 +1,925 b''
 The Mercurial wire protocol is a request-response based protocol
 with multiple wire representations.
 
 Each request is modeled as a command name, a dictionary of arguments, and
 optional raw input. Command arguments and their types are intrinsic
 properties of commands. So is the response type of the command. This means
 clients can't always send arbitrary arguments to servers and servers can't
 return multiple response types.
 
 The protocol is synchronous and does not support multiplexing (concurrent
 commands).
 
 Transport Protocols
 ===================
 
 HTTP Transport
 --------------
 
 Commands are issued as HTTP/1.0 or HTTP/1.1 requests. Commands are
 sent to the base URL of the repository with the command name sent in
 the ``cmd`` query string parameter. e.g.
 ``https://example.com/repo?cmd=capabilities``. The HTTP method is ``GET``
 or ``POST`` depending on the command and whether there is a request
 body.
 
 Command arguments can be sent multiple ways.
 
 The simplest is part of the URL query string using ``x-www-form-urlencoded``
 encoding (see Python's ``urllib.urlencode()``. However, many servers impose
 length limitations on the URL. So this mechanism is typically only used if
 the server doesn't support other mechanisms.
 
 If the server supports the ``httpheader`` capability, command arguments can
 be sent in HTTP request headers named ``X-HgArg-<N>`` where ``<N>`` is an
 integer starting at 1. A ``x-www-form-urlencoded`` representation of the
 arguments is obtained. This full string is then split into chunks and sent
 in numbered ``X-HgArg-<N>`` headers. The maximum length of each HTTP header
 is defined by the server in the ``httpheader`` capability value, which defaults
 to ``1024``. The server reassembles the encoded arguments string by
 concatenating the ``X-HgArg-<N>`` headers then URL decodes them into a
 dictionary.
 
 The list of ``X-HgArg-<N>`` headers should be added to the ``Vary`` request
 header to instruct caches to take these headers into consideration when caching
 requests.
 
 If the server supports the ``httppostargs`` capability, the client
 may send command arguments in the HTTP request body as part of an
 HTTP POST request. The command arguments will be URL encoded just like
 they would for sending them via HTTP headers. However, no splitting is
 performed: the raw arguments are included in the HTTP request body.
 
 The client sends a ``X-HgArgs-Post`` header with the string length of the
 encoded arguments data. Additional data may be included in the HTTP
 request body immediately following the argument data. The offset of the
 non-argument data is defined by the ``X-HgArgs-Post`` header. The
 ``X-HgArgs-Post`` header is not required if there is no argument data.
 
 Additional command data can be sent as part of the HTTP request body. The
 default ``Content-Type`` when sending data is ``application/mercurial-0.1``.
 A ``Content-Length`` header is currently always sent.
 
 Example HTTP requests::
 
     GET /repo?cmd=capabilities
     X-HgArg-1: foo=bar&baz=hello%20world
 
 The request media type should be chosen based on server support. If the
 ``httpmediatype`` server capability is present, the client should send
 the newest mutually supported media type. If this capability is absent,
 the client must assume the server only supports the
 ``application/mercurial-0.1`` media type.
 
 The ``Content-Type`` HTTP response header identifies the response as coming
 from Mercurial and can also be used to signal an error has occurred.
 
 The ``application/mercurial-*`` media types indicate a generic Mercurial
 data type.
 
 The ``application/mercurial-0.1`` media type is raw Mercurial data. It is the
 predecessor of the format below.
 
 The ``application/mercurial-0.2`` media type is compression framed Mercurial
 data. The first byte of the payload indicates the length of the compression
 format identifier that follows. Next are N bytes indicating the compression
 format. e.g. ``zlib``. The remaining bytes are compressed according to that
 compression format. The decompressed data behaves the same as with
 ``application/mercurial-0.1``.
 
 The ``application/hg-error`` media type indicates a generic error occurred.
 The content of the HTTP response body typically holds text describing the
 error.
 
 The ``application/hg-changegroup`` media type indicates a changegroup response
 type.
 
 Clients also accept the ``text/plain`` media type. All other media
 types should cause the client to error.
 
 Behavior of media types is further described in the ``Content Negotiation``
 section below.
 
 Clients should issue a ``User-Agent`` request header that identifies the client.
 The server should not use the ``User-Agent`` for feature detection.
 
 A command returning a ``string`` response issues a
 ``application/mercurial-0.*`` media type and the HTTP response body contains
 the raw string value (after compression decoding, if used). A
 ``Content-Length`` header is typically issued, but not required.
 
 A command returning a ``stream`` response issues a
 ``application/mercurial-0.*`` media type and the HTTP response is typically
 using *chunked transfer* (``Transfer-Encoding: chunked``).
 
 SSH Transport
 =============
 
 The SSH transport is a custom text-based protocol suitable for use over any
 bi-directional stream transport. It is most commonly used with SSH.
 
 A SSH transport server can be started with ``hg serve --stdio``. The stdin,
 stderr, and stdout file descriptors of the started process are used to exchange
 data. When Mercurial connects to a remote server over SSH, it actually starts
 a ``hg serve --stdio`` process on the remote server.
 
 Commands are issued by sending the command name followed by a trailing newline
 ``\n`` to the server. e.g. ``capabilities\n``.
 
 Command arguments are sent in the following format::
 
     <argument> <length>\n<value>
 
 That is, the argument string name followed by a space followed by the
 integer length of the value (expressed as a string) followed by a newline
 (``\n``) followed by the raw argument value.
 
 Dictionary arguments are encoded differently::
 
     <argument> <# elements>\n
     <key1> <length1>\n<value1>
     <key2> <length2>\n<value2>
     ...
 
 Non-argument data is sent immediately after the final argument value. It is
 encoded in chunks::
 
     <length>\n<data>
 
 Each command declares a list of supported arguments and their types. If a
 client sends an unknown argument to the server, the server should abort
 immediately. The special argument ``*`` in a command's definition indicates
 that all argument names are allowed.
 
 The definition of supported arguments and types is initially made when a
 new command is implemented. The client and server must initially independently
 agree on the arguments and their types. This initial set of arguments can be
 supplemented through the presence of *capabilities* advertised by the server.
 
 Each command has a defined expected response type.
 
 A ``string`` response type is a length framed value. The response consists of
 the string encoded integer length of a value followed by a newline (``\n``)
 followed by the value. Empty values are allowed (and are represented as
 ``0\n``).
 
 A ``stream`` response type consists of raw bytes of data. There is no framing.
 
 A generic error response type is also supported. It consists of a an error
 message written to ``stderr`` followed by ``\n-\n``. In addition, ``\n`` is
 written to ``stdout``.
 
 If the server receives an unknown command, it will send an empty ``string``
 response.
 
 The server terminates if it receives an empty command (a ``\n`` character).
 
 Capabilities
 ============
 
 Servers advertise supported wire protocol features. This allows clients to
 probe for server features before blindly calling a command or passing a
 specific argument.
 
 The server's features are exposed via a *capabilities* string. This is a
 space-delimited string of tokens/features. Some features are single words
 like ``lookup`` or ``batch``. Others are complicated key-value pairs
 advertising sub-features. e.g. ``httpheader=2048``. When complex, non-word
 values are used, each feature name can define its own encoding of sub-values.
 Comma-delimited and ``x-www-form-urlencoded`` values are common.
 
 The following document capabilities defined by the canonical Mercurial server
 implementation.
 
 batch
 -----
 
 Whether the server supports the ``batch`` command.
 
 This capability/command was introduced in Mercurial 1.9 (released July 2011).
 
 branchmap
 ---------
 
 Whether the server supports the ``branchmap`` command.
 
 This capability/command was introduced in Mercurial 1.3 (released July 2009).
 
 bundle2-exp
 -----------
 
 Precursor to ``bundle2`` capability that was used before bundle2 was a
 stable feature.
 
 This capability was introduced in Mercurial 3.0 behind an experimental
 flag. This capability should not be observed in the wild.
 
 bundle2
 -------
 
 Indicates whether the server supports the ``bundle2`` data exchange format.
 
 The value of the capability is a URL quoted, newline (``\n``) delimited
 list of keys or key-value pairs.
 
 A key is simply a URL encoded string.
 
 A key-value pair is a URL encoded key separated from a URL encoded value by
 an ``=``. If the value is a list, elements are delimited by a ``,`` after
 URL encoding.
 
 For example, say we have the values::
 
   {'HG20': [], 'changegroup': ['01', '02'], 'digests': ['sha1', 'sha512']}
 
 We would first construct a string::
 
   HG20\nchangegroup=01,02\ndigests=sha1,sha512
 
 We would then URL quote this string::
 
   HG20%0Achangegroup%3D01%2C02%0Adigests%3Dsha1%2Csha512
 
 This capability was introduced in Mercurial 3.4 (released May 2015).
 
 changegroupsubset
 -----------------
 
 Whether the server supports the ``changegroupsubset`` command.
 
 This capability was introduced in Mercurial 0.9.2 (released December
 2006).
 
 This capability was introduced at the same time as the ``lookup``
 capability/command.
 
 compression
 -----------
 
 Declares support for negotiating compression formats.
 
 Presence of this capability indicates the server supports dynamic selection
 of compression formats based on the client request.
 
 Servers advertising this capability are required to support the
 ``application/mercurial-0.2`` media type in response to commands returning
 streams. Servers may support this media type on any command.
 
 The value of the capability is a comma-delimited list of strings declaring
 supported compression formats. The order of the compression formats is in
 server-preferred order, most preferred first.
 
 The identifiers used by the official Mercurial distribution are:
 
 bzip2
    bzip2
 none
    uncompressed / raw data
 zlib
    zlib (no gzip header)
 zstd
    zstd
 
 This capability was introduced in Mercurial 4.1 (released February 2017).
 
 getbundle
 ---------
 
 Whether the server supports the ``getbundle`` command.
 
 This capability was introduced in Mercurial 1.9 (released July 2011).
 
 httpheader
 ----------
 
 Whether the server supports receiving command arguments via HTTP request
 headers.
 
 The value of the capability is an integer describing the max header
 length that clients should send. Clients should ignore any content after a
 comma in the value, as this is reserved for future use.
 
 This capability was introduced in Mercurial 1.9 (released July 2011).
 
 httpmediatype
 -------------
 
 Indicates which HTTP media types (``Content-Type`` header) the server is
 capable of receiving and sending.
 
 The value of the capability is a comma-delimited list of strings identifying
 support for media type and transmission direction. The following strings may
 be present:
 
 0.1rx
    Indicates server support for receiving ``application/mercurial-0.1`` media
    types.
 
 0.1tx
    Indicates server support for sending ``application/mercurial-0.1`` media
    types.
 
 0.2rx
    Indicates server support for receiving ``application/mercurial-0.2`` media
    types.
 
 0.2tx
    Indicates server support for sending ``application/mercurial-0.2`` media
    types.
 
 minrx=X
    Minimum media type version the server is capable of receiving. Value is a
    string like ``0.2``.
 
    This capability can be used by servers to limit connections from legacy
    clients not using the latest supported media type. However, only clients
    with knowledge of this capability will know to consult this value. This
    capability is present so the client may issue a more user-friendly error
    when the server has locked out a legacy client.
 
 mintx=X
    Minimum media type version the server is capable of sending. Value is a
    string like ``0.1``.
 
 Servers advertising support for the ``application/mercurial-0.2`` media type
 should also advertise the ``compression`` capability.
 
 This capability was introduced in Mercurial 4.1 (released February 2017).
 
 httppostargs
 ------------
 
 **Experimental**
 
 Indicates that the server supports and prefers clients send command arguments
 via a HTTP POST request as part of the request body.
 
 This capability was introduced in Mercurial 3.8 (released May 2016).
 
 known
 -----
 
 Whether the server supports the ``known`` command.
 
 This capability/command was introduced in Mercurial 1.9 (released July 2011).
 
 lookup
 ------
 
 Whether the server supports the ``lookup`` command.
 
 This capability was introduced in Mercurial 0.9.2 (released December
 2006).
 
 This capability was introduced at the same time as the ``changegroupsubset``
 capability/command.
 
 pushkey
 -------
 
 Whether the server supports the ``pushkey`` and ``listkeys`` commands.
 
 This capability was introduced in Mercurial 1.6 (released July 2010).
 
 standardbundle
 --------------
 
 **Unsupported**
 
 This capability was introduced during the Mercurial 0.9.2 development cycle in
 2006. It was never present in a release, as it was replaced by the ``unbundle``
 capability. This capability should not be encountered in the wild.
 
 stream-preferred
 ----------------
 
 If present the server prefers that clients clone using the streaming clone
 protocol (``hg clone --stream``) rather than the standard
 changegroup/bundle based protocol.
 
 This capability was introduced in Mercurial 2.2 (released May 2012).
 
 streamreqs
 ----------
 
 Indicates whether the server supports *streaming clones* and the *requirements*
 that clients must support to receive it.
 
 If present, the server supports the ``stream_out`` command, which transmits
 raw revlogs from the repository instead of changegroups. This provides a faster
 cloning mechanism at the expense of more bandwidth used.
 
 The value of this capability is a comma-delimited list of repo format
 *requirements*. These are requirements that impact the reading of data in
 the ``.hg/store`` directory. An example value is
 ``streamreqs=generaldelta,revlogv1`` indicating the server repo requires
 the ``revlogv1`` and ``generaldelta`` requirements.
 
 If the only format requirement is ``revlogv1``, the server may expose the
 ``stream`` capability instead of the ``streamreqs`` capability.
 
 This capability was introduced in Mercurial 1.7 (released November 2010).
 
 stream
 ------
 
 Whether the server supports *streaming clones* from ``revlogv1`` repos.
 
 If present, the server supports the ``stream_out`` command, which transmits
 raw revlogs from the repository instead of changegroups. This provides a faster
 cloning mechanism at the expense of more bandwidth used.
 
 This capability was introduced in Mercurial 0.9.1 (released July 2006).
 
 When initially introduced, the value of the capability was the numeric
 revlog revision. e.g. ``stream=1``. This indicates the changegroup is using
 ``revlogv1``. This simple integer value wasn't powerful enough, so the
 ``streamreqs`` capability was invented to handle cases where the repo
 requirements have more than just ``revlogv1``. Newer servers omit the
 ``=1`` since it was the only value supported and the value of ``1`` can
 be implied by clients.
 
 unbundlehash
 ------------
 
 Whether the ``unbundle`` commands supports receiving a hash of all the
 heads instead of a list.
 
 For more, see the documentation for the ``unbundle`` command.
 
 This capability was introduced in Mercurial 1.9 (released July 2011).
 
 unbundle
 --------
 
 Whether the server supports pushing via the ``unbundle`` command.
 
 This capability/command has been present since Mercurial 0.9.1 (released
 July 2006).
 
 Mercurial 0.9.2 (released December 2006) added values to the capability
 indicating which bundle types the server supports receiving. This value is a
 comma-delimited list. e.g. ``HG10GZ,HG10BZ,HG10UN``. The order of values
 reflects the priority/preference of that type, where the first value is the
 most preferred type.
 
 Handshake Protocol
 ==================
 
 While not explicitly required, it is common for clients to perform a
 *handshake* when connecting to a server. The handshake accomplishes 2 things:
 
 * Obtaining capabilities and other server features
 * Flushing extra server output (e.g. SSH servers may print extra text
   when connecting that may confuse the wire protocol)
 
 This isn't a traditional *handshake* as far as network protocols go because
 there is no persistent state as a result of the handshake: the handshake is
 simply the issuing of commands and commands are stateless.
 
 The canonical clients perform a capabilities lookup at connection establishment
 time. This is because clients must assume a server only supports the features
 of the original Mercurial server implementation until proven otherwise (from
 advertised capabilities). Nearly every server running today supports features
 that weren't present in the original Mercurial server implementation. Rather
 than wait for a client to perform functionality that needs to consult
 capabilities, it issues the lookup at connection start to avoid any delay later.
 
 For HTTP servers, the client sends a ``capabilities`` command request as
 soon as the connection is established. The server responds with a capabilities
 string, which the client parses.
 
 For SSH servers, the client sends the ``hello`` command (no arguments)
 and a ``between`` command with the ``pairs`` argument having the value
 ``0000000000000000000000000000000000000000-0000000000000000000000000000000000000000``.
 
 The ``between`` command has been supported since the original Mercurial
 server. Requesting the empty range will return a ``\n`` string response,
 which will be encoded as ``1\n\n`` (value length of ``1`` followed by a newline
 followed by the value, which happens to  be a newline).
 
 The ``hello`` command was later introduced. Servers supporting it will issue
 a response to that command before sending the ``1\n\n`` response to the
 ``between`` command. Servers not supporting ``hello`` will send an empty
 response (``0\n``).
 
 In addition to the expected output from the ``hello`` and ``between`` commands,
 servers may also send other output, such as *message of the day (MOTD)*
 announcements. Clients assume servers will send this output before the
 Mercurial server replies to the client-issued commands. So any server output
 not conforming to the expected command responses is assumed to be not related
 to Mercurial and can be ignored.
 
 Content Negotiation
 ===================
 
 The wire protocol has some mechanisms to help peers determine what content
 types and encoding the other side will accept. Historically, these mechanisms
 have been built into commands themselves because most commands only send a
 well-defined response type and only certain commands needed to support
 functionality like compression.
 
 Currently, only the HTTP transport supports content negotiation at the protocol
 layer.
 
 HTTP requests advertise supported response formats via the ``X-HgProto-<N>``
 request header, where ``<N>`` is an integer starting at 1 allowing the logical
 value to span multiple headers. This value consists of a list of
 space-delimited parameters. Each parameter denotes a feature or capability.
 
 The following parameters are defined:
 
 0.1
    Indicates the client supports receiving ``application/mercurial-0.1``
    responses.
 
 0.2
    Indicates the client supports receiving ``application/mercurial-0.2``
    responses.
 
 comp
    Indicates compression formats the client can decode. Value is a list of
    comma delimited strings identifying compression formats ordered from
    most preferential to least preferential. e.g. ``comp=zstd,zlib,none``.
 
    This parameter does not have an effect if only the ``0.1`` parameter
    is defined, as support for ``application/mercurial-0.2`` or greater is
    required to use arbitrary compression formats.
 
    If this parameter is not advertised, the server interprets this as
    equivalent to ``zlib,none``.
 
 Clients may choose to only send this header if the ``httpmediatype``
 server capability is present, as currently all server-side features
 consulting this header require the client to opt in to new protocol features
 advertised via the ``httpmediatype`` capability.
 
 A server that doesn't receive an ``X-HgProto-<N>`` header should infer a
 value of ``0.1``. This is compatible with legacy clients.
 
 A server receiving a request indicating support for multiple media type
 versions may respond with any of the supported media types. Not all servers
 may support all media types on all commands.
 
 Commands
 ========
 
 This section contains a list of all wire protocol commands implemented by
 the canonical Mercurial server.
 
 batch
 -----
 
 Issue multiple commands while sending a single command request. The purpose
 of this command is to allow a client to issue multiple commands while avoiding
 multiple round trips to the server therefore enabling commands to complete
 quicker.
 
 The command accepts a ``cmds`` argument that contains a list of commands to
 execute.
 
 The value of ``cmds`` is a ``;`` delimited list of strings. Each string has the
 form ``<command> <arguments>``. That is, the command name followed by a space
 followed by an argument string.
 
 The argument string is a ``,`` delimited list of ``<key>=<value>`` values
 corresponding to command arguments. Both the argument name and value are
 escaped using a special substitution map::
 
    : -> :c
    , -> :o
    ; -> :s
    = -> :e
 
 The response type for this command is ``string``. The value contains a
 ``;`` delimited list of responses for each requested command. Each value
 in this list is escaped using the same substitution map used for arguments.
 
 If an error occurs, the generic error response may be sent.
 
 between
 -------
 
 (Legacy command used for discovery in old clients)
 
 Obtain nodes between pairs of nodes.
 
 The ``pairs`` arguments contains a space-delimited list of ``-`` delimited
 hex node pairs. e.g.::
 
    a072279d3f7fd3a4aa7ffa1a5af8efc573e1c896-6dc58916e7c070f678682bfe404d2e2d68291a18
 
 Return type is a ``string``. Value consists of lines corresponding to each
 requested range. Each line contains a space-delimited list of hex nodes.
 A newline ``\n`` terminates each line, including the last one.
 
 branchmap
 ---------
 
 Obtain heads in named branches.
 
 Accepts no arguments. Return type is a ``string``.
 
 Return value contains lines with URL encoded branch names followed by a space
 followed by a space-delimited list of hex nodes of heads on that branch.
 e.g.::
 
     default a072279d3f7fd3a4aa7ffa1a5af8efc573e1c896 6dc58916e7c070f678682bfe404d2e2d68291a18
     stable baae3bf31522f41dd5e6d7377d0edd8d1cf3fccc
 
 There is no trailing newline.
 
 branches
 --------
 
 (Legacy command used for discovery in old clients. Clients with ``getbundle``
 use the ``known`` and ``heads`` commands instead.)
 
 Obtain ancestor changesets of specific nodes back to a branch point.
 
 Despite the name, this command has nothing to do with Mercurial named branches.
 Instead, it is related to DAG branches.
 
 The command accepts a ``nodes`` argument, which is a string of space-delimited
 hex nodes.
 
 For each node requested, the server will find the first ancestor node that is
 a DAG root or is a merge.
 
 Return type is a ``string``. Return value contains lines with result data for
 each requested node. Each line contains space-delimited nodes followed by a
 newline (``\n``). The 4 nodes reported on each line correspond to the requested
 node, the ancestor node found, and its 2 parent nodes (which may be the null
 node).
 
 capabilities
 ------------
 
 Obtain the capabilities string for the repo.
 
 Unlike the ``hello`` command, the capabilities string is not prefixed.
 There is no trailing newline.
 
 This command does not accept any arguments. Return type is a ``string``.
 
 changegroup
 -----------
 
 (Legacy command: use ``getbundle`` instead)
 
 Obtain a changegroup version 1 with data for changesets that are
 descendants of client-specified changesets.
 
 The ``roots`` arguments contains a list of space-delimited hex nodes.
 
 The server responds with a changegroup version 1 containing all
 changesets between the requested root/base nodes and the repo's head nodes
 at the time of the request.
 
 The return type is a ``stream``.
 
 changegroupsubset
 -----------------
 
 (Legacy command: use ``getbundle`` instead)
 
 Obtain a changegroup version 1 with data for changesetsets between
 client specified base and head nodes.
 
 The ``bases`` argument contains a list of space-delimited hex nodes.
 The ``heads`` argument contains a list of space-delimited hex nodes.
 
 The server responds with a changegroup version 1 containing all
 changesets between the requested base and head nodes at the time of the
 request.
 
 The return type is a ``stream``.
 
 clonebundles
 ------------
 
 Obtains a manifest of bundle URLs available to seed clones.
 
 Each returned line contains a URL followed by metadata. See the
 documentation in the ``clonebundles`` extension for more.
 
 The return type is a ``string``.
 
 getbundle
 ---------
 
 Obtain a bundle containing repository data.
 
 This command accepts the following arguments:
 
 heads
    List of space-delimited hex nodes of heads to retrieve.
 common
    List of space-delimited hex nodes that the client has in common with the
    server.
 obsmarkers
    Boolean indicating whether to include obsolescence markers as part
    of the response. Only works with bundle2.
 bundlecaps
    Comma-delimited set of strings defining client bundle capabilities.
 listkeys
    Comma-delimited list of strings of ``pushkey`` namespaces. For each
    namespace listed, a bundle2 part will be included with the content of
    that namespace.
 cg
    Boolean indicating whether changegroup data is requested.
 cbattempted
    Boolean indicating whether the client attempted to use the *clone bundles*
    feature before performing this request.
+bookmarks
+   Boolean indicating whether bookmark data is requested.
 phases
    Boolean indicating whether phases data is requested.
 
 The return type on success is a ``stream`` where the value is bundle.
 On the HTTP transport, the response is zlib compressed.
 
 If an error occurs, a generic error response can be sent.
 
 Unless the client sends a false value for the ``cg`` argument, the returned
 bundle contains a changegroup with the nodes between the specified ``common``
 and ``heads`` nodes. Depending on the command arguments, the type and content
 of the returned bundle can vary significantly.
 
 The default behavior is for the server to send a raw changegroup version
 ``01`` response.
 
 If the ``bundlecaps`` provided by the client contain a value beginning
 with ``HG2``, a bundle2 will be returned. The bundle2 data may contain
 additional repository data, such as ``pushkey`` namespace values.
 
 heads
 -----
 
 Returns a list of space-delimited hex nodes of repository heads followed
 by a newline. e.g.
 ``a9eeb3adc7ddb5006c088e9eda61791c777cbf7c 31f91a3da534dc849f0d6bfc00a395a97cf218a1\n``
 
 This command does not accept any arguments. The return type is a ``string``.
 
 hello
 -----
 
 Returns lines describing interesting things about the server in an RFC-822
 like format.
 
 Currently, the only line defines the server capabilities. It has the form::
 
     capabilities: <value>
 
 See above for more about the capabilities string.
 
 SSH clients typically issue this command as soon as a connection is
 established.
 
 This command does not accept any arguments. The return type is a ``string``.
 
 listkeys
 --------
 
 List values in a specified ``pushkey`` namespace.
 
 The ``namespace`` argument defines the pushkey namespace to operate on.
 
 The return type is a ``string``. The value is an encoded dictionary of keys.
 
 Key-value pairs are delimited by newlines (``\n``). Within each line, keys and
 values are separated by a tab (``\t``). Keys and values are both strings.
 
 lookup
 ------
 
 Try to resolve a value to a known repository revision.
 
 The ``key`` argument is converted from bytes to an
 ``encoding.localstr`` instance then passed into
 ``localrepository.__getitem__`` in an attempt to resolve it.
 
 The return type is a ``string``.
 
 Upon successful resolution, returns ``1 <hex node>\n``. On failure,
 returns ``0 <error string>\n``. e.g.::
 
    1 273ce12ad8f155317b2c078ec75a4eba507f1fba\n
 
    0 unknown revision 'foo'\n
 
 known
 -----
 
 Determine whether multiple nodes are known.
 
 The ``nodes`` argument is a list of space-delimited hex nodes to check
 for existence.
 
 The return type is ``string``.
 
 Returns a string consisting of ``0``s and ``1``s indicating whether nodes
 are known. If the Nth node specified in the ``nodes`` argument is known,
 a ``1`` will be returned at byte offset N. If the node isn't known, ``0``
 will be present at byte offset N.
 
 There is no trailing newline.
 
 pushkey
 -------
 
 Set a value using the ``pushkey`` protocol.
 
 Accepts arguments ``namespace``, ``key``, ``old``, and ``new``, which
 correspond to the pushkey namespace to operate on, the key within that
 namespace to change, the old value (which may be empty), and the new value.
 All arguments are string types.
 
 The return type is a ``string``. The value depends on the transport protocol.
 
 The SSH transport sends a string encoded integer followed by a newline
 (``\n``) which indicates operation result. The server may send additional
 output on the ``stderr`` stream that should be displayed to the user.
 
 The HTTP transport sends a string encoded integer followed by a newline
 followed by additional server output that should be displayed to the user.
 This may include output from hooks, etc.
 
 The integer result varies by namespace. ``0`` means an error has occurred
 and there should be additional output to display to the user.
 
 stream_out
 ----------
 
 Obtain *streaming clone* data.
 
 The return type is either a ``string`` or a ``stream``, depending on
 whether the request was fulfilled properly.
 
 A return value of ``1\n`` indicates the server is not configured to serve
 this data. If this is seen by the client, they may not have verified the
 ``stream`` capability is set before making the request.
 
 A return value of ``2\n`` indicates the server was unable to lock the
 repository to generate data.
 
 All other responses are a ``stream`` of bytes. The first line of this data
 contains 2 space-delimited integers corresponding to the path count and
 payload size, respectively::
 
     <path count> <payload size>\n
 
 The ``<payload size>`` is the total size of path data: it does not include
 the size of the per-path header lines.
 
 Following that header are ``<path count>`` entries. Each entry consists of a
 line with metadata followed by raw revlog data. The line consists of::
 
     <store path>\0<size>\n
 
 The ``<store path>`` is the encoded store path of the data that follows.
 ``<size>`` is the amount of data for this store path/revlog that follows the
 newline.
 
 There is no trailer to indicate end of data. Instead, the client should stop
 reading after ``<path count>`` entries are consumed.
 
 unbundle
 --------
 
 Send a bundle containing data (usually changegroup data) to the server.
 
 Accepts the argument ``heads``, which is a space-delimited list of hex nodes
 corresponding to server repository heads observed by the client. This is used
 to detect race conditions and abort push operations before a server performs
 too much work or a client transfers too much data.
 
 The request payload consists of a bundle to be applied to the repository,
 similarly to as if :hg:`unbundle` were called.
 
 In most scenarios, a special ``push response`` type is returned. This type
 contains an integer describing the change in heads as a result of the
 operation. A value of ``0`` indicates nothing changed. ``1`` means the number
 of heads remained the same. Values ``2`` and larger indicate the number of
 added heads minus 1. e.g. ``3`` means 2 heads were added. Negative values
 indicate the number of fewer heads, also off by 1. e.g. ``-2`` means there
 is 1 fewer head.
 
 The encoding of the ``push response`` type varies by transport.
 
 For the SSH transport, this type is composed of 2 ``string`` responses: an
 empty response (``0\n``) followed by the integer result value. e.g.
 ``1\n2``. So the full response might be ``0\n1\n2``.
 
 For the HTTP transport, the response is a ``string`` type composed of an
 integer result value followed by a newline (``\n``) followed by string
 content holding server output that should be displayed on the client (output
 hooks, etc).
 
 In some cases, the server may respond with a ``bundle2`` bundle. In this
 case, the response type is ``stream``. For the HTTP transport, the response
 is zlib compressed.
 
 The server may also respond with a generic error type, which contains a string
 indicating the failure.

mercurial/wireproto.py

@@ -1,1069 +1,1070 b''
 # wireproto.py - generic wire protocol support functions
 #
 # Copyright 2005-2010 Matt Mackall <[e-mail]>
 #
 # This software may be used and distributed according to the terms of the
 # GNU General Public License version 2 or any later version.
 
 from __future__ import absolute_import
 
 import hashlib
 import os
 import tempfile
 
 from .i18n import _
 from .node import (
     bin,
     hex,
     nullid,
 )
 
 from . import (
     bundle2,
     changegroup as changegroupmod,
     discovery,
     encoding,
     error,
     exchange,
     peer,
     pushkey as pushkeymod,
     pycompat,
     repository,
     streamclone,
     util,
 )
 
 urlerr = util.urlerr
 urlreq = util.urlreq
 
 bundle2requiredmain = _('incompatible Mercurial client; bundle2 required')
 bundle2requiredhint = _('see https://www.mercurial-scm.org/wiki/'
                         'IncompatibleClient')
 bundle2required = '%s\n(%s)\n' % (bundle2requiredmain, bundle2requiredhint)
 
 class abstractserverproto(object):
     """abstract class that summarizes the protocol API
 
     Used as reference and documentation.
     """
 
     def getargs(self, args):
         """return the value for arguments in <args>
 
         returns a list of values (same order as <args>)"""
         raise NotImplementedError()
 
     def getfile(self, fp):
         """write the whole content of a file into a file like object
 
         The file is in the form::
 
             (<chunk-size>\n<chunk>)+0\n
 
         chunk size is the ascii version of the int.
         """
         raise NotImplementedError()
 
     def redirect(self):
         """may setup interception for stdout and stderr
 
         See also the `restore` method."""
         raise NotImplementedError()
 
     # If the `redirect` function does install interception, the `restore`
     # function MUST be defined. If interception is not used, this function
     # MUST NOT be defined.
     #
     # left commented here on purpose
     #
     #def restore(self):
     #    """reinstall previous stdout and stderr and return intercepted stdout
     #    """
     #    raise NotImplementedError()
 
 class remoteiterbatcher(peer.iterbatcher):
     def __init__(self, remote):
         super(remoteiterbatcher, self).__init__()
         self._remote = remote
 
     def __getattr__(self, name):
         # Validate this method is batchable, since submit() only supports
         # batchable methods.
         fn = getattr(self._remote, name)
         if not getattr(fn, 'batchable', None):
             raise error.ProgrammingError('Attempted to batch a non-batchable '
                                          'call to %r' % name)
 
         return super(remoteiterbatcher, self).__getattr__(name)
 
     def submit(self):
         """Break the batch request into many patch calls and pipeline them.
 
         This is mostly valuable over http where request sizes can be
         limited, but can be used in other places as well.
         """
         # 2-tuple of (command, arguments) that represents what will be
         # sent over the wire.
         requests = []
 
         # 4-tuple of (command, final future, @batchable generator, remote
         # future).
         results = []
 
         for command, args, opts, finalfuture in self.calls:
             mtd = getattr(self._remote, command)
             batchable = mtd.batchable(mtd.__self__, *args, **opts)
 
             commandargs, fremote = next(batchable)
             assert fremote
             requests.append((command, commandargs))
             results.append((command, finalfuture, batchable, fremote))
 
         if requests:
             self._resultiter = self._remote._submitbatch(requests)
 
         self._results = results
 
     def results(self):
         for command, finalfuture, batchable, remotefuture in self._results:
             # Get the raw result, set it in the remote future, feed it
             # back into the @batchable generator so it can be decoded, and
             # set the result on the final future to this value.
             remoteresult = next(self._resultiter)
             remotefuture.set(remoteresult)
             finalfuture.set(next(batchable))
 
             # Verify our @batchable generators only emit 2 values.
             try:
                 next(batchable)
             except StopIteration:
                 pass
             else:
                 raise error.ProgrammingError('%s @batchable generator emitted '
                                              'unexpected value count' % command)
 
             yield finalfuture.value
 
 # Forward a couple of names from peer to make wireproto interactions
 # slightly more sensible.
 batchable = peer.batchable
 future = peer.future
 
 # list of nodes encoding / decoding
 
 def decodelist(l, sep=' '):
     if l:
         return [bin(v) for v in  l.split(sep)]
     return []
 
 def encodelist(l, sep=' '):
     try:
         return sep.join(map(hex, l))
     except TypeError:
         raise
 
 # batched call argument encoding
 
 def escapearg(plain):
     return (plain
             .replace(':', ':c')
             .replace(',', ':o')
             .replace(';', ':s')
             .replace('=', ':e'))
 
 def unescapearg(escaped):
     return (escaped
             .replace(':e', '=')
             .replace(':s', ';')
             .replace(':o', ',')
             .replace(':c', ':'))
 
 def encodebatchcmds(req):
     """Return a ``cmds`` argument value for the ``batch`` command."""
     cmds = []
     for op, argsdict in req:
         # Old servers didn't properly unescape argument names. So prevent
         # the sending of argument names that may not be decoded properly by
         # servers.
         assert all(escapearg(k) == k for k in argsdict)
 
         args = ','.join('%s=%s' % (escapearg(k), escapearg(v))
                         for k, v in argsdict.iteritems())
         cmds.append('%s %s' % (op, args))
 
     return ';'.join(cmds)
 
 # mapping of options accepted by getbundle and their types
 #
 # Meant to be extended by extensions. It is extensions responsibility to ensure
 # such options are properly processed in exchange.getbundle.
 #
 # supported types are:
 #
 # :nodes: list of binary nodes
 # :csv:   list of comma-separated values
 # :scsv:  list of comma-separated values return as set
 # :plain: string with no transformation needed.
 gboptsmap = {'heads':  'nodes',
+             'bookmarks': 'boolean',
              'common': 'nodes',
              'obsmarkers': 'boolean',
              'phases': 'boolean',
              'bundlecaps': 'scsv',
              'listkeys': 'csv',
              'cg': 'boolean',
              'cbattempted': 'boolean'}
 
 # client side
 
 class wirepeer(repository.legacypeer):
     """Client-side interface for communicating with a peer repository.
 
     Methods commonly call wire protocol commands of the same name.
 
     See also httppeer.py and sshpeer.py for protocol-specific
     implementations of this interface.
     """
     # Begin of basewirepeer interface.
 
     def iterbatch(self):
         return remoteiterbatcher(self)
 
     @batchable
     def lookup(self, key):
         self.requirecap('lookup', _('look up remote revision'))
         f = future()
         yield {'key': encoding.fromlocal(key)}, f
         d = f.value
         success, data = d[:-1].split(" ", 1)
         if int(success):
             yield bin(data)
         else:
             self._abort(error.RepoError(data))
 
     @batchable
     def heads(self):
         f = future()
         yield {}, f
         d = f.value
         try:
             yield decodelist(d[:-1])
         except ValueError:
             self._abort(error.ResponseError(_("unexpected response:"), d))
 
     @batchable
     def known(self, nodes):
         f = future()
         yield {'nodes': encodelist(nodes)}, f
         d = f.value
         try:
             yield [bool(int(b)) for b in d]
         except ValueError:
             self._abort(error.ResponseError(_("unexpected response:"), d))
 
     @batchable
     def branchmap(self):
         f = future()
         yield {}, f
         d = f.value
         try:
             branchmap = {}
             for branchpart in d.splitlines():
                 branchname, branchheads = branchpart.split(' ', 1)
                 branchname = encoding.tolocal(urlreq.unquote(branchname))
                 branchheads = decodelist(branchheads)
                 branchmap[branchname] = branchheads
             yield branchmap
         except TypeError:
             self._abort(error.ResponseError(_("unexpected response:"), d))
 
     @batchable
     def listkeys(self, namespace):
         if not self.capable('pushkey'):
             yield {}, None
         f = future()
         self.ui.debug('preparing listkeys for "%s"\n' % namespace)
         yield {'namespace': encoding.fromlocal(namespace)}, f
         d = f.value
         self.ui.debug('received listkey for "%s": %i bytes\n'
                       % (namespace, len(d)))
         yield pushkeymod.decodekeys(d)
 
     @batchable
     def pushkey(self, namespace, key, old, new):
         if not self.capable('pushkey'):
             yield False, None
         f = future()
         self.ui.debug('preparing pushkey for "%s:%s"\n' % (namespace, key))
         yield {'namespace': encoding.fromlocal(namespace),
                'key': encoding.fromlocal(key),
                'old': encoding.fromlocal(old),
                'new': encoding.fromlocal(new)}, f
         d = f.value
         d, output = d.split('\n', 1)
         try:
             d = bool(int(d))
         except ValueError:
             raise error.ResponseError(
                 _('push failed (unexpected response):'), d)
         for l in output.splitlines(True):
             self.ui.status(_('remote: '), l)
         yield d
 
     def stream_out(self):
         return self._callstream('stream_out')
 
     def getbundle(self, source, **kwargs):
         kwargs = pycompat.byteskwargs(kwargs)
         self.requirecap('getbundle', _('look up remote changes'))
         opts = {}
         bundlecaps = kwargs.get('bundlecaps')
         if bundlecaps is not None:
             kwargs['bundlecaps'] = sorted(bundlecaps)
         else:
             bundlecaps = () # kwargs could have it to None
         for key, value in kwargs.iteritems():
             if value is None:
                 continue
             keytype = gboptsmap.get(key)
             if keytype is None:
                 raise error.ProgrammingError(
                     'Unexpectedly None keytype for key %s' % key)
             elif keytype == 'nodes':
                 value = encodelist(value)
             elif keytype in ('csv', 'scsv'):
                 value = ','.join(value)
             elif keytype == 'boolean':
                 value = '%i' % bool(value)
             elif keytype != 'plain':
                 raise KeyError('unknown getbundle option type %s'
                                % keytype)
             opts[key] = value
         f = self._callcompressable("getbundle", **pycompat.strkwargs(opts))
         if any((cap.startswith('HG2') for cap in bundlecaps)):
             return bundle2.getunbundler(self.ui, f)
         else:
             return changegroupmod.cg1unpacker(f, 'UN')
 
     def unbundle(self, cg, heads, url):
         '''Send cg (a readable file-like object representing the
         changegroup to push, typically a chunkbuffer object) to the
         remote server as a bundle.
 
         When pushing a bundle10 stream, return an integer indicating the
         result of the push (see changegroup.apply()).
 
         When pushing a bundle20 stream, return a bundle20 stream.
 
         `url` is the url the client thinks it's pushing to, which is
         visible to hooks.
         '''
 
         if heads != ['force'] and self.capable('unbundlehash'):
             heads = encodelist(['hashed',
                                 hashlib.sha1(''.join(sorted(heads))).digest()])
         else:
             heads = encodelist(heads)
 
         if util.safehasattr(cg, 'deltaheader'):
             # this a bundle10, do the old style call sequence
             ret, output = self._callpush("unbundle", cg, heads=heads)
             if ret == "":
                 raise error.ResponseError(
                     _('push failed:'), output)
             try:
                 ret = int(ret)
             except ValueError:
                 raise error.ResponseError(
                     _('push failed (unexpected response):'), ret)
 
             for l in output.splitlines(True):
                 self.ui.status(_('remote: '), l)
         else:
             # bundle2 push. Send a stream, fetch a stream.
             stream = self._calltwowaystream('unbundle', cg, heads=heads)
             ret = bundle2.getunbundler(self.ui, stream)
         return ret
 
     # End of basewirepeer interface.
 
     # Begin of baselegacywirepeer interface.
 
     def branches(self, nodes):
         n = encodelist(nodes)
         d = self._call("branches", nodes=n)
         try:
             br = [tuple(decodelist(b)) for b in d.splitlines()]
             return br
         except ValueError:
             self._abort(error.ResponseError(_("unexpected response:"), d))
 
     def between(self, pairs):
         batch = 8 # avoid giant requests
         r = []
         for i in xrange(0, len(pairs), batch):
             n = " ".join([encodelist(p, '-') for p in pairs[i:i + batch]])
             d = self._call("between", pairs=n)
             try:
                 r.extend(l and decodelist(l) or [] for l in d.splitlines())
             except ValueError:
                 self._abort(error.ResponseError(_("unexpected response:"), d))
         return r
 
     def changegroup(self, nodes, kind):
         n = encodelist(nodes)
         f = self._callcompressable("changegroup", roots=n)
         return changegroupmod.cg1unpacker(f, 'UN')
 
     def changegroupsubset(self, bases, heads, kind):
         self.requirecap('changegroupsubset', _('look up remote changes'))
         bases = encodelist(bases)
         heads = encodelist(heads)
         f = self._callcompressable("changegroupsubset",
                                    bases=bases, heads=heads)
         return changegroupmod.cg1unpacker(f, 'UN')
 
     # End of baselegacywirepeer interface.
 
     def _submitbatch(self, req):
         """run batch request <req> on the server
 
         Returns an iterator of the raw responses from the server.
         """
         rsp = self._callstream("batch", cmds=encodebatchcmds(req))
         chunk = rsp.read(1024)
         work = [chunk]
         while chunk:
             while ';' not in chunk and chunk:
                 chunk = rsp.read(1024)
                 work.append(chunk)
             merged = ''.join(work)
             while ';' in merged:
                 one, merged = merged.split(';', 1)
                 yield unescapearg(one)
             chunk = rsp.read(1024)
             work = [merged, chunk]
         yield unescapearg(''.join(work))
 
     def _submitone(self, op, args):
         return self._call(op, **pycompat.strkwargs(args))
 
     def debugwireargs(self, one, two, three=None, four=None, five=None):
         # don't pass optional arguments left at their default value
         opts = {}
         if three is not None:
             opts['three'] = three
         if four is not None:
             opts['four'] = four
         return self._call('debugwireargs', one=one, two=two, **opts)
 
     def _call(self, cmd, **args):
         """execute <cmd> on the server
 
         The command is expected to return a simple string.
 
         returns the server reply as a string."""
         raise NotImplementedError()
 
     def _callstream(self, cmd, **args):
         """execute <cmd> on the server
 
         The command is expected to return a stream. Note that if the
         command doesn't return a stream, _callstream behaves
         differently for ssh and http peers.
 
         returns the server reply as a file like object.
         """
         raise NotImplementedError()
 
     def _callcompressable(self, cmd, **args):
         """execute <cmd> on the server
 
         The command is expected to return a stream.
 
         The stream may have been compressed in some implementations. This
         function takes care of the decompression. This is the only difference
         with _callstream.
 
         returns the server reply as a file like object.
         """
         raise NotImplementedError()
 
     def _callpush(self, cmd, fp, **args):
         """execute a <cmd> on server
 
         The command is expected to be related to a push. Push has a special
         return method.
 
         returns the server reply as a (ret, output) tuple. ret is either
         empty (error) or a stringified int.
         """
         raise NotImplementedError()
 
     def _calltwowaystream(self, cmd, fp, **args):
         """execute <cmd> on server
 
         The command will send a stream to the server and get a stream in reply.
         """
         raise NotImplementedError()
 
     def _abort(self, exception):
         """clearly abort the wire protocol connection and raise the exception
         """
         raise NotImplementedError()
 
 # server side
 
 # wire protocol command can either return a string or one of these classes.
 class streamres(object):
     """wireproto reply: binary stream
 
     The call was successful and the result is a stream.
 
     Accepts either a generator or an object with a ``read(size)`` method.
 
     ``v1compressible`` indicates whether this data can be compressed to
     "version 1" clients (technically: HTTP peers using
     application/mercurial-0.1 media type). This flag should NOT be used on
     new commands because new clients should support a more modern compression
     mechanism.
     """
     def __init__(self, gen=None, reader=None, v1compressible=False):
         self.gen = gen
         self.reader = reader
         self.v1compressible = v1compressible
 
 class pushres(object):
     """wireproto reply: success with simple integer return
 
     The call was successful and returned an integer contained in `self.res`.
     """
     def __init__(self, res):
         self.res = res
 
 class pusherr(object):
     """wireproto reply: failure
 
     The call failed. The `self.res` attribute contains the error message.
     """
     def __init__(self, res):
         self.res = res
 
 class ooberror(object):
     """wireproto reply: failure of a batch of operation
 
     Something failed during a batch call. The error message is stored in
     `self.message`.
     """
     def __init__(self, message):
         self.message = message
 
 def getdispatchrepo(repo, proto, command):
     """Obtain the repo used for processing wire protocol commands.
 
     The intent of this function is to serve as a monkeypatch point for
     extensions that need commands to operate on different repo views under
     specialized circumstances.
     """
     return repo.filtered('served')
 
 def dispatch(repo, proto, command):
     repo = getdispatchrepo(repo, proto, command)
     func, spec = commands[command]
     args = proto.getargs(spec)
     return func(repo, proto, *args)
 
 def options(cmd, keys, others):
     opts = {}
     for k in keys:
         if k in others:
             opts[k] = others[k]
             del others[k]
     if others:
         util.stderr.write("warning: %s ignored unexpected arguments %s\n"
                           % (cmd, ",".join(others)))
     return opts
 
 def bundle1allowed(repo, action):
     """Whether a bundle1 operation is allowed from the server.
 
     Priority is:
 
     1. server.bundle1gd.<action> (if generaldelta active)
     2. server.bundle1.<action>
     3. server.bundle1gd (if generaldelta active)
     4. server.bundle1
     """
     ui = repo.ui
     gd = 'generaldelta' in repo.requirements
 
     if gd:
         v = ui.configbool('server', 'bundle1gd.%s' % action)
         if v is not None:
             return v
 
     v = ui.configbool('server', 'bundle1.%s' % action)
     if v is not None:
         return v
 
     if gd:
         v = ui.configbool('server', 'bundle1gd')
         if v is not None:
             return v
 
     return ui.configbool('server', 'bundle1')
 
 def supportedcompengines(ui, proto, role):
     """Obtain the list of supported compression engines for a request."""
     assert role in (util.CLIENTROLE, util.SERVERROLE)
 
     compengines = util.compengines.supportedwireengines(role)
 
     # Allow config to override default list and ordering.
     if role == util.SERVERROLE:
         configengines = ui.configlist('server', 'compressionengines')
         config = 'server.compressionengines'
     else:
         # This is currently implemented mainly to facilitate testing. In most
         # cases, the server should be in charge of choosing a compression engine
         # because a server has the most to lose from a sub-optimal choice. (e.g.
         # CPU DoS due to an expensive engine or a network DoS due to poor
         # compression ratio).
         configengines = ui.configlist('experimental',
                                       'clientcompressionengines')
         config = 'experimental.clientcompressionengines'
 
     # No explicit config. Filter out the ones that aren't supposed to be
     # advertised and return default ordering.
     if not configengines:
         attr = 'serverpriority' if role == util.SERVERROLE else 'clientpriority'
         return [e for e in compengines
                 if getattr(e.wireprotosupport(), attr) > 0]
 
     # If compression engines are listed in the config, assume there is a good
     # reason for it (like server operators wanting to achieve specific
     # performance characteristics). So fail fast if the config references
     # unusable compression engines.
     validnames = set(e.name() for e in compengines)
     invalidnames = set(e for e in configengines if e not in validnames)
     if invalidnames:
         raise error.Abort(_('invalid compression engine defined in %s: %s') %
                           (config, ', '.join(sorted(invalidnames))))
 
     compengines = [e for e in compengines if e.name() in configengines]
     compengines = sorted(compengines,
                          key=lambda e: configengines.index(e.name()))
 
     if not compengines:
         raise error.Abort(_('%s config option does not specify any known '
                             'compression engines') % config,
                           hint=_('usable compression engines: %s') %
                           ', '.sorted(validnames))
 
     return compengines
 
 # list of commands
 commands = {}
 
 def wireprotocommand(name, args=''):
     """decorator for wire protocol command"""
     def register(func):
         commands[name] = (func, args)
         return func
     return register
 
 @wireprotocommand('batch', 'cmds *')
 def batch(repo, proto, cmds, others):
     repo = repo.filtered("served")
     res = []
     for pair in cmds.split(';'):
         op, args = pair.split(' ', 1)
         vals = {}
         for a in args.split(','):
             if a:
                 n, v = a.split('=')
                 vals[unescapearg(n)] = unescapearg(v)
         func, spec = commands[op]
         if spec:
             keys = spec.split()
             data = {}
             for k in keys:
                 if k == '*':
                     star = {}
                     for key in vals.keys():
                         if key not in keys:
                             star[key] = vals[key]
                     data['*'] = star
                 else:
                     data[k] = vals[k]
             result = func(repo, proto, *[data[k] for k in keys])
         else:
             result = func(repo, proto)
         if isinstance(result, ooberror):
             return result
         res.append(escapearg(result))
     return ';'.join(res)
 
 @wireprotocommand('between', 'pairs')
 def between(repo, proto, pairs):
     pairs = [decodelist(p, '-') for p in pairs.split(" ")]
     r = []
     for b in repo.between(pairs):
         r.append(encodelist(b) + "\n")
     return "".join(r)
 
 @wireprotocommand('branchmap')
 def branchmap(repo, proto):
     branchmap = repo.branchmap()
     heads = []
     for branch, nodes in branchmap.iteritems():
         branchname = urlreq.quote(encoding.fromlocal(branch))
         branchnodes = encodelist(nodes)
         heads.append('%s %s' % (branchname, branchnodes))
     return '\n'.join(heads)
 
 @wireprotocommand('branches', 'nodes')
 def branches(repo, proto, nodes):
     nodes = decodelist(nodes)
     r = []
     for b in repo.branches(nodes):
         r.append(encodelist(b) + "\n")
     return "".join(r)
 
 @wireprotocommand('clonebundles', '')
 def clonebundles(repo, proto):
     """Server command for returning info for available bundles to seed clones.
 
     Clients will parse this response and determine what bundle to fetch.
 
     Extensions may wrap this command to filter or dynamically emit data
     depending on the request. e.g. you could advertise URLs for the closest
     data center given the client's IP address.
     """
     return repo.vfs.tryread('clonebundles.manifest')
 
 wireprotocaps = ['lookup', 'changegroupsubset', 'branchmap', 'pushkey',
                  'known', 'getbundle', 'unbundlehash', 'batch']
 
 def _capabilities(repo, proto):
     """return a list of capabilities for a repo
 
     This function exists to allow extensions to easily wrap capabilities
     computation
 
     - returns a lists: easy to alter
     - change done here will be propagated to both `capabilities` and `hello`
       command without any other action needed.
     """
     # copy to prevent modification of the global list
     caps = list(wireprotocaps)
     if streamclone.allowservergeneration(repo):
         if repo.ui.configbool('server', 'preferuncompressed'):
             caps.append('stream-preferred')
         requiredformats = repo.requirements & repo.supportedformats
         # if our local revlogs are just revlogv1, add 'stream' cap
         if not requiredformats - {'revlogv1'}:
             caps.append('stream')
         # otherwise, add 'streamreqs' detailing our local revlog format
         else:
             caps.append('streamreqs=%s' % ','.join(sorted(requiredformats)))
     if repo.ui.configbool('experimental', 'bundle2-advertise'):
         capsblob = bundle2.encodecaps(bundle2.getrepocaps(repo))
         caps.append('bundle2=' + urlreq.quote(capsblob))
     caps.append('unbundle=%s' % ','.join(bundle2.bundlepriority))
 
     if proto.name == 'http':
         caps.append('httpheader=%d' %
                     repo.ui.configint('server', 'maxhttpheaderlen'))
         if repo.ui.configbool('experimental', 'httppostargs'):
             caps.append('httppostargs')
 
         # FUTURE advertise 0.2rx once support is implemented
         # FUTURE advertise minrx and mintx after consulting config option
         caps.append('httpmediatype=0.1rx,0.1tx,0.2tx')
 
         compengines = supportedcompengines(repo.ui, proto, util.SERVERROLE)
         if compengines:
             comptypes = ','.join(urlreq.quote(e.wireprotosupport().name)
                                  for e in compengines)
             caps.append('compression=%s' % comptypes)
 
     return caps
 
 # If you are writing an extension and consider wrapping this function. Wrap
 # `_capabilities` instead.
 @wireprotocommand('capabilities')
 def capabilities(repo, proto):
     return ' '.join(_capabilities(repo, proto))
 
 @wireprotocommand('changegroup', 'roots')
 def changegroup(repo, proto, roots):
     nodes = decodelist(roots)
     outgoing = discovery.outgoing(repo, missingroots=nodes,
                                   missingheads=repo.heads())
     cg = changegroupmod.makechangegroup(repo, outgoing, '01', 'serve')
     return streamres(reader=cg, v1compressible=True)
 
 @wireprotocommand('changegroupsubset', 'bases heads')
 def changegroupsubset(repo, proto, bases, heads):
     bases = decodelist(bases)
     heads = decodelist(heads)
     outgoing = discovery.outgoing(repo, missingroots=bases,
                                   missingheads=heads)
     cg = changegroupmod.makechangegroup(repo, outgoing, '01', 'serve')
     return streamres(reader=cg, v1compressible=True)
 
 @wireprotocommand('debugwireargs', 'one two *')
 def debugwireargs(repo, proto, one, two, others):
     # only accept optional args from the known set
     opts = options('debugwireargs', ['three', 'four'], others)
     return repo.debugwireargs(one, two, **opts)
 
 @wireprotocommand('getbundle', '*')
 def getbundle(repo, proto, others):
     opts = options('getbundle', gboptsmap.keys(), others)
     for k, v in opts.iteritems():
         keytype = gboptsmap[k]
         if keytype == 'nodes':
             opts[k] = decodelist(v)
         elif keytype == 'csv':
             opts[k] = list(v.split(','))
         elif keytype == 'scsv':
             opts[k] = set(v.split(','))
         elif keytype == 'boolean':
             # Client should serialize False as '0', which is a non-empty string
             # so it evaluates as a True bool.
             if v == '0':
                 opts[k] = False
             else:
                 opts[k] = bool(v)
         elif keytype != 'plain':
             raise KeyError('unknown getbundle option type %s'
                            % keytype)
 
     if not bundle1allowed(repo, 'pull'):
         if not exchange.bundle2requested(opts.get('bundlecaps')):
             if proto.name == 'http':
                 return ooberror(bundle2required)
             raise error.Abort(bundle2requiredmain,
                               hint=bundle2requiredhint)
 
     try:
         if repo.ui.configbool('server', 'disablefullbundle'):
             # Check to see if this is a full clone.
             clheads = set(repo.changelog.heads())
             heads = set(opts.get('heads', set()))
             common = set(opts.get('common', set()))
             common.discard(nullid)
             if not common and clheads == heads:
                 raise error.Abort(
                     _('server has pull-based clones disabled'),
                     hint=_('remove --pull if specified or upgrade Mercurial'))
 
         chunks = exchange.getbundlechunks(repo, 'serve',
                                           **pycompat.strkwargs(opts))
     except error.Abort as exc:
         # cleanly forward Abort error to the client
         if not exchange.bundle2requested(opts.get('bundlecaps')):
             if proto.name == 'http':
                 return ooberror(str(exc) + '\n')
             raise # cannot do better for bundle1 + ssh
         # bundle2 request expect a bundle2 reply
         bundler = bundle2.bundle20(repo.ui)
         manargs = [('message', str(exc))]
         advargs = []
         if exc.hint is not None:
             advargs.append(('hint', exc.hint))
         bundler.addpart(bundle2.bundlepart('error:abort',
                                            manargs, advargs))
         return streamres(gen=bundler.getchunks(), v1compressible=True)
     return streamres(gen=chunks, v1compressible=True)
 
 @wireprotocommand('heads')
 def heads(repo, proto):
     h = repo.heads()
     return encodelist(h) + "\n"
 
 @wireprotocommand('hello')
 def hello(repo, proto):
     '''the hello command returns a set of lines describing various
     interesting things about the server, in an RFC822-like format.
     Currently the only one defined is "capabilities", which
     consists of a line in the form:
 
     capabilities: space separated list of tokens
     '''
     return "capabilities: %s\n" % (capabilities(repo, proto))
 
 @wireprotocommand('listkeys', 'namespace')
 def listkeys(repo, proto, namespace):
     d = repo.listkeys(encoding.tolocal(namespace)).items()
     return pushkeymod.encodekeys(d)
 
 @wireprotocommand('lookup', 'key')
 def lookup(repo, proto, key):
     try:
         k = encoding.tolocal(key)
         c = repo[k]
         r = c.hex()
         success = 1
     except Exception as inst:
         r = str(inst)
         success = 0
     return "%d %s\n" % (success, r)
 
 @wireprotocommand('known', 'nodes *')
 def known(repo, proto, nodes, others):
     return ''.join(b and "1" or "0" for b in repo.known(decodelist(nodes)))
 
 @wireprotocommand('pushkey', 'namespace key old new')
 def pushkey(repo, proto, namespace, key, old, new):
     # compatibility with pre-1.8 clients which were accidentally
     # sending raw binary nodes rather than utf-8-encoded hex
     if len(new) == 20 and util.escapestr(new) != new:
         # looks like it could be a binary node
         try:
             new.decode('utf-8')
             new = encoding.tolocal(new) # but cleanly decodes as UTF-8
         except UnicodeDecodeError:
             pass # binary, leave unmodified
     else:
         new = encoding.tolocal(new) # normal path
 
     if util.safehasattr(proto, 'restore'):
 
         proto.redirect()
 
         try:
             r = repo.pushkey(encoding.tolocal(namespace), encoding.tolocal(key),
                              encoding.tolocal(old), new) or False
         except error.Abort:
             r = False
 
         output = proto.restore()
 
         return '%s\n%s' % (int(r), output)
 
     r = repo.pushkey(encoding.tolocal(namespace), encoding.tolocal(key),
                      encoding.tolocal(old), new)
     return '%s\n' % int(r)
 
 @wireprotocommand('stream_out')
 def stream(repo, proto):
     '''If the server supports streaming clone, it advertises the "stream"
     capability with a value representing the version and flags of the repo
     it is serving. Client checks to see if it understands the format.
     '''
     if not streamclone.allowservergeneration(repo):
         return '1\n'
 
     def getstream(it):
         yield '0\n'
         for chunk in it:
             yield chunk
 
     try:
         # LockError may be raised before the first result is yielded. Don't
         # emit output until we're sure we got the lock successfully.
         it = streamclone.generatev1wireproto(repo)
         return streamres(gen=getstream(it))
     except error.LockError:
         return '2\n'
 
 @wireprotocommand('unbundle', 'heads')
 def unbundle(repo, proto, heads):
     their_heads = decodelist(heads)
 
     try:
         proto.redirect()
 
         exchange.check_heads(repo, their_heads, 'preparing changes')
 
         # write bundle data to temporary file because it can be big
         fd, tempname = tempfile.mkstemp(prefix='hg-unbundle-')
         fp = os.fdopen(fd, pycompat.sysstr('wb+'))
         r = 0
         try:
             proto.getfile(fp)
             fp.seek(0)
             gen = exchange.readbundle(repo.ui, fp, None)
             if (isinstance(gen, changegroupmod.cg1unpacker)
                 and not bundle1allowed(repo, 'push')):
                 if proto.name == 'http':
                     # need to special case http because stderr do not get to
                     # the http client on failed push so we need to abuse some
                     # other error type to make sure the message get to the
                     # user.
                     return ooberror(bundle2required)
                 raise error.Abort(bundle2requiredmain,
                                   hint=bundle2requiredhint)
 
             r = exchange.unbundle(repo, gen, their_heads, 'serve',
                                   proto._client())
             if util.safehasattr(r, 'addpart'):
                 # The return looks streamable, we are in the bundle2 case and
                 # should return a stream.
                 return streamres(gen=r.getchunks())
             return pushres(r)
 
         finally:
             fp.close()
             os.unlink(tempname)
 
     except (error.BundleValueError, error.Abort, error.PushRaced) as exc:
         # handle non-bundle2 case first
         if not getattr(exc, 'duringunbundle2', False):
             try:
                 raise
             except error.Abort:
                 # The old code we moved used util.stderr directly.
                 # We did not change it to minimise code change.
                 # This need to be moved to something proper.
                 # Feel free to do it.
                 util.stderr.write("abort: %s\n" % exc)
                 if exc.hint is not None:
                     util.stderr.write("(%s)\n" % exc.hint)
                 return pushres(0)
             except error.PushRaced:
                 return pusherr(str(exc))
 
         bundler = bundle2.bundle20(repo.ui)
         for out in getattr(exc, '_bundle2salvagedoutput', ()):
             bundler.addpart(out)
         try:
             try:
                 raise
             except error.PushkeyFailed as exc:
                 # check client caps
                 remotecaps = getattr(exc, '_replycaps', None)
                 if (remotecaps is not None
                         and 'pushkey' not in remotecaps.get('error', ())):
                     # no support remote side, fallback to Abort handler.
                     raise
                 part = bundler.newpart('error:pushkey')
                 part.addparam('in-reply-to', exc.partid)
                 if exc.namespace is not None:
                     part.addparam('namespace', exc.namespace, mandatory=False)
                 if exc.key is not None:
                     part.addparam('key', exc.key, mandatory=False)
                 if exc.new is not None:
                     part.addparam('new', exc.new, mandatory=False)
                 if exc.old is not None:
                     part.addparam('old', exc.old, mandatory=False)
                 if exc.ret is not None:
                     part.addparam('ret', exc.ret, mandatory=False)
         except error.BundleValueError as exc:
             errpart = bundler.newpart('error:unsupportedcontent')
             if exc.parttype is not None:
                 errpart.addparam('parttype', exc.parttype)
             if exc.params:
                 errpart.addparam('params', '\0'.join(exc.params))
         except error.Abort as exc:
             manargs = [('message', str(exc))]
             advargs = []
             if exc.hint is not None:
                 advargs.append(('hint', exc.hint))
             bundler.addpart(bundle2.bundlepart('error:abort',
                                                manargs, advargs))
         except error.PushRaced as exc:
             bundler.newpart('error:pushraced', [('message', str(exc))])
         return streamres(gen=bundler.getchunks())