comparison mercurial/streamclone.py @ 43076:2372284d9457

formatting: blacken the codebase

This is using my patch to black
(https://github.com/psf/black/pull/826) so we don't un-wrap collection
literals.

Done with:

  hg files 'set:**.py - mercurial/thirdparty/** - "contrib/python-zstandard/**"' | xargs black -S

# skip-blame mass-reformatting only

# no-check-commit reformats foo_bar functions

Differential Revision: https://phab.mercurial-scm.org/D6971

43075:57875cf423c9

import contextlib
import os
import struct

from .i18n import _
from .interfaces import (
    repository,
)
from . import (
    cacheutil,
    error,
    narrowspec,
    phases,
    pycompat,
    store,
    util,
)

def canperformstreamclone(pullop, bundle2=False):
    """Whether it is possible to perform a streaming clone as part of pull.

    ``bundle2`` will cause the function to consider stream clone through
    bundle2 and only through bundle2.

    bundle2supported = False
    if pullop.canusebundle2:
        if 'v2' in pullop.remotebundle2caps.get('stream', []):
            bundle2supported = True
        # else
            # Server doesn't support bundle2 stream clone or doesn't support
            # the versions we support. Fall back and possibly allow legacy.

    # Ensures legacy code path uses available bundle2.
    if bundle2supported and not bundle2:
        return False, None
    # Ensures bundle2 doesn't try to do a stream clone if it isn't supported.

        requirements.add('revlogv1')
    else:
        streamreqs = remote.capable('streamreqs')
        # This is weird and shouldn't happen with modern servers.
        if not streamreqs:
            pullop.repo.ui.warn(_(
                'warning: stream clone requested but server has them '
                'disabled\n'))
            return False, None

        streamreqs = set(streamreqs.split(','))
        # Server requires something we don't support. Bail.
        missingreqs = streamreqs - repo.supportedformats
        if missingreqs:
            pullop.repo.ui.warn(_(
                'warning: stream clone requested but client is missing '
                'requirements: %s\n') % ', '.join(sorted(missingreqs)))
            pullop.repo.ui.warn(
                _('(see https://www.mercurial-scm.org/wiki/MissingRequirement '
                  'for more information)\n'))
            return False, None
        requirements = streamreqs

    return True, requirements

def maybeperformlegacystreamclone(pullop):
    """Possibly perform a legacy stream clone operation.

    Legacy stream clones are performed as part of pull but before all other

    l = fp.readline()
    try:
        resp = int(l)
    except ValueError:
        raise error.ResponseError(
            _('unexpected response from remote server:'), l)
    if resp == 1:
        raise error.Abort(_('operation forbidden by server'))
    elif resp == 2:
        raise error.Abort(_('locking the remote repository failed'))
    elif resp != 0:

    l = fp.readline()
    try:
        filecount, bytecount = map(int, l.split(' ', 1))
    except (ValueError, TypeError):
        raise error.ResponseError(
            _('unexpected response from remote server:'), l)

    with repo.lock():
        consumev1(repo, fp, filecount, bytecount)

        # new requirements = old non-format requirements +
        #                    new format-related remote requirements
        # requirements from the streamed-in repository
        repo.requirements = requirements | (
                repo.requirements - repo.supportedformats)
        repo.svfs.options = localrepo.resolvestorevfsoptions(
            repo.ui, repo.requirements, repo.features)
        repo._writerequirements()

        if rbranchmap:
            repo._branchcaches.replace(repo, rbranchmap)

        repo.invalidate()

def allowservergeneration(repo):
    """Whether streaming clones are allowed from the server."""
    if repository.REPO_FEATURE_STREAM_CLONE not in repo.features:
        return False

    if secret:
        return repo.ui.configbool('server', 'uncompressedallowsecret')

    return True

# This is it's own function so extensions can override it.
def _walkstreamfiles(repo, matcher=None):
    return repo.store.walk(matcher)

def generatev1(repo):
    """Emit content for version 1 of a streaming clone.

    This returns a 3-tuple of (file count, byte size, data iterator).

        for name, ename, size in _walkstreamfiles(repo):
            if size:
                entries.append((name, size))
                total_bytes += size

    repo.ui.debug('%d files, %d bytes to transfer\n' %
                  (len(entries), total_bytes))

    svfs = repo.svfs
    debugflag = repo.ui.debugflag

    def emitrevlogdata():

                    for chunk in util.filechunkiter(fp, limit=size):
                        yield chunk

    return len(entries), total_bytes, emitrevlogdata()

def generatev1wireproto(repo):
    """Emit content for version 1 of streaming clone suitable for the wire.

    This is the data output from ``generatev1()`` with 2 header lines. The
    first line indicates overall success. The 2nd contains the file count and

    yield '0\n'
    yield '%d %d\n' % (filecount, bytecount)
    for chunk in it:
        yield chunk

def generatebundlev1(repo, compression='UN'):
    """Emit content for version 1 of a stream clone bundle.

    The first 4 bytes of the output ("HGS1") denote this as stream clone
    bundle version 1.

    def gen():
        yield 'HGS1'
        yield compression

        filecount, bytecount, it = generatev1(repo)
        repo.ui.status(_('writing %d bytes for %d files\n') %
                         (bytecount, filecount))

        yield struct.pack('>QQ', filecount, bytecount)
        yield struct.pack('>H', len(requires) + 1)
        yield requires + '\0'

        # This is where we'll add compression in the future.
        assert compression == 'UN'

        progress = repo.ui.makeprogress(_('bundle'), total=bytecount,
                                        unit=_('bytes'))
        progress.update(0)

        for chunk in it:
            progress.increment(step=len(chunk))
            yield chunk

        progress.complete()

    return requirements, gen()

def consumev1(repo, fp, filecount, bytecount):
    """Apply the contents from version 1 of a streaming clone file handle.

    This takes the output from "stream_out" and applies it to the specified
    repository.

    Like "stream_out," the status line added by the wire protocol is not
    handled by this function.
    """
    with repo.lock():
        repo.ui.status(_('%d files to transfer, %s of data\n') %
                       (filecount, util.bytecount(bytecount)))
        progress = repo.ui.makeprogress(_('clone'), total=bytecount,
                                        unit=_('bytes'))
        progress.update(0)
        start = util.timer()

        # TODO: get rid of (potential) inconsistency
        #

                    try:
                        name, size = l.split('\0', 1)
                        size = int(size)
                    except (ValueError, TypeError):
                        raise error.ResponseError(
                            _('unexpected response from remote server:'), l)
                    if repo.ui.debugflag:
                        repo.ui.debug('adding %s (%s)\n' %
                                      (name, util.bytecount(size)))
                    # for backwards compat, name was partially encoded
                    path = store.decodedir(name)
                    with repo.svfs(path, 'w', backgroundclose=True) as ofp:
                        for chunk in util.filechunkiter(fp, limit=size):
                            progress.increment(step=len(chunk))

        elapsed = util.timer() - start
        if elapsed <= 0:
            elapsed = 0.001
        progress.complete()
        repo.ui.status(_('transferred %s in %.1f seconds (%s/sec)\n') %
                       (util.bytecount(bytecount), elapsed,
                        util.bytecount(bytecount / elapsed)))

def readbundle1header(fp):
    compression = fp.read(2)
    if compression != 'UN':
        raise error.Abort(_('only uncompressed stream clone bundles are '
            'supported; got %s') % compression)

    filecount, bytecount = struct.unpack('>QQ', fp.read(16))
    requireslen = struct.unpack('>H', fp.read(2))[0]
    requires = fp.read(requireslen)

    if not requires.endswith('\0'):
        raise error.Abort(_('malformed stream clone bundle: '
                            'requirements not properly encoded'))

    requirements = set(requires.rstrip('\0').split(','))

    return filecount, bytecount, requirements

def applybundlev1(repo, fp):
    """Apply the content from a stream clone bundle version 1.

    We assume the 4 byte header has been read and validated and the file handle
    is at the 2 byte compression identifier.
    """
    if len(repo):
        raise error.Abort(_('cannot apply stream clone bundle on non-empty '
                            'repo'))

    filecount, bytecount, requirements = readbundle1header(fp)
    missingreqs = requirements - repo.supportedformats
    if missingreqs:
        raise error.Abort(_('unable to apply stream clone: '
                            'unsupported format: %s') %
                            ', '.join(sorted(missingreqs)))

    consumev1(repo, fp, filecount, bytecount)

class streamcloneapplier(object):
    """Class to manage applying streaming clone bundles.

    We need to wrap ``applybundlev1()`` in a dedicated type to enable bundle
    readers to perform bundle type-specific functionality.
    """
    def __init__(self, fh):
        self._fh = fh

    def apply(self, repo):
        return applybundlev1(repo, self._fh)

# type of file to stream
_fileappend = 0 # append only file
_filefull = 1   # full snapshot file

# Source of the file
_srcstore = 's' # store (svfs)
_srccache = 'c' # cache (cache)

# This is it's own function so extensions can override it.
def _walkstreamfullstorefiles(repo):
    """list snapshot file from the store"""
    fnames = []
    if not repo.publishing():
        fnames.append('phaseroots')
    return fnames

def _filterfull(entry, copy, vfsmap):
    """actually copy the snapshot files"""
    src, name, ftype, data = entry
    if ftype != _filefull:
        return entry
    return (src, name, ftype, copy(vfsmap[src].join(name)))

@contextlib.contextmanager
def maketempcopies():
    """return a function to temporary copy file"""
    files = []
    try:
        def copy(src):
            fd, dst = pycompat.mkstemp()
            os.close(fd)
            files.append(dst)
            util.copyfiles(src, dst, hardlink=True)
            return dst
        yield copy
    finally:
        for tmp in files:
            util.tryunlink(tmp)

def _makemap(repo):
    """make a (src -> vfs) map for the repo"""
    vfsmap = {
        _srcstore: repo.svfs,

    # (eg: .hg/hgrc)
    assert repo.vfs not in vfsmap.values()

    return vfsmap

def _emit2(repo, entries, totalfilesize):
    """actually emit the stream bundle"""
    vfsmap = _makemap(repo)
    progress = repo.ui.makeprogress(_('bundle'), total=totalfilesize,
                                    unit=_('bytes'))
    progress.update(0)
    with maketempcopies() as copy, progress:
        # copy is delayed until we are in the try
        entries = [_filterfull(e, copy, vfsmap) for e in entries]
        yield None # this release the lock on the repository
        seen = 0

        for src, name, ftype, data in entries:
            vfs = vfsmap[src]
            yield src

                    progress.update(seen)
                    yield chunk
            finally:
                fp.close()

def generatev2(repo, includes, excludes, includeobsmarkers):
    """Emit content for version 2 of a streaming clone.

    the data stream consists the following entries:
    1) A char representing the file destination (eg: store or cache)

        first = next(chunks)
        assert first is None

    return len(entries), totalfilesize, chunks

