transaction: split new files into a separate set

Journal entries with size 0 are common as they represent new revlog
files. Move them from the dictionary into a set as the former is more
dense. This reduces peak RSS by 70MB for the NetBSD test repository with
around 450k files under .hg/store.

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

mercurial/repair.py

@@ -1,544 +1,545 b''
 # repair.py - functions for repository repair for mercurial
 #
 # Copyright 2005, 2006 Chris Mason <[e-mail]>
 # Copyright 2007 Matt 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 __future__ import absolute_import
 
 import errno
 
 from .i18n import _
 from .node import (
     hex,
     short,
 )
 from . import (
     bundle2,
     changegroup,
     discovery,
     error,
     exchange,
     obsolete,
     obsutil,
     pathutil,
     phases,
     pycompat,
     requirements,
     scmutil,
     util,
 )
 from .utils import (
     hashutil,
     stringutil,
 )
 
 
 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 pycompat.xrange(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)
                 if not repo.ui.verbose:
                     # silence internal shuffling chatter
                     repo.ui.pushbuffer()
                 tmpbundleurl = b'bundle:' + vfs.join(tmpbundlefile)
                 txnname = b'strip'
                 if not isinstance(gen, bundle2.unbundle20):
                     txnname = b"strip\n%s" % util.hidepassword(tmpbundleurl)
                 with repo.transaction(txnname) as tr:
                     bundle2.applybundle(
                         repo, gen, tr, source=b'strip', url=tmpbundleurl
                     )
                 if not repo.ui.verbose:
                     repo.ui.popbuffer()
                 f.close()
 
             with repo.transaction(b'repair') as tr:
                 bmchanges = [(m, repo[newbmtarget].node()) for m in updatebm]
                 repo._bookmarks.applychanges(repo, tr, bmchanges)
 
             # remove undo files
             for undovfs, undofile in repo.undofiles():
                 try:
                     undovfs.unlink(undofile)
                 except OSError as e:
                     if e.errno != errno.ENOENT:
                         ui.warn(
                             _(b'error removing %s: %s\n')
                             % (
                                 undovfs.join(undofile),
                                 stringutil.forcebytestr(e),
                             )
                         )
 
         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
 
     newbmtarget, updatebm = _bookmarkmovements(repo, tostrip)
     if backup:
         node = tostrip[0]
         backupfile = _createstripbackup(repo, tostrip, node, topic)
 
     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(object):
     """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 unencoded, encoded, 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):
     """Rebuilds the fncache file from repo history.
 
     Missing entries will be added. Extra entries will be removed.
     """
     repo = repo.unfiltered()
 
     if b'fncache' 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()
 
         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/transaction.py

@@ -1,737 +1,765 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 Matt Mackall <[e-mail]>
 #
 # This software may be used and distributed according to the terms of the
 # GNU General Public License version 2 or any later version.
 
 from __future__ import absolute_import
 
 import errno
 
 from .i18n import _
 from . import (
     error,
     pycompat,
     util,
 )
 from .utils import stringutil
 
 version = 2
 
 # These are the file generators that should only be executed after the
 # finalizers are done, since they rely on the output of the finalizers (like
 # the changelog having been written).
 postfinalizegenerators = {b'bookmarks', b'dirstate'}
 
 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
 
 
 def _playback(
     journal,
     report,
     opener,
     vfsmap,
     entries,
     backupentries,
     unlink=True,
     checkambigfiles=None,
 ):
     for f, o in entries:
         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 (IOError, OSError) as inst:
                 if inst.errno != errno.ENOENT:
                     raise
 
     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:
                     report(_(b"failed to recover %s\n") % f)
             else:
                 target = f or b
                 try:
                     vfs.unlink(target)
                 except (IOError, OSError) as inst:
                     if inst.errno != errno.ENOENT:
                         raise
         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._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()
 
     @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._offsetmap or file in self._backupmap:
+        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._offsetmap or file in self._backupmap:
+        if (
+            file in self._newfiles
+            or file in self._offsetmap
+            or file in self._backupmap
+        ):
             return
-        self._offsetmap[file] = offset
+        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._offsetmap or file in self._backupmap:
+        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''
     ):
         """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.
         """
         # For now, we are unable to do proper backup and restore of custom vfs
         # but for bookmarks that are handled outside this mechanism.
         self._filegenerators[genid] = (order, filenames, genfunc, location)
 
     @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(pycompat.iteritems(self._filegenerators)):
             any = True
             order, filenames, genfunc, location = entry
 
             # for generation at closing, check if it's before or after finalize
             is_post = id in postfinalizegenerators
             if skip_post and is_post:
                 continue
             elif skip_pre and not is_post:
                 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:
             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 not in self._offsetmap:
+        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._offsetmap[file] = offset
         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()
 
     def _writeundo(self):
         """write transaction data for possible future undo call"""
         if self._undoname is None:
             return
         undobackupfile = self._opener.open(
             b"%s.backupfiles" % self._undoname, 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]
                 base, name = vfs.split(b)
                 assert name.startswith(self._journal), name
                 uname = name.replace(self._journal, self._undoname, 1)
                 u = vfs.reljoin(base, uname)
                 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()
 
     def _abort(self):
         entries = self.readjournal()
         self._count = 0
         self._usages = 0
         self._file.close()
         self._backupsfile.close()
 
         try:
-            if not self._offsetmap and not self._backupentries:
+            if not entries and not self._backupentries:
                 if self._backupjournal:
                     self._opener.unlink(self._backupjournal)
                 if self._journal:
                     self._opener.unlink(self._journal)
                 return
 
             self._report(_(b"transaction abort!\n"))
 
             try:
                 for cat in sorted(self._abortcallback):
                     self._abortcallback[cat](self)
                 # Prevent double usage and help clear cycles.
                 self._abortcallback = None
                 _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)
                 )
         finally:
             self._journal = None
             self._releasefn(self, False)  # notify failure of transaction
             self._releasefn = None  # Help prevent cycles.
 
 
 def rollback(opener, vfsmap, file, report, checkambigfiles=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 = []
 
     fp = opener.open(file)
     lines = fp.readlines()
     fp.close()
     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):
         fp = opener.open(backupjournal)
         lines = fp.readlines()
         if lines:
             ver = lines[0][:-1]
             if ver == (b'%d' % version):
                 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)))
             else:
                 report(
                     _(
                         b"journal was created by a different version of "
                         b"Mercurial\n"
                     )
                 )
 
     _playback(
         file,
         report,
         opener,
         vfsmap,
         entries,
         backupentries,
         checkambigfiles=checkambigfiles,
     )