undo-files: no longer pass the `repo` to `cleanup_undo_files`

As foretold in the previous changesets, we no longer need a full repository
object here.

hgext/narrow/narrowbundle2.py

@@ -1,345 +1,345 b''
 # narrowbundle2.py - bundle2 extensions for narrow repository support
 #
 # Copyright 2017 Google, Inc.
 #
 # This software may be used and distributed according to the terms of the
 # GNU General Public License version 2 or any later version.
 
 
 import struct
 
 from mercurial.i18n import _
 from mercurial import (
     bundle2,
     changegroup,
     error,
     exchange,
     localrepo,
     narrowspec,
     repair,
     requirements,
     scmutil,
     transaction,
     util,
     wireprototypes,
 )
 
 _NARROWACL_SECTION = b'narrowacl'
 _CHANGESPECPART = b'narrow:changespec'
 _RESSPECS = b'narrow:responsespec'
 _SPECPART = b'narrow:spec'
 _SPECPART_INCLUDE = b'include'
 _SPECPART_EXCLUDE = b'exclude'
 _KILLNODESIGNAL = b'KILL'
 _DONESIGNAL = b'DONE'
 _ELIDEDCSHEADER = b'>20s20s20sl'  # cset id, p1, p2, len(text)
 _ELIDEDMFHEADER = b'>20s20s20s20sl'  # manifest id, p1, p2, link id, len(text)
 _CSHEADERSIZE = struct.calcsize(_ELIDEDCSHEADER)
 _MFHEADERSIZE = struct.calcsize(_ELIDEDMFHEADER)
 
 # Serve a changegroup for a client with a narrow clone.
 def getbundlechangegrouppart_narrow(
     bundler,
     repo,
     source,
     bundlecaps=None,
     b2caps=None,
     heads=None,
     common=None,
     **kwargs
 ):
     assert repo.ui.configbool(b'experimental', b'narrowservebrokenellipses')
 
     cgversions = b2caps.get(b'changegroup')
     cgversions = [
         v
         for v in cgversions
         if v in changegroup.supportedoutgoingversions(repo)
     ]
     if not cgversions:
         raise ValueError(_(b'no common changegroup version'))
     version = max(cgversions)
 
     include = sorted(filter(bool, kwargs.get('includepats', [])))
     exclude = sorted(filter(bool, kwargs.get('excludepats', [])))
     generateellipsesbundle2(
         bundler,
         repo,
         include,
         exclude,
         version,
         common,
         heads,
         kwargs.get('depth', None),
     )
 
 
 def generateellipsesbundle2(
     bundler,
     repo,
     include,
     exclude,
     version,
     common,
     heads,
     depth,
 ):
     match = narrowspec.match(repo.root, include=include, exclude=exclude)
     if depth is not None:
         depth = int(depth)
         if depth < 1:
             raise error.Abort(_(b'depth must be positive, got %d') % depth)
 
     heads = set(heads or repo.heads())
     common = set(common or [repo.nullid])
 
     visitnodes, relevant_nodes, ellipsisroots = exchange._computeellipsis(
         repo, common, heads, set(), match, depth=depth
     )
 
     repo.ui.debug(b'Found %d relevant revs\n' % len(relevant_nodes))
     if visitnodes:
         packer = changegroup.getbundler(
             version,
             repo,
             matcher=match,
             ellipses=True,
             shallow=depth is not None,
             ellipsisroots=ellipsisroots,
             fullnodes=relevant_nodes,
         )
         cgdata = packer.generate(common, visitnodes, False, b'narrow_widen')
 
         part = bundler.newpart(b'changegroup', data=cgdata)
         part.addparam(b'version', version)
         if scmutil.istreemanifest(repo):
             part.addparam(b'treemanifest', b'1')
 
 
 def generate_ellipses_bundle2_for_widening(
     bundler,
     repo,
     oldmatch,
     newmatch,
     version,
     common,
     known,
 ):
     common = set(common or [repo.nullid])
     # Steps:
     # 1. Send kill for "$known & ::common"
     #
     # 2. Send changegroup for ::common
     #
     # 3. Proceed.
     #
     # In the future, we can send kills for only the specific
     # nodes we know should go away or change shape, and then
     # send a data stream that tells the client something like this:
     #
     # a) apply this changegroup
     # b) apply nodes XXX, YYY, ZZZ that you already have
     # c) goto a
     #
     # until they've built up the full new state.
     knownrevs = {repo.changelog.rev(n) for n in known}
     # TODO: we could send only roots() of this set, and the
     # list of nodes in common, and the client could work out
     # what to strip, instead of us explicitly sending every
     # single node.
     deadrevs = knownrevs
 
     def genkills():
         for r in deadrevs:
             yield _KILLNODESIGNAL
             yield repo.changelog.node(r)
         yield _DONESIGNAL
 
     bundler.newpart(_CHANGESPECPART, data=genkills())
     newvisit, newfull, newellipsis = exchange._computeellipsis(
         repo, set(), common, knownrevs, newmatch
     )
     if newvisit:
         packer = changegroup.getbundler(
             version,
             repo,
             matcher=newmatch,
             ellipses=True,
             shallow=False,
             ellipsisroots=newellipsis,
             fullnodes=newfull,
         )
         cgdata = packer.generate(common, newvisit, False, b'narrow_widen')
 
         part = bundler.newpart(b'changegroup', data=cgdata)
         part.addparam(b'version', version)
         if scmutil.istreemanifest(repo):
             part.addparam(b'treemanifest', b'1')
 
 
 @bundle2.parthandler(_SPECPART, (_SPECPART_INCLUDE, _SPECPART_EXCLUDE))
 def _handlechangespec_2(op, inpart):
     # XXX: This bundle2 handling is buggy and should be removed after hg5.2 is
     # released. New servers will send a mandatory bundle2 part named
     # 'Narrowspec' and will send specs as data instead of params.
     # Refer to issue5952 and 6019
     includepats = set(inpart.params.get(_SPECPART_INCLUDE, b'').splitlines())
     excludepats = set(inpart.params.get(_SPECPART_EXCLUDE, b'').splitlines())
     narrowspec.validatepatterns(includepats)
     narrowspec.validatepatterns(excludepats)
 
     if not requirements.NARROW_REQUIREMENT in op.repo.requirements:
         op.repo.requirements.add(requirements.NARROW_REQUIREMENT)
         scmutil.writereporequirements(op.repo)
     op.repo.setnarrowpats(includepats, excludepats)
     narrowspec.copytoworkingcopy(op.repo)
 
 
 @bundle2.parthandler(_RESSPECS)
 def _handlenarrowspecs(op, inpart):
     data = inpart.read()
     inc, exc = data.split(b'\0')
     includepats = set(inc.splitlines())
     excludepats = set(exc.splitlines())
     narrowspec.validatepatterns(includepats)
     narrowspec.validatepatterns(excludepats)
 
     if requirements.NARROW_REQUIREMENT not in op.repo.requirements:
         op.repo.requirements.add(requirements.NARROW_REQUIREMENT)
         scmutil.writereporequirements(op.repo)
     op.repo.setnarrowpats(includepats, excludepats)
     narrowspec.copytoworkingcopy(op.repo)
 
 
 @bundle2.parthandler(_CHANGESPECPART)
 def _handlechangespec(op, inpart):
     repo = op.repo
     cl = repo.changelog
 
     # changesets which need to be stripped entirely. either they're no longer
     # needed in the new narrow spec, or the server is sending a replacement
     # in the changegroup part.
     clkills = set()
 
     # A changespec part contains all the updates to ellipsis nodes
     # that will happen as a result of widening or narrowing a
     # repo. All the changes that this block encounters are ellipsis
     # nodes or flags to kill an existing ellipsis.
     chunksignal = changegroup.readexactly(inpart, 4)
     while chunksignal != _DONESIGNAL:
         if chunksignal == _KILLNODESIGNAL:
             # a node used to be an ellipsis but isn't anymore
             ck = changegroup.readexactly(inpart, 20)
             if cl.hasnode(ck):
                 clkills.add(ck)
         else:
             raise error.Abort(
                 _(b'unexpected changespec node chunk type: %s') % chunksignal
             )
         chunksignal = changegroup.readexactly(inpart, 4)
 
     if clkills:
         # preserve bookmarks that repair.strip() would otherwise strip
         op._bookmarksbackup = repo._bookmarks
 
         class dummybmstore(dict):
             def applychanges(self, repo, tr, changes):
                 pass
 
         localrepo.localrepository._bookmarks.set(repo, dummybmstore())
         chgrpfile = repair.strip(
             op.ui, repo, list(clkills), backup=True, topic=b'widen'
         )
         if chgrpfile:
             op._widen_uninterr = repo.ui.uninterruptible()
             op._widen_uninterr.__enter__()
             # presence of _widen_bundle attribute activates widen handler later
             op._widen_bundle = chgrpfile
     # Set the new narrowspec if we're widening. The setnewnarrowpats() method
     # will currently always be there when using the core+narrowhg server, but
     # other servers may include a changespec part even when not widening (e.g.
     # because we're deepening a shallow repo).
     if util.safehasattr(repo, 'setnewnarrowpats'):
         op.gettransaction()
         repo.setnewnarrowpats()
 
 
 def handlechangegroup_widen(op, inpart):
     """Changegroup exchange handler which restores temporarily-stripped nodes"""
     # We saved a bundle with stripped node data we must now restore.
     # This approach is based on mercurial/repair.py@6ee26a53c111.
     repo = op.repo
     ui = op.ui
 
     chgrpfile = op._widen_bundle
     del op._widen_bundle
     vfs = repo.vfs
 
     ui.note(_(b"adding branch\n"))
     f = vfs.open(chgrpfile, b"rb")
     try:
         gen = exchange.readbundle(ui, f, chgrpfile, vfs)
         # silence internal shuffling chatter
         maybe_silent = (
             ui.silent() if not ui.verbose else util.nullcontextmanager()
         )
         with maybe_silent:
             if isinstance(gen, bundle2.unbundle20):
                 with repo.transaction(b'strip') as tr:
                     bundle2.processbundle(repo, gen, lambda: tr)
             else:
                 gen.apply(
                     repo, b'strip', b'bundle:' + vfs.join(chgrpfile), True
                 )
     finally:
         f.close()
 