@contextlib.contextmanager
def nested(*ctxs):
    this = ctxs[0]
    rest = ctxs[1:]
    with this:

            with nested(*rest):
                yield
        else:
            yield

def consumev2(repo, fp, filecount, filesize):
    """Apply the contents from a version 2 streaming clone.

    Data is read from an object that only needs to provide a ``read(size)``
    method.
    """
    with repo.lock():
        repo.ui.status(_('%d files to transfer, %s of data\n') %
                       (filecount, util.bytecount(filesize)))

        start = util.timer()
        progress = repo.ui.makeprogress(_('clone'), total=filesize,
                                        unit=_('bytes'))
        progress.update(0)

        vfsmap = _makemap(repo)

        with repo.transaction('clone'):
            ctxs = (vfs.backgroundclosing(repo.ui)
                    for vfs in vfsmap.values())
            with nested(*ctxs):
                for i in range(filecount):
                    src = util.readexactly(fp, 1)
                    vfs = vfsmap[src]
                    namelen = util.uvarintdecodestream(fp)
                    datalen = util.uvarintdecodestream(fp)

                    name = util.readexactly(fp, namelen)

                    if repo.ui.debugflag:
                        repo.ui.debug('adding [%s] %s (%s)\n' %
                                      (src, name, util.bytecount(datalen)))

                    with vfs(name, 'w') as ofp:
                        for chunk in util.filechunkiter(fp, limit=datalen):
                            progress.increment(step=len(chunk))
                            ofp.write(chunk)

            repo.invalidate(clearfilecache=True)

        elapsed = util.timer() - start
        if elapsed <= 0:
            elapsed = 0.001
        repo.ui.status(_('transferred %s in %.1f seconds (%s/sec)\n') %
                       (util.bytecount(progress.pos), elapsed,
                        util.bytecount(progress.pos / elapsed)))
        progress.complete()

def applybundlev2(repo, fp, filecount, filesize, requirements):
    from . import localrepo

    missingreqs = [r for r in requirements if r not in repo.supported]
    if missingreqs:
        raise error.Abort(_('unable to apply stream clone: '
                            'unsupported format: %s') %
                          ', '.join(sorted(missingreqs)))

    consumev2(repo, fp, filecount, filesize)

    # new requirements = old non-format requirements +
    #                    new format-related remote requirements
    # requirements from the streamed-in repository
    repo.requirements = set(requirements) | (
            repo.requirements - repo.supportedformats)
    repo.svfs.options = localrepo.resolvestorevfsoptions(
        repo.ui, repo.requirements, repo.features)
    repo._writerequirements()

43076:2372284d9457

import contextlib
import os
import struct

from .i18n import _
from .interfaces import repository
from . import (
    cacheutil,
    error,
    narrowspec,
    phases,
    pycompat,
    store,
    util,
)


def canperformstreamclone(pullop, bundle2=False):
    """Whether it is possible to perform a streaming clone as part of pull.

    ``bundle2`` will cause the function to consider stream clone through
    bundle2 and only through bundle2.

    bundle2supported = False
    if pullop.canusebundle2:
        if 'v2' in pullop.remotebundle2caps.get('stream', []):
            bundle2supported = True
        # else
        # Server doesn't support bundle2 stream clone or doesn't support
        # the versions we support. Fall back and possibly allow legacy.

    # Ensures legacy code path uses available bundle2.
    if bundle2supported and not bundle2:
        return False, None
    # Ensures bundle2 doesn't try to do a stream clone if it isn't supported.

        requirements.add('revlogv1')
    else:
        streamreqs = remote.capable('streamreqs')
        # This is weird and shouldn't happen with modern servers.
        if not streamreqs:
            pullop.repo.ui.warn(
                _(
                    'warning: stream clone requested but server has them '
                    'disabled\n'
                )
            )
            return False, None

        streamreqs = set(streamreqs.split(','))
        # Server requires something we don't support. Bail.
        missingreqs = streamreqs - repo.supportedformats
        if missingreqs:
            pullop.repo.ui.warn(
                _(
                    'warning: stream clone requested but client is missing '
                    'requirements: %s\n'
                )
                % ', '.join(sorted(missingreqs))
            )
            pullop.repo.ui.warn(
                _(
                    '(see https://www.mercurial-scm.org/wiki/MissingRequirement '
                    'for more information)\n'
                )
            )
            return False, None
        requirements = streamreqs

    return True, requirements


def maybeperformlegacystreamclone(pullop):
    """Possibly perform a legacy stream clone operation.

    Legacy stream clones are performed as part of pull but before all other

    l = fp.readline()
    try:
        resp = int(l)
    except ValueError:
        raise error.ResponseError(
            _('unexpected response from remote server:'), l
        )
    if resp == 1:
        raise error.Abort(_('operation forbidden by server'))
    elif resp == 2:
        raise error.Abort(_('locking the remote repository failed'))
    elif resp != 0:

    l = fp.readline()
    try:
        filecount, bytecount = map(int, l.split(' ', 1))
    except (ValueError, TypeError):
        raise error.ResponseError(
            _('unexpected response from remote server:'), l
        )

    with repo.lock():
        consumev1(repo, fp, filecount, bytecount)

        # new requirements = old non-format requirements +
        #                    new format-related remote requirements
        # requirements from the streamed-in repository
        repo.requirements = requirements | (
            repo.requirements - repo.supportedformats
        )
        repo.svfs.options = localrepo.resolvestorevfsoptions(
            repo.ui, repo.requirements, repo.features
        )
        repo._writerequirements()

        if rbranchmap:
            repo._branchcaches.replace(repo, rbranchmap)

        repo.invalidate()