-    transaction.cleanup_undo_files(repo)
+    transaction.cleanup_undo_files(repo.ui.warn, repo.vfs_map)
 
     # Remove partial backup only if there were no exceptions
     op._widen_uninterr.__exit__(None, None, None)
     vfs.unlink(chgrpfile)
 
 
 def setup():
     """Enable narrow repo support in bundle2-related extension points."""
     getbundleargs = wireprototypes.GETBUNDLE_ARGUMENTS
 
     getbundleargs[b'narrow'] = b'boolean'
     getbundleargs[b'depth'] = b'plain'
     getbundleargs[b'oldincludepats'] = b'csv'
     getbundleargs[b'oldexcludepats'] = b'csv'
     getbundleargs[b'known'] = b'csv'
 
     # Extend changegroup serving to handle requests from narrow clients.
     origcgfn = exchange.getbundle2partsmapping[b'changegroup']
 
     def wrappedcgfn(*args, **kwargs):
         repo = args[1]
         if repo.ui.has_section(_NARROWACL_SECTION):
             kwargs = exchange.applynarrowacl(repo, kwargs)
 
         if kwargs.get('narrow', False) and repo.ui.configbool(
             b'experimental', b'narrowservebrokenellipses'
         ):
             getbundlechangegrouppart_narrow(*args, **kwargs)
         else:
             origcgfn(*args, **kwargs)
 
     exchange.getbundle2partsmapping[b'changegroup'] = wrappedcgfn
 
     # Extend changegroup receiver so client can fixup after widen requests.
     origcghandler = bundle2.parthandlermapping[b'changegroup']
 
     def wrappedcghandler(op, inpart):
         origcghandler(op, inpart)
         if util.safehasattr(op, '_widen_bundle'):
             handlechangegroup_widen(op, inpart)
         if util.safehasattr(op, '_bookmarksbackup'):
             localrepo.localrepository._bookmarks.set(
                 op.repo, op._bookmarksbackup
             )
             del op._bookmarksbackup
 
     wrappedcghandler.params = origcghandler.params
     bundle2.parthandlermapping[b'changegroup'] = wrappedcghandler

mercurial/repair.py

@@ -1,547 +1,547 b''
 # repair.py - functions for repository repair for mercurial
 #
 # Copyright 2005, 2006 Chris Mason <[e-mail]>
 # Copyright 2007 Olivia Mackall
 #
 # This software may be used and distributed according to the terms of the
 # GNU General Public License version 2 or any later version.
 
 
 from .i18n import _
 from .node import (
     hex,
     short,
 )
 from . import (
     bundle2,
     changegroup,
     discovery,
     error,
     exchange,
     obsolete,
     obsutil,
     pathutil,
     phases,
     requirements,
     scmutil,
     transaction,
     util,
 )
 from .utils import (
     hashutil,
     urlutil,
 )
 
 
 def backupbundle(
     repo, bases, heads, node, suffix, compress=True, obsolescence=True
 ):
     """create a bundle with the specified revisions as a backup"""
 
     backupdir = b"strip-backup"
     vfs = repo.vfs
     if not vfs.isdir(backupdir):
         vfs.mkdir(backupdir)
 
     # Include a hash of all the nodes in the filename for uniqueness
     allcommits = repo.set(b'%ln::%ln', bases, heads)
     allhashes = sorted(c.hex() for c in allcommits)
     totalhash = hashutil.sha1(b''.join(allhashes)).digest()
     name = b"%s/%s-%s-%s.hg" % (
         backupdir,
         short(node),
         hex(totalhash[:4]),
         suffix,
     )
 
     cgversion = changegroup.localversion(repo)
     comp = None
     if cgversion != b'01':
         bundletype = b"HG20"
         if compress:
             comp = b'BZ'
     elif compress:
         bundletype = b"HG10BZ"
     else:
         bundletype = b"HG10UN"
 
     outgoing = discovery.outgoing(repo, missingroots=bases, ancestorsof=heads)
     contentopts = {
         b'cg.version': cgversion,
         b'obsolescence': obsolescence,
         b'phases': True,
     }
     return bundle2.writenewbundle(
         repo.ui,
         repo,
         b'strip',
         name,
         bundletype,
         outgoing,
         contentopts,
         vfs,
         compression=comp,
     )
 
 
 def _collectfiles(repo, striprev):
     """find out the filelogs affected by the strip"""
     files = set()
 
     for x in range(striprev, len(repo)):
         files.update(repo[x].files())
 
     return sorted(files)
 
 
 def _collectrevlog(revlog, striprev):
     _, brokenset = revlog.getstrippoint(striprev)
     return [revlog.linkrev(r) for r in brokenset]
 
 
 def _collectbrokencsets(repo, files, striprev):
     """return the changesets which will be broken by the truncation"""
     s = set()
 
     for revlog in manifestrevlogs(repo):
         s.update(_collectrevlog(revlog, striprev))
     for fname in files:
         s.update(_collectrevlog(repo.file(fname), striprev))
 
     return s
 
 
 def strip(ui, repo, nodelist, backup=True, topic=b'backup'):
     # This function requires the caller to lock the repo, but it operates
     # within a transaction of its own, and thus requires there to be no current
     # transaction when it is called.
     if repo.currenttransaction() is not None:
         raise error.ProgrammingError(b'cannot strip from inside a transaction')
 
     # Simple way to maintain backwards compatibility for this
     # argument.
     if backup in [b'none', b'strip']:
         backup = False
 
     repo = repo.unfiltered()
     repo.destroying()
     vfs = repo.vfs
     # load bookmark before changelog to avoid side effect from outdated
     # changelog (see repo._refreshchangelog)
     repo._bookmarks
     cl = repo.changelog
 
     # TODO handle undo of merge sets
     if isinstance(nodelist, bytes):
         nodelist = [nodelist]
     striplist = [cl.rev(node) for node in nodelist]
     striprev = min(striplist)
 
     files = _collectfiles(repo, striprev)
     saverevs = _collectbrokencsets(repo, files, striprev)
 
     # Some revisions with rev > striprev may not be descendants of striprev.
     # We have to find these revisions and put them in a bundle, so that
     # we can restore them after the truncations.
     # To create the bundle we use repo.changegroupsubset which requires
     # the list of heads and bases of the set of interesting revisions.
     # (head = revision in the set that has no descendant in the set;
     #  base = revision in the set that has no ancestor in the set)
     tostrip = set(striplist)
     saveheads = set(saverevs)
     for r in cl.revs(start=striprev + 1):
         if any(p in tostrip for p in cl.parentrevs(r)):
             tostrip.add(r)
 
         if r not in tostrip:
             saverevs.add(r)
             saveheads.difference_update(cl.parentrevs(r))
             saveheads.add(r)
     saveheads = [cl.node(r) for r in saveheads]
 
     # compute base nodes
     if saverevs:
         descendants = set(cl.descendants(saverevs))
         saverevs.difference_update(descendants)
     savebases = [cl.node(r) for r in saverevs]
     stripbases = [cl.node(r) for r in tostrip]
 
     stripobsidx = obsmarkers = ()
     if repo.ui.configbool(b'devel', b'strip-obsmarkers'):
         obsmarkers = obsutil.exclusivemarkers(repo, stripbases)
     if obsmarkers:
         stripobsidx = [
             i for i, m in enumerate(repo.obsstore) if m in obsmarkers
         ]
 
     newbmtarget, updatebm = _bookmarkmovements(repo, tostrip)
 
     backupfile = None
     node = nodelist[-1]
     if backup:
         backupfile = _createstripbackup(repo, stripbases, node, topic)
     # create a changegroup for all the branches we need to keep
     tmpbundlefile = None
     if saveheads:
         # do not compress temporary bundle if we remove it from disk later
         #
         # We do not include obsolescence, it might re-introduce prune markers
         # we are trying to strip.  This is harmless since the stripped markers
         # are already backed up and we did not touched the markers for the
         # saved changesets.
         tmpbundlefile = backupbundle(
             repo,
             savebases,
             saveheads,
             node,
             b'temp',
             compress=False,
             obsolescence=False,
         )
 
     with ui.uninterruptible():
         try:
             with repo.transaction(b"strip") as tr:
                 # TODO this code violates the interface abstraction of the
                 # transaction and makes assumptions that file storage is
                 # using append-only files. We'll need some kind of storage
                 # API to handle stripping for us.
                 oldfiles = set(tr._offsetmap.keys())
                 oldfiles.update(tr._newfiles)
 
                 tr.startgroup()
                 cl.strip(striprev, tr)
                 stripmanifest(repo, striprev, tr, files)
 
                 for fn in files:
                     repo.file(fn).strip(striprev, tr)
                 tr.endgroup()
 
                 entries = tr.readjournal()
 
                 for file, troffset in entries:
                     if file in oldfiles:
                         continue
                     with repo.svfs(file, b'a', checkambig=True) as fp:
                         fp.truncate(troffset)
                     if troffset == 0:
                         repo.store.markremoved(file)
 
                 deleteobsmarkers(repo.obsstore, stripobsidx)
                 del repo.obsstore
                 repo.invalidatevolatilesets()
                 repo._phasecache.filterunknown(repo)
 
             if tmpbundlefile:
                 ui.note(_(b"adding branch\n"))
                 f = vfs.open(tmpbundlefile, b"rb")
                 gen = exchange.readbundle(ui, f, tmpbundlefile, vfs)
                 # silence internal shuffling chatter
                 maybe_silent = (
                     repo.ui.silent()
                     if not repo.ui.verbose
                     else util.nullcontextmanager()
                 )
                 with maybe_silent:
                     tmpbundleurl = b'bundle:' + vfs.join(tmpbundlefile)
                     txnname = b'strip'
                     if not isinstance(gen, bundle2.unbundle20):
                         txnname = b"strip\n%s" % urlutil.hidepassword(
                             tmpbundleurl
                         )
                     with repo.transaction(txnname) as tr:
                         bundle2.applybundle(
                             repo, gen, tr, source=b'strip', url=tmpbundleurl
                         )
                 f.close()
 
             with repo.transaction(b'repair') as tr:
                 bmchanges = [(m, repo[newbmtarget].node()) for m in updatebm]
                 repo._bookmarks.applychanges(repo, tr, bmchanges)
 