def allowservergeneration(repo):
    """Whether streaming clones are allowed from the server."""
    if repository.REPO_FEATURE_STREAM_CLONE not in repo.features:
        return False

    if secret:
        return repo.ui.configbool('server', 'uncompressedallowsecret')

    return True


# This is it's own function so extensions can override it.
def _walkstreamfiles(repo, matcher=None):
    return repo.store.walk(matcher)


def generatev1(repo):
    """Emit content for version 1 of a streaming clone.

    This returns a 3-tuple of (file count, byte size, data iterator).

        for name, ename, size in _walkstreamfiles(repo):
            if size:
                entries.append((name, size))
                total_bytes += size

    repo.ui.debug(
        '%d files, %d bytes to transfer\n' % (len(entries), total_bytes)
    )

    svfs = repo.svfs
    debugflag = repo.ui.debugflag

    def emitrevlogdata():

                    for chunk in util.filechunkiter(fp, limit=size):
                        yield chunk

    return len(entries), total_bytes, emitrevlogdata()


def generatev1wireproto(repo):
    """Emit content for version 1 of streaming clone suitable for the wire.

    This is the data output from ``generatev1()`` with 2 header lines. The
    first line indicates overall success. The 2nd contains the file count and

    yield '0\n'
    yield '%d %d\n' % (filecount, bytecount)
    for chunk in it:
        yield chunk


def generatebundlev1(repo, compression='UN'):
    """Emit content for version 1 of a stream clone bundle.

    The first 4 bytes of the output ("HGS1") denote this as stream clone
    bundle version 1.

    def gen():
        yield 'HGS1'
        yield compression

        filecount, bytecount, it = generatev1(repo)
        repo.ui.status(
            _('writing %d bytes for %d files\n') % (bytecount, filecount)
        )

        yield struct.pack('>QQ', filecount, bytecount)
        yield struct.pack('>H', len(requires) + 1)
        yield requires + '\0'

        # This is where we'll add compression in the future.
        assert compression == 'UN'

        progress = repo.ui.makeprogress(
            _('bundle'), total=bytecount, unit=_('bytes')
        )
        progress.update(0)

        for chunk in it:
            progress.increment(step=len(chunk))
            yield chunk

        progress.complete()

    return requirements, gen()


def consumev1(repo, fp, filecount, bytecount):
    """Apply the contents from version 1 of a streaming clone file handle.

    This takes the output from "stream_out" and applies it to the specified
    repository.

    Like "stream_out," the status line added by the wire protocol is not
    handled by this function.
    """
    with repo.lock():
        repo.ui.status(
            _('%d files to transfer, %s of data\n')
            % (filecount, util.bytecount(bytecount))
        )
        progress = repo.ui.makeprogress(
            _('clone'), total=bytecount, unit=_('bytes')
        )
        progress.update(0)
        start = util.timer()

        # TODO: get rid of (potential) inconsistency
        #

                    try:
                        name, size = l.split('\0', 1)
                        size = int(size)
                    except (ValueError, TypeError):
                        raise error.ResponseError(
                            _('unexpected response from remote server:'), l
                        )
                    if repo.ui.debugflag:
                        repo.ui.debug(
                            'adding %s (%s)\n' % (name, util.bytecount(size))
                        )
                    # for backwards compat, name was partially encoded
                    path = store.decodedir(name)
                    with repo.svfs(path, 'w', backgroundclose=True) as ofp:
                        for chunk in util.filechunkiter(fp, limit=size):
                            progress.increment(step=len(chunk))

        elapsed = util.timer() - start
        if elapsed <= 0:
            elapsed = 0.001
        progress.complete()
        repo.ui.status(
            _('transferred %s in %.1f seconds (%s/sec)\n')
            % (
                util.bytecount(bytecount),
                elapsed,
                util.bytecount(bytecount / elapsed),
            )
        )


def readbundle1header(fp):
    compression = fp.read(2)
    if compression != 'UN':
        raise error.Abort(
            _('only uncompressed stream clone bundles are ' 'supported; got %s')
            % compression
        )

    filecount, bytecount = struct.unpack('>QQ', fp.read(16))
    requireslen = struct.unpack('>H', fp.read(2))[0]
    requires = fp.read(requireslen)

    if not requires.endswith('\0'):
        raise error.Abort(
            _(
                'malformed stream clone bundle: '
                'requirements not properly encoded'
            )
        )

    requirements = set(requires.rstrip('\0').split(','))

    return filecount, bytecount, requirements


def applybundlev1(repo, fp):
    """Apply the content from a stream clone bundle version 1.

    We assume the 4 byte header has been read and validated and the file handle
    is at the 2 byte compression identifier.
    """
    if len(repo):
        raise error.Abort(
            _('cannot apply stream clone bundle on non-empty ' 'repo')
        )

    filecount, bytecount, requirements = readbundle1header(fp)
    missingreqs = requirements - repo.supportedformats
    if missingreqs:
        raise error.Abort(
            _('unable to apply stream clone: ' 'unsupported format: %s')
            % ', '.join(sorted(missingreqs))
        )

    consumev1(repo, fp, filecount, bytecount)


class streamcloneapplier(object):
    """Class to manage applying streaming clone bundles.

    We need to wrap ``applybundlev1()`` in a dedicated type to enable bundle
    readers to perform bundle type-specific functionality.
    """

    def __init__(self, fh):
        self._fh = fh

    def apply(self, repo):
        return applybundlev1(repo, self._fh)


# type of file to stream
_fileappend = 0  # append only file
_filefull = 1  # full snapshot file

# Source of the file
_srcstore = 's'  # store (svfs)
_srccache = 'c'  # cache (cache)

# This is it's own function so extensions can override it.
def _walkstreamfullstorefiles(repo):
    """list snapshot file from the store"""
    fnames = []
    if not repo.publishing():
        fnames.append('phaseroots')
    return fnames


def _filterfull(entry, copy, vfsmap):
    """actually copy the snapshot files"""
    src, name, ftype, data = entry
    if ftype != _filefull:
        return entry
    return (src, name, ftype, copy(vfsmap[src].join(name)))


@contextlib.contextmanager
def maketempcopies():
    """return a function to temporary copy file"""
    files = []
    try:

        def copy(src):
            fd, dst = pycompat.mkstemp()
            os.close(fd)
            files.append(dst)
            util.copyfiles(src, dst, hardlink=True)
            return dst

        yield copy
    finally:
        for tmp in files:
            util.tryunlink(tmp)


def _makemap(repo):
    """make a (src -> vfs) map for the repo"""
    vfsmap = {
        _srcstore: repo.svfs,

    # (eg: .hg/hgrc)
    assert repo.vfs not in vfsmap.values()

    return vfsmap


def _emit2(repo, entries, totalfilesize):
    """actually emit the stream bundle"""
    vfsmap = _makemap(repo)
    progress = repo.ui.makeprogress(
        _('bundle'), total=totalfilesize, unit=_('bytes')
    )
    progress.update(0)
    with maketempcopies() as copy, progress:
        # copy is delayed until we are in the try
        entries = [_filterfull(e, copy, vfsmap) for e in entries]
        yield None  # this release the lock on the repository
        seen = 0

        for src, name, ftype, data in entries:
            vfs = vfsmap[src]
            yield src

                    progress.update(seen)
                    yield chunk
            finally:
                fp.close()


def generatev2(repo, includes, excludes, includeobsmarkers):
    """Emit content for version 2 of a streaming clone.

    the data stream consists the following entries:
    1) A char representing the file destination (eg: store or cache)

        first = next(chunks)
        assert first is None

    return len(entries), totalfilesize, chunks