-            transaction.cleanup_undo_files(repo)
+            transaction.cleanup_undo_files(repo.ui.warn, repo.vfs_map)
 
         except:  # re-raises
             if backupfile:
                 ui.warn(
                     _(b"strip failed, backup bundle stored in '%s'\n")
                     % vfs.join(backupfile)
                 )
             if tmpbundlefile:
                 ui.warn(
                     _(b"strip failed, unrecovered changes stored in '%s'\n")
                     % vfs.join(tmpbundlefile)
                 )
                 ui.warn(
                     _(
                         b"(fix the problem, then recover the changesets with "
                         b"\"hg unbundle '%s'\")\n"
                     )
                     % vfs.join(tmpbundlefile)
                 )
             raise
         else:
             if tmpbundlefile:
                 # Remove temporary bundle only if there were no exceptions
                 vfs.unlink(tmpbundlefile)
 
     repo.destroyed()
     # return the backup file path (or None if 'backup' was False) so
     # extensions can use it
     return backupfile
 
 
 def softstrip(ui, repo, nodelist, backup=True, topic=b'backup'):
     """perform a "soft" strip using the archived phase"""
     tostrip = [c.node() for c in repo.set(b'sort(%ln::)', nodelist)]
     if not tostrip:
         return None
 
     backupfile = None
     if backup:
         node = tostrip[0]
         backupfile = _createstripbackup(repo, tostrip, node, topic)
 
     newbmtarget, updatebm = _bookmarkmovements(repo, tostrip)
     with repo.transaction(b'strip') as tr:
         phases.retractboundary(repo, tr, phases.archived, tostrip)
         bmchanges = [(m, repo[newbmtarget].node()) for m in updatebm]
         repo._bookmarks.applychanges(repo, tr, bmchanges)
     return backupfile
 
 
 def _bookmarkmovements(repo, tostrip):
     # compute necessary bookmark movement
     bm = repo._bookmarks
     updatebm = []
     for m in bm:
         rev = repo[bm[m]].rev()
         if rev in tostrip:
             updatebm.append(m)
     newbmtarget = None
     # If we need to move bookmarks, compute bookmark
     # targets. Otherwise we can skip doing this logic.
     if updatebm:
         # For a set s, max(parents(s) - s) is the same as max(heads(::s - s)),
         # but is much faster
         newbmtarget = repo.revs(b'max(parents(%ld) - (%ld))', tostrip, tostrip)
         if newbmtarget:
             newbmtarget = repo[newbmtarget.first()].node()
         else:
             newbmtarget = b'.'
     return newbmtarget, updatebm
 
 
 def _createstripbackup(repo, stripbases, node, topic):
     # backup the changeset we are about to strip
     vfs = repo.vfs
     cl = repo.changelog
     backupfile = backupbundle(repo, stripbases, cl.heads(), node, topic)
     repo.ui.status(_(b"saved backup bundle to %s\n") % vfs.join(backupfile))
     repo.ui.log(
         b"backupbundle", b"saved backup bundle to %s\n", vfs.join(backupfile)
     )
     return backupfile
 
 
 def safestriproots(ui, repo, nodes):
     """return list of roots of nodes where descendants are covered by nodes"""
     torev = repo.unfiltered().changelog.rev
     revs = {torev(n) for n in nodes}
     # tostrip = wanted - unsafe = wanted - ancestors(orphaned)
     # orphaned = affected - wanted
     # affected = descendants(roots(wanted))
     # wanted = revs
     revset = b'%ld - ( ::( (roots(%ld):: and not _phase(%s)) -%ld) )'
     tostrip = set(repo.revs(revset, revs, revs, phases.internal, revs))
     notstrip = revs - tostrip
     if notstrip:
         nodestr = b', '.join(sorted(short(repo[n].node()) for n in notstrip))
         ui.warn(
             _(b'warning: orphaned descendants detected, not stripping %s\n')
             % nodestr
         )
     return [c.node() for c in repo.set(b'roots(%ld)', tostrip)]
 
 
 class stripcallback:
     """used as a transaction postclose callback"""
 
     def __init__(self, ui, repo, backup, topic):
         self.ui = ui
         self.repo = repo
         self.backup = backup
         self.topic = topic or b'backup'
         self.nodelist = []
 
     def addnodes(self, nodes):
         self.nodelist.extend(nodes)
 
     def __call__(self, tr):
         roots = safestriproots(self.ui, self.repo, self.nodelist)
         if roots:
             strip(self.ui, self.repo, roots, self.backup, self.topic)
 
 
 def delayedstrip(ui, repo, nodelist, topic=None, backup=True):
     """like strip, but works inside transaction and won't strip irreverent revs
 
     nodelist must explicitly contain all descendants. Otherwise a warning will
     be printed that some nodes are not stripped.
 
     Will do a backup if `backup` is True. The last non-None "topic" will be
     used as the backup topic name. The default backup topic name is "backup".
     """
     tr = repo.currenttransaction()
     if not tr:
         nodes = safestriproots(ui, repo, nodelist)
         return strip(ui, repo, nodes, backup=backup, topic=topic)
     # transaction postclose callbacks are called in alphabet order.
     # use '\xff' as prefix so we are likely to be called last.
     callback = tr.getpostclose(b'\xffstrip')
     if callback is None:
         callback = stripcallback(ui, repo, backup=backup, topic=topic)
         tr.addpostclose(b'\xffstrip', callback)
     if topic:
         callback.topic = topic
     callback.addnodes(nodelist)
 
 
 def stripmanifest(repo, striprev, tr, files):
     for revlog in manifestrevlogs(repo):
         revlog.strip(striprev, tr)
 
 
 def manifestrevlogs(repo):
     yield repo.manifestlog.getstorage(b'')
     if scmutil.istreemanifest(repo):
         # This logic is safe if treemanifest isn't enabled, but also
         # pointless, so we skip it if treemanifest isn't enabled.
         for t, unencoded, size in repo.store.datafiles():
             if unencoded.startswith(b'meta/') and unencoded.endswith(
                 b'00manifest.i'
             ):
                 dir = unencoded[5:-12]
                 yield repo.manifestlog.getstorage(dir)
 
 
 def rebuildfncache(ui, repo, only_data=False):
     """Rebuilds the fncache file from repo history.
 
     Missing entries will be added. Extra entries will be removed.
     """
     repo = repo.unfiltered()
 
     if requirements.FNCACHE_REQUIREMENT not in repo.requirements:
         ui.warn(
             _(
                 b'(not rebuilding fncache because repository does not '
                 b'support fncache)\n'
             )
         )
         return
 
     with repo.lock():
         fnc = repo.store.fncache
         fnc.ensureloaded(warn=ui.warn)
 
         oldentries = set(fnc.entries)
         newentries = set()
         seenfiles = set()
 
         if only_data:
             # Trust the listing of .i from the fncache, but not the .d. This is
             # much faster, because we only need to stat every possible .d files,
             # instead of reading the full changelog
             for f in fnc:
                 if f[:5] == b'data/' and f[-2:] == b'.i':
                     seenfiles.add(f[5:-2])
                     newentries.add(f)
                     dataf = f[:-2] + b'.d'
                     if repo.store._exists(dataf):
                         newentries.add(dataf)
         else:
             progress = ui.makeprogress(
                 _(b'rebuilding'), unit=_(b'changesets'), total=len(repo)
             )
             for rev in repo:
                 progress.update(rev)
 
                 ctx = repo[rev]
                 for f in ctx.files():
                     # This is to minimize I/O.
                     if f in seenfiles:
                         continue
                     seenfiles.add(f)
 
                     i = b'data/%s.i' % f
                     d = b'data/%s.d' % f
 
                     if repo.store._exists(i):
                         newentries.add(i)
                     if repo.store._exists(d):
                         newentries.add(d)
 
             progress.complete()
 
         if requirements.TREEMANIFEST_REQUIREMENT in repo.requirements:
             # This logic is safe if treemanifest isn't enabled, but also
             # pointless, so we skip it if treemanifest isn't enabled.
             for dir in pathutil.dirs(seenfiles):
                 i = b'meta/%s/00manifest.i' % dir
                 d = b'meta/%s/00manifest.d' % dir
 
                 if repo.store._exists(i):
                     newentries.add(i)
                 if repo.store._exists(d):
                     newentries.add(d)
 
         addcount = len(newentries - oldentries)
         removecount = len(oldentries - newentries)
         for p in sorted(oldentries - newentries):
             ui.write(_(b'removing %s\n') % p)
         for p in sorted(newentries - oldentries):
             ui.write(_(b'adding %s\n') % p)
 
         if addcount or removecount:
             ui.write(
                 _(b'%d items added, %d removed from fncache\n')
                 % (addcount, removecount)
             )
             fnc.entries = newentries
             fnc._dirty = True
 
             with repo.transaction(b'fncache') as tr:
                 fnc.write(tr)
         else:
             ui.write(_(b'fncache already up to date\n'))
 
 
 def deleteobsmarkers(obsstore, indices):
     """Delete some obsmarkers from obsstore and return how many were deleted
 
     'indices' is a list of ints which are the indices
     of the markers to be deleted.
 
     Every invocation of this function completely rewrites the obsstore file,
     skipping the markers we want to be removed. The new temporary file is
     created, remaining markers are written there and on .close() this file
     gets atomically renamed to obsstore, thus guaranteeing consistency."""
     if not indices:
         # we don't want to rewrite the obsstore with the same content
         return
 
     left = []
     current = obsstore._all
     n = 0
     for i, m in enumerate(current):
         if i in indices:
             n += 1
             continue
         left.append(m)
 
     newobsstorefile = obsstore.svfs(b'obsstore', b'w', atomictemp=True)
     for bytes in obsolete.encodemarkers(left, True, obsstore._version):
         newobsstorefile.write(bytes)
     newobsstorefile.close()
     return n