@contextlib.contextmanager
def nested(*ctxs):
    this = ctxs[0]
    rest = ctxs[1:]
    with this:

            with nested(*rest):
                yield
        else:
            yield


def consumev2(repo, fp, filecount, filesize):
    """Apply the contents from a version 2 streaming clone.

    Data is read from an object that only needs to provide a ``read(size)``
    method.
    """
    with repo.lock():
        repo.ui.status(
            _('%d files to transfer, %s of data\n')
            % (filecount, util.bytecount(filesize))
        )

        start = util.timer()
        progress = repo.ui.makeprogress(
            _('clone'), total=filesize, unit=_('bytes')
        )
        progress.update(0)

        vfsmap = _makemap(repo)

        with repo.transaction('clone'):
            ctxs = (vfs.backgroundclosing(repo.ui) for vfs in vfsmap.values())
            with nested(*ctxs):
                for i in range(filecount):
                    src = util.readexactly(fp, 1)
                    vfs = vfsmap[src]
                    namelen = util.uvarintdecodestream(fp)
                    datalen = util.uvarintdecodestream(fp)

                    name = util.readexactly(fp, namelen)

                    if repo.ui.debugflag:
                        repo.ui.debug(
                            'adding [%s] %s (%s)\n'
                            % (src, name, util.bytecount(datalen))
                        )

                    with vfs(name, 'w') as ofp:
                        for chunk in util.filechunkiter(fp, limit=datalen):
                            progress.increment(step=len(chunk))
                            ofp.write(chunk)

            repo.invalidate(clearfilecache=True)

        elapsed = util.timer() - start
        if elapsed <= 0:
            elapsed = 0.001
        repo.ui.status(
            _('transferred %s in %.1f seconds (%s/sec)\n')
            % (
                util.bytecount(progress.pos),
                elapsed,
                util.bytecount(progress.pos / elapsed),
            )
        )
        progress.complete()


def applybundlev2(repo, fp, filecount, filesize, requirements):
    from . import localrepo

    missingreqs = [r for r in requirements if r not in repo.supported]
    if missingreqs:
        raise error.Abort(
            _('unable to apply stream clone: ' 'unsupported format: %s')
            % ', '.join(sorted(missingreqs))
        )

    consumev2(repo, fp, filecount, filesize)

    # new requirements = old non-format requirements +
    #                    new format-related remote requirements
    # requirements from the streamed-in repository
    repo.requirements = set(requirements) | (
        repo.requirements - repo.supportedformats
    )
    repo.svfs.options = localrepo.resolvestorevfsoptions(
        repo.ui, repo.requirements, repo.features
    )
    repo._writerequirements()