mercurial/streamclone.py

@@ -1,935 +1,935 b''
 # streamclone.py - producing and consuming streaming repository data
 #
 # Copyright 2015 Gregory Szorc <[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.
 
 
 import contextlib
 import os
 import struct
 
 from .i18n import _
 from .pycompat import open
 from .interfaces import repository
 from . import (
     bookmarks,
     cacheutil,
     error,
     narrowspec,
     phases,
     pycompat,
     requirements as requirementsmod,
     scmutil,
     store,
     transaction,
     util,
 )
 from .revlogutils import (
     nodemap,
 )
 
 
 def new_stream_clone_requirements(default_requirements, streamed_requirements):
     """determine the final set of requirement for a new stream clone
 
     this method combine the "default" requirements that a new repository would
     use with the constaint we get from the stream clone content. We keep local
     configuration choice when possible.
     """
     requirements = set(default_requirements)
     requirements -= requirementsmod.STREAM_FIXED_REQUIREMENTS
     requirements.update(streamed_requirements)
     return requirements
 
 
 def streamed_requirements(repo):
     """the set of requirement the new clone will have to support
 
     This is used for advertising the stream options and to generate the actual
     stream content."""
     requiredformats = (
         repo.requirements & requirementsmod.STREAM_FIXED_REQUIREMENTS
     )
     return requiredformats
 
 
 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.
 
     Returns a tuple of (supported, requirements). ``supported`` is True if
     streaming clone is supported and False otherwise. ``requirements`` is
     a set of repo requirements from the remote, or ``None`` if stream clone
     isn't supported.
     """
     repo = pullop.repo
     remote = pullop.remote
 
     bundle2supported = False
     if pullop.canusebundle2:
         if b'v2' in pullop.remotebundle2caps.get(b'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.
     elif bundle2 and not bundle2supported:
         return False, None
 
     # Streaming clone only works on empty repositories.
     if len(repo):
         return False, None
 
     # Streaming clone only works if all data is being requested.
     if pullop.heads:
         return False, None
 
     streamrequested = pullop.streamclonerequested
 
     # If we don't have a preference, let the server decide for us. This
     # likely only comes into play in LANs.
     if streamrequested is None:
         # The server can advertise whether to prefer streaming clone.
         streamrequested = remote.capable(b'stream-preferred')
 
     if not streamrequested:
         return False, None
 
     # In order for stream clone to work, the client has to support all the
     # requirements advertised by the server.
     #
     # The server advertises its requirements via the "stream" and "streamreqs"
     # capability. "stream" (a value-less capability) is advertised if and only
     # if the only requirement is "revlogv1." Else, the "streamreqs" capability
     # is advertised and contains a comma-delimited list of requirements.
     requirements = set()
     if remote.capable(b'stream'):
         requirements.add(requirementsmod.REVLOGV1_REQUIREMENT)
     else:
         streamreqs = remote.capable(b'streamreqs')
         # This is weird and shouldn't happen with modern servers.
         if not streamreqs:
             pullop.repo.ui.warn(
                 _(
                     b'warning: stream clone requested but server has them '
                     b'disabled\n'
                 )
             )
             return False, None
 
         streamreqs = set(streamreqs.split(b','))
         # Server requires something we don't support. Bail.
         missingreqs = streamreqs - repo.supported
         if missingreqs:
             pullop.repo.ui.warn(
                 _(
                     b'warning: stream clone requested but client is missing '
                     b'requirements: %s\n'
                 )
                 % b', '.join(sorted(missingreqs))
             )
             pullop.repo.ui.warn(
                 _(
                     b'(see https://www.mercurial-scm.org/wiki/MissingRequirement '
                     b'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
     operations.
 
     A legacy stream clone will not be performed if a bundle2 stream clone is
     supported.
     """
     from . import localrepo
 
     supported, requirements = canperformstreamclone(pullop)
 
     if not supported:
         return
 
     repo = pullop.repo
     remote = pullop.remote
 
     # Save remote branchmap. We will use it later to speed up branchcache
     # creation.
     rbranchmap = None
     if remote.capable(b'branchmap'):
         with remote.commandexecutor() as e:
             rbranchmap = e.callcommand(b'branchmap', {}).result()
 
     repo.ui.status(_(b'streaming all changes\n'))
 
     with remote.commandexecutor() as e:
         fp = e.callcommand(b'stream_out', {}).result()
 
     # TODO strictly speaking, this code should all be inside the context
     # manager because the context manager is supposed to ensure all wire state
     # is flushed when exiting. But the legacy peers don't do this, so it
     # doesn't matter.
     l = fp.readline()
     try:
         resp = int(l)
     except ValueError:
         raise error.ResponseError(
             _(b'unexpected response from remote server:'), l
         )
     if resp == 1:
         raise error.Abort(_(b'operation forbidden by server'))
     elif resp == 2:
         raise error.Abort(_(b'locking the remote repository failed'))
     elif resp != 0:
         raise error.Abort(_(b'the server sent an unknown error code'))
 
     l = fp.readline()
     try:
         filecount, bytecount = map(int, l.split(b' ', 1))
     except (ValueError, TypeError):
         raise error.ResponseError(
             _(b'unexpected response from remote server:'), l
         )
 
     with repo.lock():
         consumev1(repo, fp, filecount, bytecount)
         repo.requirements = new_stream_clone_requirements(
             repo.requirements,
             requirements,
         )
         repo.svfs.options = localrepo.resolvestorevfsoptions(
             repo.ui, repo.requirements, repo.features
         )
         scmutil.writereporequirements(repo)
         nodemap.post_stream_cleanup(repo)
 
         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 not repo.ui.configbool(b'server', b'uncompressed', untrusted=True):
         return False
 
     # The way stream clone works makes it impossible to hide secret changesets.
     # So don't allow this by default.
     secret = phases.hassecret(repo)
     if secret:
         return repo.ui.configbool(b'server', b'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).
 
     The data iterator consists of N entries for each file being transferred.
     Each file entry starts as a line with the file name and integer size
     delimited by a null byte.
 
     The raw file data follows. Following the raw file data is the next file
     entry, or EOF.
 
     When used on the wire protocol, an additional line indicating protocol
     success will be prepended to the stream. This function is not responsible
     for adding it.
 
     This function will obtain a repository lock to ensure a consistent view of
     the store is captured. It therefore may raise LockError.
     """
     entries = []
     total_bytes = 0
     # Get consistent snapshot of repo, lock during scan.
     with repo.lock():
         repo.ui.debug(b'scanning\n')
         for file_type, name, size in _walkstreamfiles(repo):
             if size:
                 entries.append((name, size))
                 total_bytes += size
         _test_sync_point_walk_1(repo)
     _test_sync_point_walk_2(repo)
 
     repo.ui.debug(
         b'%d files, %d bytes to transfer\n' % (len(entries), total_bytes)
     )
 
     svfs = repo.svfs
     debugflag = repo.ui.debugflag
 
     def emitrevlogdata():
         for name, size in entries:
             if debugflag:
                 repo.ui.debug(b'sending %s (%d bytes)\n' % (name, size))
             # partially encode name over the wire for backwards compat
             yield b'%s\0%d\n' % (store.encodedir(name), size)
             # auditing at this stage is both pointless (paths are already
             # trusted by the local repo) and expensive
             with svfs(name, b'rb', auditpath=False) as fp:
                 if size <= 65536:
                     yield fp.read(size)
                 else:
                     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
     byte size of payload.
 
     The success line contains "0" for success, "1" for stream generation not
     allowed, and "2" for error locking the repository (possibly indicating
     a permissions error for the server process).
     """
     if not allowservergeneration(repo):
         yield b'1\n'
         return
 
     try:
         filecount, bytecount, it = generatev1(repo)
     except error.LockError:
         yield b'2\n'
         return
 
     # Indicates successful response.
     yield b'0\n'
     yield b'%d %d\n' % (filecount, bytecount)
     for chunk in it:
         yield chunk
 
 
 def generatebundlev1(repo, compression=b'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.
 
     The next 2 bytes indicate the compression type. Only "UN" is currently
     supported.
 
     The next 16 bytes are two 64-bit big endian unsigned integers indicating
     file count and byte count, respectively.
 
     The next 2 bytes is a 16-bit big endian unsigned short declaring the length
     of the requirements string, including a trailing \0. The following N bytes
     are the requirements string, which is ASCII containing a comma-delimited
     list of repo requirements that are needed to support the data.
 
     The remaining content is the output of ``generatev1()`` (which may be
     compressed in the future).
 
     Returns a tuple of (requirements, data generator).
     """
     if compression != b'UN':
         raise ValueError(b'we do not support the compression argument yet')
 
     requirements = streamed_requirements(repo)
     requires = b','.join(sorted(requirements))
 
     def gen():
         yield b'HGS1'
         yield compression
 
         filecount, bytecount, it = generatev1(repo)
         repo.ui.status(
             _(b'writing %d bytes for %d files\n') % (bytecount, filecount)
         )
 
         yield struct.pack(b'>QQ', filecount, bytecount)
         yield struct.pack(b'>H', len(requires) + 1)
         yield requires + b'\0'
 
         # This is where we'll add compression in the future.
         assert compression == b'UN'
 
         progress = repo.ui.makeprogress(
             _(b'bundle'), total=bytecount, unit=_(b'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(
             _(b'%d files to transfer, %s of data\n')
             % (filecount, util.bytecount(bytecount))
         )
         progress = repo.ui.makeprogress(
             _(b'clone'), total=bytecount, unit=_(b'bytes')
         )
         progress.update(0)
         start = util.timer()
 
         # TODO: get rid of (potential) inconsistency
         #
         # If transaction is started and any @filecache property is
         # changed at this point, it causes inconsistency between
         # in-memory cached property and streamclone-ed file on the
         # disk. Nested transaction prevents transaction scope "clone"
         # below from writing in-memory changes out at the end of it,
         # even though in-memory changes are discarded at the end of it
         # regardless of transaction nesting.
         #
         # But transaction nesting can't be simply prohibited, because
         # nesting occurs also in ordinary case (e.g. enabling
         # clonebundles).
 
         with repo.transaction(b'clone'):
             with repo.svfs.backgroundclosing(repo.ui, expectedcount=filecount):
                 for i in range(filecount):
                     # XXX doesn't support '\n' or '\r' in filenames
                     l = fp.readline()
                     try:
                         name, size = l.split(b'\0', 1)
                         size = int(size)
                     except (ValueError, TypeError):
                         raise error.ResponseError(
                             _(b'unexpected response from remote server:'), l
                         )
                     if repo.ui.debugflag:
                         repo.ui.debug(
                             b'adding %s (%s)\n' % (name, util.bytecount(size))
                         )
                     # for backwards compat, name was partially encoded
                     path = store.decodedir(name)
                     with repo.svfs(path, b'w', backgroundclose=True) as ofp:
                         for chunk in util.filechunkiter(fp, limit=size):
                             progress.increment(step=len(chunk))
                             ofp.write(chunk)
 
             # force @filecache properties to be reloaded from
             # streamclone-ed file at next access
             repo.invalidate(clearfilecache=True)
 
         elapsed = util.timer() - start
         if elapsed <= 0:
             elapsed = 0.001
         progress.complete()
         repo.ui.status(
             _(b'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 != b'UN':
         raise error.Abort(
             _(
                 b'only uncompressed stream clone bundles are '
                 b'supported; got %s'
             )
             % compression
         )
 
     filecount, bytecount = struct.unpack(b'>QQ', fp.read(16))
     requireslen = struct.unpack(b'>H', fp.read(2))[0]
     requires = fp.read(requireslen)
 
     if not requires.endswith(b'\0'):
         raise error.Abort(
             _(
                 b'malformed stream clone bundle: '
                 b'requirements not properly encoded'
             )
         )
 
     requirements = set(requires.rstrip(b'\0').split(b','))
 
     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(
             _(b'cannot apply stream clone bundle on non-empty repo')
         )
 
     filecount, bytecount, requirements = readbundle1header(fp)
     missingreqs = requirements - repo.supported
     if missingreqs:
         raise error.Abort(
             _(b'unable to apply stream clone: unsupported format: %s')
             % b', '.join(sorted(missingreqs))
         )
 
     consumev1(repo, fp, filecount, bytecount)
     nodemap.post_stream_cleanup(repo)
 
 
 class streamcloneapplier:
     """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 = b's'  # store (svfs)
 _srccache = b'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(b'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 = []
     dst_dir = pycompat.mkdtemp(prefix=b'hg-clone-')
     try:
 
         def copy(src):
             fd, dst = pycompat.mkstemp(
                 prefix=os.path.basename(src), dir=dst_dir
             )
             os.close(fd)
             files.append(dst)
             util.copyfiles(src, dst, hardlink=True)
             return dst
 
         yield copy
     finally:
         for tmp in files:
             util.tryunlink(tmp)
         util.tryrmdir(dst_dir)
 
 
 def _makemap(repo):
     """make a (src -> vfs) map for the repo"""
     vfsmap = {
         _srcstore: repo.svfs,
         _srccache: repo.cachevfs,
     }
     # we keep repo.vfs out of the on purpose, ther are too many danger there
     # (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)
     # we keep repo.vfs out of the on purpose, ther are too many danger there
     # (eg: .hg/hgrc),
     #
     # this assert is duplicated (from _makemap) as author might think this is
     # fine, while this is really not fine.
     if repo.vfs in vfsmap.values():
         raise error.ProgrammingError(
             b'repo.vfs must not be added to vfsmap for security reasons'
         )
 
     progress = repo.ui.makeprogress(
         _(b'bundle'), total=totalfilesize, unit=_(b'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
         totalbytecount = 0
 
         for src, name, ftype, data in entries:
             vfs = vfsmap[src]
             yield src
             yield util.uvarintencode(len(name))
             if ftype == _fileappend:
                 fp = vfs(name)
                 size = data
             elif ftype == _filefull:
                 fp = open(data, b'rb')
                 size = util.fstat(fp).st_size
             bytecount = 0
             try:
                 yield util.uvarintencode(size)
                 yield name
                 if size <= 65536:
                     chunks = (fp.read(size),)
                 else:
                     chunks = util.filechunkiter(fp, limit=size)
                 for chunk in chunks:
                     bytecount += len(chunk)
                     totalbytecount += len(chunk)
                     progress.update(totalbytecount)
                     yield chunk
                 if bytecount != size:
                     # Would most likely be caused by a race due to `hg strip` or
                     # a revlog split
                     raise error.Abort(
                         _(
                             b'clone could only read %d bytes from %s, but '
                             b'expected %d bytes'
                         )
                         % (bytecount, name, size)
                     )
             finally:
                 fp.close()
 
 
 def _test_sync_point_walk_1(repo):
     """a function for synchronisation during tests"""
 
 
 def _test_sync_point_walk_2(repo):
     """a function for synchronisation during tests"""
 
 
 def _v2_walk(repo, includes, excludes, includeobsmarkers):
     """emit a seris of files information useful to clone a repo
 
     return (entries, totalfilesize)
 
     entries is a list of tuple (vfs-key, file-path, file-type, size)
 
     - `vfs-key`: is a key to the right vfs to write the file (see _makemap)
     - `name`: file path of the file to copy (to be feed to the vfss)
     - `file-type`: do this file need to be copied with the source lock ?
     - `size`: the size of the file (or None)
     """
     assert repo._currentlock(repo._lockref) is not None
     entries = []
     totalfilesize = 0
 
     matcher = None
     if includes or excludes:
         matcher = narrowspec.match(repo.root, includes, excludes)
 
     for rl_type, name, size in _walkstreamfiles(repo, matcher):
         if size:
             ft = _fileappend
             if rl_type & store.FILEFLAGS_VOLATILE:
                 ft = _filefull
             entries.append((_srcstore, name, ft, size))
             totalfilesize += size
     for name in _walkstreamfullstorefiles(repo):
         if repo.svfs.exists(name):
             totalfilesize += repo.svfs.lstat(name).st_size
             entries.append((_srcstore, name, _filefull, None))
     if includeobsmarkers and repo.svfs.exists(b'obsstore'):
         totalfilesize += repo.svfs.lstat(b'obsstore').st_size
         entries.append((_srcstore, b'obsstore', _filefull, None))
     for name in cacheutil.cachetocopy(repo):
         if repo.cachevfs.exists(name):
             totalfilesize += repo.cachevfs.lstat(name).st_size
             entries.append((_srccache, name, _filefull, None))
     return entries, totalfilesize
 
 
 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)
     2) A varint containing the length of the filename
     3) A varint containing the length of file data
     4) N bytes containing the filename (the internal, store-agnostic form)
     5) N bytes containing the file data
 
     Returns a 3-tuple of (file count, file size, data iterator).
     """
 
     with repo.lock():
 
         repo.ui.debug(b'scanning\n')
 
         entries, totalfilesize = _v2_walk(
             repo,
             includes=includes,
             excludes=excludes,
             includeobsmarkers=includeobsmarkers,
         )
 
         chunks = _emit2(repo, entries, totalfilesize)
         first = next(chunks)
         assert first is None
         _test_sync_point_walk_1(repo)
     _test_sync_point_walk_2(repo)
 
     return len(entries), totalfilesize, chunks
 
 
 @contextlib.contextmanager
 def nested(*ctxs):
     this = ctxs[0]
     rest = ctxs[1:]
     with this:
         if rest:
             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(
             _(b'%d files to transfer, %s of data\n')
             % (filecount, util.bytecount(filesize))
         )
 
         start = util.timer()
         progress = repo.ui.makeprogress(
             _(b'clone'), total=filesize, unit=_(b'bytes')
         )
         progress.update(0)
 
         vfsmap = _makemap(repo)
         # we keep repo.vfs out of the on purpose, ther are too many danger
         # there (eg: .hg/hgrc),
         #
         # this assert is duplicated (from _makemap) as author might think this
         # is fine, while this is really not fine.
         if repo.vfs in vfsmap.values():
             raise error.ProgrammingError(
                 b'repo.vfs must not be added to vfsmap for security reasons'
             )
 
         with repo.transaction(b'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(
                             b'adding [%s] %s (%s)\n'
                             % (src, name, util.bytecount(datalen))
                         )
 
                     with vfs(name, b'w') as ofp:
                         for chunk in util.filechunkiter(fp, limit=datalen):
                             progress.increment(step=len(chunk))
                             ofp.write(chunk)
 
             # force @filecache properties to be reloaded from
             # streamclone-ed file at next access
             repo.invalidate(clearfilecache=True)
 
         elapsed = util.timer() - start
         if elapsed <= 0:
             elapsed = 0.001
         repo.ui.status(
             _(b'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(
             _(b'unable to apply stream clone: unsupported format: %s')
             % b', '.join(sorted(missingreqs))
         )
 
     consumev2(repo, fp, filecount, filesize)
 
     repo.requirements = new_stream_clone_requirements(
         repo.requirements,
         requirements,
     )
     repo.svfs.options = localrepo.resolvestorevfsoptions(
         repo.ui, repo.requirements, repo.features
     )
     scmutil.writereporequirements(repo)
     nodemap.post_stream_cleanup(repo)
 
 
 def _copy_files(src_vfs_map, dst_vfs_map, entries, progress):
     hardlink = [True]
 
     def copy_used():
         hardlink[0] = False
         progress.topic = _(b'copying')
 
     for k, path, size in entries:
         src_vfs = src_vfs_map[k]
         dst_vfs = dst_vfs_map[k]
         src_path = src_vfs.join(path)
         dst_path = dst_vfs.join(path)
         # We cannot use dirname and makedirs of dst_vfs here because the store
         # encoding confuses them. See issue 6581 for details.
         dirname = os.path.dirname(dst_path)
         if not os.path.exists(dirname):
             util.makedirs(dirname)
         dst_vfs.register_file(path)
         # XXX we could use the #nb_bytes argument.
         util.copyfile(
             src_path,
             dst_path,
             hardlink=hardlink[0],
             no_hardlink_cb=copy_used,
             check_fs_hardlink=False,
         )
         progress.increment()
     return hardlink[0]
 
 
 def local_copy(src_repo, dest_repo):
     """copy all content from one local repository to another
 
     This is useful for local clone"""
     src_store_requirements = {
         r
         for r in src_repo.requirements
         if r not in requirementsmod.WORKING_DIR_REQUIREMENTS
     }
     dest_store_requirements = {
         r
         for r in dest_repo.requirements
         if r not in requirementsmod.WORKING_DIR_REQUIREMENTS
     }
     assert src_store_requirements == dest_store_requirements
 
     with dest_repo.lock():
         with src_repo.lock():
 
             # bookmark is not integrated to the streaming as it might use the
             # `repo.vfs` and they are too many sentitive data accessible
             # through `repo.vfs` to expose it to streaming clone.
             src_book_vfs = bookmarks.bookmarksvfs(src_repo)
             srcbookmarks = src_book_vfs.join(b'bookmarks')
             bm_count = 0
             if os.path.exists(srcbookmarks):
                 bm_count = 1
 
             entries, totalfilesize = _v2_walk(
                 src_repo,
                 includes=None,
                 excludes=None,
                 includeobsmarkers=True,
             )
             src_vfs_map = _makemap(src_repo)
             dest_vfs_map = _makemap(dest_repo)
             progress = src_repo.ui.makeprogress(
                 topic=_(b'linking'),
                 total=len(entries) + bm_count,
                 unit=_(b'files'),
             )
             # copy  files
             #
             # We could copy the full file while the source repository is locked
             # and the other one without the lock. However, in the linking case,
             # this would also requires checks that nobody is appending any data
             # to the files while we do the clone, so this is not done yet. We
             # could do this blindly when copying files.
             files = ((k, path, size) for k, path, ftype, size in entries)
             hardlink = _copy_files(src_vfs_map, dest_vfs_map, files, progress)
 
             # copy bookmarks over
             if bm_count:
                 dst_book_vfs = bookmarks.bookmarksvfs(dest_repo)
                 dstbookmarks = dst_book_vfs.join(b'bookmarks')
                 util.copyfile(srcbookmarks, dstbookmarks)
         progress.complete()
         if hardlink:
             msg = b'linked %d files\n'
         else:
             msg = b'copied %d files\n'
         src_repo.ui.debug(msg % (len(entries) + bm_count))
 
         with dest_repo.transaction(b"localclone") as tr:
             dest_repo.store.write(tr)
 
         # clean up transaction file as they do not make sense
-        transaction.cleanup_undo_files(dest_repo)
+        transaction.cleanup_undo_files(dest_repo.ui.warn, dest_repo.vfs_map)

mercurial/transaction.py

@@ -1,884 +1,884 b''
 # transaction.py - simple journaling scheme for mercurial
 #
 # This transaction scheme is intended to gracefully handle program
 # errors and interruptions. More serious failures like system crashes
 # can be recovered with an fsck-like tool. As the whole repository is
 # effectively log-structured, this should amount to simply truncating
 # anything that isn't referenced in the changelog.
 #
 # Copyright 2005, 2006 Olivia 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.
 
 import errno
 import os
 
 from .i18n import _
 from . import (
     error,
     pycompat,
     util,
 )
 from .utils import stringutil
 
 version = 2
 
 GEN_GROUP_ALL = b'all'
 GEN_GROUP_PRE_FINALIZE = b'prefinalize'
 GEN_GROUP_POST_FINALIZE = b'postfinalize'
 
 
 def active(func):
     def _active(self, *args, **kwds):
         if self._count == 0:
             raise error.ProgrammingError(
                 b'cannot use transaction when it is already committed/aborted'
             )
         return func(self, *args, **kwds)
 
     return _active
 
 
 UNDO_BACKUP = b'undo.backupfiles'
 
 UNDO_FILES_MAY_NEED_CLEANUP = [
     (b'plain', b'undo.desc'),
     # Always delete undo last to make sure we detect that a clean up is needed if
     # the process is interrupted.
     (b'store', b'undo'),
 ]
 
 
-def cleanup_undo_files(repo):
+def cleanup_undo_files(report, vfsmap):
     """remove "undo" files used by the rollback logic
 
     This is useful to prevent rollback running in situation were it does not
     make sense. For example after a strip.
     """
     backup_entries = []
     undo_files = []
-    vfsmap = repo.vfs_map
+    svfs = vfsmap[b'store']
     try:
-        with repo.svfs(UNDO_BACKUP) as f:
-            backup_entries = read_backup_files(repo.ui.warn, f)
+        with svfs(UNDO_BACKUP) as f:
+            backup_entries = read_backup_files(report, f)
     except OSError as e:
         if e.errno != errno.ENOENT:
             msg = _(b'could not read %s: %s\n')
-            msg %= (repo.svfs.join(UNDO_BACKUP), stringutil.forcebytestr(e))
-            repo.ui.warn(msg)
+            msg %= (svfs.join(UNDO_BACKUP), stringutil.forcebytestr(e))
+            report(msg)
 
     for location, f, backup_path, c in backup_entries:
         if location in vfsmap and backup_path:
             undo_files.append((vfsmap[location], backup_path))
 
-    undo_files.append((repo.svfs, UNDO_BACKUP))
+    undo_files.append((svfs, UNDO_BACKUP))
     for location, undo_path in UNDO_FILES_MAY_NEED_CLEANUP:
         undo_files.append((vfsmap[location], undo_path))
     for undovfs, undofile in undo_files:
         try:
             undovfs.unlink(undofile)
         except OSError as e:
             if e.errno != errno.ENOENT:
                 msg = _(b'error removing %s: %s\n')
                 msg %= (undovfs.join(undofile), stringutil.forcebytestr(e))
-                repo.ui.warn(msg)
+                report(msg)
 
 
 def _playback(
     journal,
     report,
     opener,
     vfsmap,
     entries,
     backupentries,
     unlink=True,
     checkambigfiles=None,
 ):
     for f, o in sorted(dict(entries).items()):
         if o or not unlink:
             checkambig = checkambigfiles and (f, b'') in checkambigfiles
             try:
                 fp = opener(f, b'a', checkambig=checkambig)
                 if fp.tell() < o:
                     raise error.Abort(
                         _(
                             b"attempted to truncate %s to %d bytes, but it was "
                             b"already %d bytes\n"
                         )
                         % (f, o, fp.tell())
                     )
                 fp.truncate(o)
                 fp.close()
             except IOError:
                 report(_(b"failed to truncate %s\n") % f)
                 raise
         else:
             try:
                 opener.unlink(f)
             except FileNotFoundError:
                 pass
 
     backupfiles = []
     for l, f, b, c in backupentries:
         if l not in vfsmap and c:
             report(b"couldn't handle %s: unknown cache location %s\n" % (b, l))
         vfs = vfsmap[l]
         try:
             if f and b:
                 filepath = vfs.join(f)
                 backuppath = vfs.join(b)
                 checkambig = checkambigfiles and (f, l) in checkambigfiles
                 try:
                     util.copyfile(backuppath, filepath, checkambig=checkambig)
                     backupfiles.append(b)
                 except IOError as exc:
                     e_msg = stringutil.forcebytestr(exc)
                     report(_(b"failed to recover %s (%s)\n") % (f, e_msg))
             else:
                 target = f or b
                 try:
                     vfs.unlink(target)
                 except FileNotFoundError:
                     pass
         except (IOError, OSError, error.Abort):
             if not c:
                 raise
 
     backuppath = b"%s.backupfiles" % journal
     if opener.exists(backuppath):
         opener.unlink(backuppath)
     opener.unlink(journal)
     try:
         for f in backupfiles:
             if opener.exists(f):
                 opener.unlink(f)
     except (IOError, OSError, error.Abort):
         # only pure backup file remains, it is sage to ignore any error
         pass
 
 
 class transaction(util.transactional):
     def __init__(
         self,
         report,
         opener,
         vfsmap,
         journalname,
         undoname=None,
         after=None,
         createmode=None,
         validator=None,
         releasefn=None,
         checkambigfiles=None,
         name='<unnamed>',
     ):
         """Begin a new transaction
 
         Begins a new transaction that allows rolling back writes in the event of
         an exception.
 
         * `after`: called after the transaction has been committed
         * `createmode`: the mode of the journal file that will be created
         * `releasefn`: called after releasing (with transaction and result)
 
         `checkambigfiles` is a set of (path, vfs-location) tuples,
         which determine whether file stat ambiguity should be avoided
         for corresponded files.
         """
         self._count = 1
         self._usages = 1
         self._report = report
         # a vfs to the store content
         self._opener = opener
         # a map to access file in various {location -> vfs}
         vfsmap = vfsmap.copy()
         vfsmap[b''] = opener  # set default value
         self._vfsmap = vfsmap
         self._after = after
         self._offsetmap = {}
         self._newfiles = set()
         self._journal = journalname
         self._journal_files = []
         self._undoname = undoname
         self._queue = []
         # A callback to do something just after releasing transaction.
         if releasefn is None:
             releasefn = lambda tr, success: None
         self._releasefn = releasefn
 
         self._checkambigfiles = set()
         if checkambigfiles:
             self._checkambigfiles.update(checkambigfiles)
 
         self._names = [name]
 
         # A dict dedicated to precisely tracking the changes introduced in the
         # transaction.
         self.changes = {}
 
         # a dict of arguments to be passed to hooks
         self.hookargs = {}
         self._file = opener.open(self._journal, b"w+")
 
         # a list of ('location', 'path', 'backuppath', cache) entries.
         # - if 'backuppath' is empty, no file existed at backup time
         # - if 'path' is empty, this is a temporary transaction file
         # - if 'location' is not empty, the path is outside main opener reach.
         #   use 'location' value as a key in a vfsmap to find the right 'vfs'
         # (cache is currently unused)
         self._backupentries = []
         self._backupmap = {}
         self._backupjournal = b"%s.backupfiles" % self._journal
         self._backupsfile = opener.open(self._backupjournal, b'w')
         self._backupsfile.write(b'%d\n' % version)
 
         if createmode is not None:
             opener.chmod(self._journal, createmode & 0o666)
             opener.chmod(self._backupjournal, createmode & 0o666)
 
         # hold file generations to be performed on commit
         self._filegenerators = {}
         # hold callback to write pending data for hooks
         self._pendingcallback = {}
         # True is any pending data have been written ever
         self._anypending = False
         # holds callback to call when writing the transaction
         self._finalizecallback = {}
         # holds callback to call when validating the transaction
         # should raise exception if anything is wrong
         self._validatecallback = {}
         if validator is not None:
             self._validatecallback[b'001-userhooks'] = validator
         # hold callback for post transaction close
         self._postclosecallback = {}
         # holds callbacks to call during abort
         self._abortcallback = {}
 
     def __repr__(self):
         name = '/'.join(self._names)
         return '<transaction name=%s, count=%d, usages=%d>' % (
             name,
             self._count,
             self._usages,
         )
 
     def __del__(self):
         if self._journal:
             self._abort()
 
     @property
     def finalized(self):
         return self._finalizecallback is None
 
     @active
     def startgroup(self):
         """delay registration of file entry
 
         This is used by strip to delay vision of strip offset. The transaction
         sees either none or all of the strip actions to be done."""
         self._queue.append([])
 
     @active
     def endgroup(self):
         """apply delayed registration of file entry.
 
         This is used by strip to delay vision of strip offset. The transaction
         sees either none or all of the strip actions to be done."""
         q = self._queue.pop()
         for f, o in q:
             self._addentry(f, o)
 
     @active
     def add(self, file, offset):
         """record the state of an append-only file before update"""
         if (
             file in self._newfiles
             or file in self._offsetmap
             or file in self._backupmap
         ):
             return
         if self._queue:
             self._queue[-1].append((file, offset))
             return
 
         self._addentry(file, offset)
 
     def _addentry(self, file, offset):
         """add a append-only entry to memory and on-disk state"""
         if (
             file in self._newfiles
             or file in self._offsetmap
             or file in self._backupmap
         ):
             return
         if offset:
             self._offsetmap[file] = offset
         else:
             self._newfiles.add(file)
         # add enough data to the journal to do the truncate
         self._file.write(b"%s\0%d\n" % (file, offset))
         self._file.flush()
 
     @active
     def addbackup(self, file, hardlink=True, location=b''):
         """Adds a backup of the file to the transaction
 
         Calling addbackup() creates a hardlink backup of the specified file
         that is used to recover the file in the event of the transaction
         aborting.
 
         * `file`: the file path, relative to .hg/store
         * `hardlink`: use a hardlink to quickly create the backup
         """
         if self._queue:
             msg = b'cannot use transaction.addbackup inside "group"'
             raise error.ProgrammingError(msg)
 
         if (
             file in self._newfiles
             or file in self._offsetmap
             or file in self._backupmap
         ):
             return
         vfs = self._vfsmap[location]
         dirname, filename = vfs.split(file)
         backupfilename = b"%s.backup.%s" % (self._journal, filename)
         backupfile = vfs.reljoin(dirname, backupfilename)
         if vfs.exists(file):
             filepath = vfs.join(file)
             backuppath = vfs.join(backupfile)
             util.copyfile(filepath, backuppath, hardlink=hardlink)
         else:
             backupfile = b''
 
         self._addbackupentry((location, file, backupfile, False))
 
     def _addbackupentry(self, entry):
         """register a new backup entry and write it to disk"""
         self._backupentries.append(entry)
         self._backupmap[entry[1]] = len(self._backupentries) - 1
         self._backupsfile.write(b"%s\0%s\0%s\0%d\n" % entry)
         self._backupsfile.flush()
 
     @active
     def registertmp(self, tmpfile, location=b''):
         """register a temporary transaction file
 
         Such files will be deleted when the transaction exits (on both
         failure and success).
         """
         self._addbackupentry((location, b'', tmpfile, False))
 
     @active
     def addfilegenerator(
         self,
         genid,
         filenames,
         genfunc,
         order=0,
         location=b'',
         post_finalize=False,
     ):
         """add a function to generates some files at transaction commit
 
         The `genfunc` argument is a function capable of generating proper
         content of each entry in the `filename` tuple.
 
         At transaction close time, `genfunc` will be called with one file
         object argument per entries in `filenames`.
 
         The transaction itself is responsible for the backup, creation and
         final write of such file.
 
         The `genid` argument is used to ensure the same set of file is only
         generated once. Call to `addfilegenerator` for a `genid` already
         present will overwrite the old entry.
 
         The `order` argument may be used to control the order in which multiple
         generator will be executed.
 
         The `location` arguments may be used to indicate the files are located
         outside of the the standard directory for transaction. It should match
         one of the key of the `transaction.vfsmap` dictionary.
 
         The `post_finalize` argument can be set to `True` for file generation
         that must be run after the transaction has been finalized.
         """
         # For now, we are unable to do proper backup and restore of custom vfs
         # but for bookmarks that are handled outside this mechanism.
         entry = (order, filenames, genfunc, location, post_finalize)
         self._filegenerators[genid] = entry
 
     @active
     def removefilegenerator(self, genid):
         """reverse of addfilegenerator, remove a file generator function"""
         if genid in self._filegenerators:
             del self._filegenerators[genid]
 
     def _generatefiles(self, suffix=b'', group=GEN_GROUP_ALL):
         # write files registered for generation
         any = False
 
         if group == GEN_GROUP_ALL:
             skip_post = skip_pre = False
         else:
             skip_pre = group == GEN_GROUP_POST_FINALIZE
             skip_post = group == GEN_GROUP_PRE_FINALIZE
 
         for id, entry in sorted(self._filegenerators.items()):
             any = True
             order, filenames, genfunc, location, post_finalize = entry
 
             # for generation at closing, check if it's before or after finalize
             if skip_post and post_finalize:
                 continue
             elif skip_pre and not post_finalize:
                 continue
 
             vfs = self._vfsmap[location]
             files = []
             try:
                 for name in filenames:
                     name += suffix
                     if suffix:
                         self.registertmp(name, location=location)
                         checkambig = False
                     else:
                         self.addbackup(name, location=location)
                         checkambig = (name, location) in self._checkambigfiles
                     files.append(
                         vfs(name, b'w', atomictemp=True, checkambig=checkambig)
                     )
                 genfunc(*files)
                 for f in files:
                     f.close()
                 # skip discard() loop since we're sure no open file remains
                 del files[:]
             finally:
                 for f in files:
                     f.discard()
         return any
 
     @active
     def findoffset(self, file):
         if file in self._newfiles:
             return 0
         return self._offsetmap.get(file)
 
     @active
     def readjournal(self):
         self._file.seek(0)
         entries = []
         for l in self._file.readlines():
             file, troffset = l.split(b'\0')
             entries.append((file, int(troffset)))
         return entries
 
     @active
     def replace(self, file, offset):
         """
         replace can only replace already committed entries
         that are not pending in the queue
         """
         if file in self._newfiles:
             if not offset:
                 return
             self._newfiles.remove(file)
             self._offsetmap[file] = offset
         elif file in self._offsetmap:
             if not offset:
                 del self._offsetmap[file]
                 self._newfiles.add(file)
             else:
                 self._offsetmap[file] = offset
         else:
             raise KeyError(file)
         self._file.write(b"%s\0%d\n" % (file, offset))
         self._file.flush()
 
     @active
     def nest(self, name='<unnamed>'):
         self._count += 1
         self._usages += 1
         self._names.append(name)
         return self
 
     def release(self):
         if self._count > 0:
             self._usages -= 1
         if self._names:
             self._names.pop()
         # if the transaction scopes are left without being closed, fail
         if self._count > 0 and self._usages == 0:
             self._abort()
 
     def running(self):
         return self._count > 0
 
     def addpending(self, category, callback):
         """add a callback to be called when the transaction is pending
 
         The transaction will be given as callback's first argument.
 
         Category is a unique identifier to allow overwriting an old callback
         with a newer callback.
         """
         self._pendingcallback[category] = callback
 
     @active
     def writepending(self):
         """write pending file to temporary version
 
         This is used to allow hooks to view a transaction before commit"""
         categories = sorted(self._pendingcallback)
         for cat in categories:
             # remove callback since the data will have been flushed
             any = self._pendingcallback.pop(cat)(self)
             self._anypending = self._anypending or any
         self._anypending |= self._generatefiles(suffix=b'.pending')
         return self._anypending
 
     @active
     def hasfinalize(self, category):
         """check is a callback already exist for a category"""
         return category in self._finalizecallback
 
     @active
     def addfinalize(self, category, callback):
         """add a callback to be called when the transaction is closed
 
         The transaction will be given as callback's first argument.
 
         Category is a unique identifier to allow overwriting old callbacks with
         newer callbacks.
         """
         self._finalizecallback[category] = callback
 
     @active
     def addpostclose(self, category, callback):
         """add or replace a callback to be called after the transaction closed
 
         The transaction will be given as callback's first argument.
 
         Category is a unique identifier to allow overwriting an old callback
         with a newer callback.
         """
         self._postclosecallback[category] = callback
 
     @active
     def getpostclose(self, category):
         """return a postclose callback added before, or None"""
         return self._postclosecallback.get(category, None)
 
     @active
     def addabort(self, category, callback):
         """add a callback to be called when the transaction is aborted.
 
         The transaction will be given as the first argument to the callback.
 
         Category is a unique identifier to allow overwriting an old callback
         with a newer callback.
         """
         self._abortcallback[category] = callback
 
     @active
     def addvalidator(self, category, callback):
         """adds a callback to be called when validating the transaction.
 
         The transaction will be given as the first argument to the callback.
 
         callback should raise exception if to abort transaction"""
         self._validatecallback[category] = callback
 
     @active
     def close(self):
         '''commit the transaction'''
         if self._count == 1:
             for category in sorted(self._validatecallback):
                 self._validatecallback[category](self)
             self._validatecallback = None  # Help prevent cycles.
             self._generatefiles(group=GEN_GROUP_PRE_FINALIZE)
             while self._finalizecallback:
                 callbacks = self._finalizecallback
                 self._finalizecallback = {}
                 categories = sorted(callbacks)
                 for cat in categories:
                     callbacks[cat](self)
             # Prevent double usage and help clear cycles.
             self._finalizecallback = None
             self._generatefiles(group=GEN_GROUP_POST_FINALIZE)
 
         self._count -= 1
         if self._count != 0:
             return
         self._file.close()
         self._backupsfile.close()
         # cleanup temporary files
         for l, f, b, c in self._backupentries:
             if l not in self._vfsmap and c:
                 self._report(
                     b"couldn't remove %s: unknown cache location %s\n" % (b, l)
                 )
                 continue
             vfs = self._vfsmap[l]
             if not f and b and vfs.exists(b):
                 try:
                     vfs.unlink(b)
                 except (IOError, OSError, error.Abort) as inst:
                     if not c:
                         raise
                     # Abort may be raise by read only opener
                     self._report(
                         b"couldn't remove %s: %s\n" % (vfs.join(b), inst)
                     )
         self._offsetmap = {}
         self._newfiles = set()
         self._writeundo()
         if self._after:
             self._after()
             self._after = None  # Help prevent cycles.
         if self._opener.isfile(self._backupjournal):
             self._opener.unlink(self._backupjournal)
         if self._opener.isfile(self._journal):
             self._opener.unlink(self._journal)
         for l, _f, b, c in self._backupentries:
             if l not in self._vfsmap and c:
                 self._report(
                     b"couldn't remove %s: unknown cache location"
                     b"%s\n" % (b, l)
                 )
                 continue
             vfs = self._vfsmap[l]
             if b and vfs.exists(b):
                 try:
                     vfs.unlink(b)
                 except (IOError, OSError, error.Abort) as inst:
                     if not c:
                         raise
                     # Abort may be raise by read only opener
                     self._report(
                         b"couldn't remove %s: %s\n" % (vfs.join(b), inst)
                     )
         self._backupentries = []
         self._journal = None
 
         self._releasefn(self, True)  # notify success of closing transaction
         self._releasefn = None  # Help prevent cycles.
 
         # run post close action
         categories = sorted(self._postclosecallback)
         for cat in categories:
             self._postclosecallback[cat](self)
         # Prevent double usage and help clear cycles.
         self._postclosecallback = None
 
     @active
     def abort(self):
         """abort the transaction (generally called on error, or when the
         transaction is not explicitly committed before going out of
         scope)"""
         self._abort()
 
     @active
     def add_journal(self, vfs_id, path):
         self._journal_files.append((vfs_id, path))
 
     def _writeundo(self):
         """write transaction data for possible future undo call"""
         if self._undoname is None:
             return
 
         def undoname(fn: bytes) -> bytes:
             base, name = os.path.split(fn)
             assert name.startswith(self._journal)
             new_name = name.replace(self._journal, self._undoname, 1)
             return os.path.join(base, new_name)
 
         undo_backup_path = b"%s.backupfiles" % self._undoname
         undobackupfile = self._opener.open(undo_backup_path, b'w')
         undobackupfile.write(b'%d\n' % version)
         for l, f, b, c in self._backupentries:
             if not f:  # temporary file
                 continue
             if not b:
                 u = b''
             else:
                 if l not in self._vfsmap and c:
                     self._report(
                         b"couldn't remove %s: unknown cache location"
                         b"%s\n" % (b, l)
                     )
                     continue
                 vfs = self._vfsmap[l]
                 u = undoname(b)
                 util.copyfile(vfs.join(b), vfs.join(u), hardlink=True)
             undobackupfile.write(b"%s\0%s\0%s\0%d\n" % (l, f, u, c))
         undobackupfile.close()
         for vfs, src in self._journal_files:
             dest = undoname(src)
             # if src and dest refer to a same file, vfs.rename is a no-op,
             # leaving both src and dest on disk. delete dest to make sure
             # the rename couldn't be such a no-op.
             vfs.tryunlink(dest)
             try:
                 vfs.rename(src, dest)
             except FileNotFoundError:  # journal file does not yet exist
                 pass
 
     def _abort(self):
         entries = self.readjournal()
         self._count = 0
         self._usages = 0
         self._file.close()
         self._backupsfile.close()
 
         quick = self._can_quick_abort(entries)
         try:
             if not quick:
                 self._report(_(b"transaction abort!\n"))
             for cat in sorted(self._abortcallback):
                 self._abortcallback[cat](self)
             # Prevent double usage and help clear cycles.
             self._abortcallback = None
             if quick:
                 self._do_quick_abort(entries)
             else:
                 self._do_full_abort(entries)
         finally:
             self._journal = None
             self._releasefn(self, False)  # notify failure of transaction
             self._releasefn = None  # Help prevent cycles.
 
     def _can_quick_abort(self, entries):
         """False if any semantic content have been written on disk
 
         True if nothing, except temporary files has been writen on disk."""
         if entries:
             return False
         for e in self._backupentries:
             if e[1]:
                 return False
         return True
 
     def _do_quick_abort(self, entries):
         """(Silently) do a quick cleanup (see _can_quick_abort)"""
         assert self._can_quick_abort(entries)
         tmp_files = [e for e in self._backupentries if not e[1]]
         for vfs_id, old_path, tmp_path, xxx in tmp_files:
             vfs = self._vfsmap[vfs_id]
             try:
                 vfs.unlink(tmp_path)
             except FileNotFoundError:
                 pass
         if self._backupjournal:
             self._opener.unlink(self._backupjournal)
         if self._journal:
             self._opener.unlink(self._journal)
 
     def _do_full_abort(self, entries):
         """(Noisily) rollback all the change introduced by the transaction"""
         try:
             _playback(
                 self._journal,
                 self._report,
                 self._opener,
                 self._vfsmap,
                 entries,
                 self._backupentries,
                 False,
                 checkambigfiles=self._checkambigfiles,
             )
             self._report(_(b"rollback completed\n"))
         except BaseException as exc:
             self._report(_(b"rollback failed - please run hg recover\n"))
             self._report(
                 _(b"(failure reason: %s)\n") % stringutil.forcebytestr(exc)
             )
 
 
 BAD_VERSION_MSG = _(
     b"journal was created by a different version of Mercurial\n"
 )
 
 
 def read_backup_files(report, fp):
     """parse an (already open) backup file an return contained backup entries
 
     entries are in the form: (location, file, backupfile, xxx)
 
     :location:   the vfs identifier (vfsmap's key)
     :file:       original file path (in the vfs)
     :backupfile: path of the backup (in the vfs)
     :cache:      a boolean currently always set to False
     """
     lines = fp.readlines()
     backupentries = []
     if lines:
         ver = lines[0][:-1]
         if ver != (b'%d' % version):
             report(BAD_VERSION_MSG)
         else:
             for line in lines[1:]:
                 if line:
                     # Shave off the trailing newline
                     line = line[:-1]
                     l, f, b, c = line.split(b'\0')
                     backupentries.append((l, f, b, bool(c)))
     return backupentries
 
 
 def rollback(
     opener,
     vfsmap,
     file,
     report,
     checkambigfiles=None,
     skip_journal_pattern=None,
 ):
     """Rolls back the transaction contained in the given file
 
     Reads the entries in the specified file, and the corresponding
     '*.backupfiles' file, to recover from an incomplete transaction.
 
     * `file`: a file containing a list of entries, specifying where
     to truncate each file.  The file should contain a list of
     file\0offset pairs, delimited by newlines. The corresponding
     '*.backupfiles' file should contain a list of file\0backupfile
     pairs, delimited by \0.
 
     `checkambigfiles` is a set of (path, vfs-location) tuples,
     which determine whether file stat ambiguity should be avoided at
     restoring corresponded files.
     """
     entries = []
     backupentries = []
 
     with opener.open(file) as fp:
         lines = fp.readlines()
     for l in lines:
         try:
             f, o = l.split(b'\0')
             entries.append((f, int(o)))
         except ValueError:
             report(
                 _(b"couldn't read journal entry %r!\n") % pycompat.bytestr(l)
             )
 
     backupjournal = b"%s.backupfiles" % file
     if opener.exists(backupjournal):
         with opener.open(backupjournal) as fp:
             backupentries = read_backup_files(report, fp)
     if skip_journal_pattern is not None:
         keep = lambda x: not skip_journal_pattern.match(x[1])
         backupentries = [x for x in backupentries if keep(x)]
 
     _playback(
         file,
         report,
         opener,
         vfsmap,
         entries,
         backupentries,
         checkambigfiles=checkambigfiles,
     )