repack: flush new pack files before proceeding to delete old ones

Summary:
Make repacks flush the newly created pack files to disk before proceeding to
delete the old datapacks. This ensures the data is committed to disk before we
delete the old pack files. There had been user complaints of data corruption.

The changes affects repack only to avoid slowing down performance for other
operations with manipulates datapack and historypack.

Reviewed By: DurhamG

Differential Revision: D12920147

fbshipit-source-id: 907b64d7763a6212fb49487cfc3bc403f8e3dce2

hgext/remotefilelog/basepack.py

@@ -537,8 +537,16 @@ class mutablebasepack(versionmixin):
 
         try:
             sha = self.sha.hexdigest()
+
+            # Doing a repack. Sync the newly created pack files to disk
+            # now so that the new pack files are persisted before
+            # deletion of the old datapacks.
+            syncfile = ledger is not None
+
+            if syncfile:
+                util.syncfile(self.packfp)
             self.packfp.close()
-            self.writeindex()
+            self.writeindex(syncfile=syncfile)
 
             if len(self.entries) == 0:
                 # Empty pack
@@ -578,7 +586,7 @@ class mutablebasepack(versionmixin):
         except Exception:
             pass
 
-    def writeindex(self):
+    def writeindex(self, syncfile=False):
         rawindex = ""
 
         largefanout = len(self.entries) > SMALLFANOUTCUTOFF
@@ -622,6 +630,8 @@ class mutablebasepack(versionmixin):
         if self.VERSION == 1:
             self.idxfp.write(rawentrieslength)
         self.idxfp.write(rawindex)
+        if syncfile:
+            util.syncfile(self.idxfp)
         self.idxfp.close()
 
     def createindex(self, nodelocations):

hgext/remotefilelog/repack.py

@@ -481,7 +481,7 @@ def _runrepack(
     with datapack.mutabledatapack(repo.ui, packpath, version=dv) as dpack:
         with historypack.mutablehistorypack(repo.ui, packpath) as hpack:
             try:
-                packer.run(dpack, hpack)
+                packer.run(packpath, dpack, hpack)
             except error.LockHeld:
                 raise RepackAlreadyRunning(
                     _("skipping repack - another repack " "is already running")
@@ -582,7 +582,7 @@ class repacker(object):
             self.keepkeys = keepset(repo, lambda f, n: (f, n))
             self.isold = isold
 
-    def run(self, targetdata, targethistory):
+    def run(self, packpath, targetdata, targethistory):
         ledger = repackledger()
 
         with flock(
@@ -606,6 +606,9 @@ class repacker(object):
             self.repackdata(ledger, targetdata)
             self.repackhistory(ledger, targethistory)
 
+            # Flush renames in the directory
+            util.syncdir(packpath)
+
             # Call cleanup on each non-corrupt source
             for source in ledger.sources:
                 if source not in ledger.corruptsources:

mercurial/posix.py

@@ -1041,3 +1041,98 @@ def bindunixsocket(sock, path):
     if bakwdfd:
         os.fchdir(bakwdfd)
         os.close(bakwdfd)
+
+
+def _safehasattr(thing, attr):
+    # deferred import to avoid circular import
+    from . import util
+
+    return util.safehasattr(thing, attr)
+
+
+def syncfile(fp):
+    """Makes best effort attempt to make sure all contents previously written
+    to the fp is persisted to a permanent storage device."""
+    try:
+        fp.flush()
+        if _safehasattr(fcntl, "F_FULLFSYNC"):
+            # OSX specific. See comments in syncdir for discussion on this topic.
+            fcntl.fcntl(fp.fileno(), fcntl.F_FULLFSYNC)
+        else:
+            os.fsync(fp.fileno())
+    except (OSError, IOError):
+        # do nothing since this is just best effort
+        pass
+
+
+def syncdir(dirpath):
+    """Makes best effort attempt to make sure previously issued renames where
+    target is a file immediately inside the specified dirpath is persisted
+    to a permanent storage device."""
+
+    # Syncing a file is not as simple as it seems.
+    #
+    # The most common sequence is to sync a file correctly in Unix is `open`,
+    # `fflush`, `fsync`, `close`.
+    #
+    # However, what is the right sequence in case a temporary staging file is
+    # involved? This [LWN article][lwn] lists a sequence of necessary actions.
+    #
+    # 1. create a new temp file (on the same file system!)
+    # 2. write data to the temp file
+    # 3. fsync() the temp file
+    # 4. rename the temp file to the appropriate name
+    # 5. fsync() the containing directory
+    #
+    # While the above step didn't mention flush, it is important to realize
+    # that step 2 implies flush. This is also emphasized by the python
+    # documentation for [os][os]: one should first do `f.flush()`, and then do
+    # `os.fsync(f.fileno())`.
+    #
+    # Performance wise, this [blog article][thunk] points out that the
+    # performance may be affected by other write operations. Here are two of
+    # the many reasons, to help provide an intuitive understanding:
+    #
+    # 1. There is no requirement to prioritize persistence of the file
+    # descriptor with an outstanding fsync call;
+    # 2. Some filesystems require a certain order of data persistence (for
+    # example, to match the order writes were issued).
+    #
+    # There are also platform specific complexities.
+    #
+    # * On [OSX][osx], it is helpful to call fcntl with a particular flag
+    #   in addition to calling flush. There is an unresolved
+    #   [issue][pythonissue] related to hiding this detail from Python
+    #   programmers. In Java, implementation of FileChannel.force was changed
+    #   to use fcntl since [JDK-8080589][jdk-rfr].
+    # * On [Windows][msdn], it is not possible to call FlushFileBuffers on a
+    #   Directory Handle. And this [svn mailing list thread][svn] shows that
+    #   MoveFile does not provide durability guarantee. It may be possible to
+    #   get durability by using MOVEFILE_WRITE_THROUGH flag.
+    #
+    # It is important that one does not retry `fsync` on failures, which is a
+    # point that PostgreSQL learned the hard way, now known as [fsyncgate][pg].
+    # The same thread also points out that the sequence of close/re-open/fsync
+    # does not provide the same durability guarantee in the presence of sync
+    # failures.
+    #
+    # [lwn]: https://lwn.net/Articles/457667/
+    # [os]: https://docs.python.org/3/library/os.html
+    # [osx]: https://github.com/untitaker/python-atomicwrites/pull/16/files
+    # [jdk-rfr]: http://mail.openjdk.java.net/pipermail/nio-dev/2015-May/003174.html
+    # [pg]: https://www.postgresql.org/message-id/flat/CAMsr%2BYHh%2B5Oq4xziwwoEfhoTZgr07vdGG%2Bhu%3D1adXx59aTeaoQ%40mail.gmail.com
+    # [thunk]: https://thunk.org/tytso/blog/2009/03/15/dont-fear-the-fsync/
+    # [pythonissue]: https://bugs.python.org/issue11877
+    # [msdn]: https://docs.microsoft.com/en-us/windows/desktop/FileIO/obtaining-a-handle-to-a-directory
+    # [svn]: http://mail-archives.apache.org/mod_mbox/subversion-dev/201506.mbox/%3cCA+t0gk00nz1f+5bpxjNSK5Xnr4rXZx7ywQ_twr5CN6MyZSKw+w@mail.gmail.com%3e
+    try:
+        dirfd = os.open(dirpath, os.O_DIRECTORY)
+        if _safehasattr(fcntl, "F_FULLFSYNC"):
+            # osx specific
+            fcntl.fcntl(dirfd, fcntl.F_FULLFSYNC)
+        else:
+            os.fsync(dirfd)
+        os.close(dirfd)
+    except (OSError, IOError):
+        # do nothing since this is just best effort
+        pass

mercurial/util.py

@@ -158,6 +158,8 @@ sshargs = platform.sshargs
 statfiles = getattr(osutil, "statfiles", platform.statfiles)
 statisexec = platform.statisexec
 statislink = platform.statislink
+syncfile = platform.syncfile
+syncdir = platform.syncdir
 testpid = platform.testpid
 umask = platform.umask
 unlink = platform.unlink

mercurial/windows.py

@@ -452,6 +452,27 @@ def rename(src, dst):
         os.rename(src, dst)
 
 
+def syncfile(fp):
+    """Makes best effort attempt to make sure all contents previously written
+    to the fp is persisted to a permanent storage device."""
+    # See comments in posix implementation of syncdir for discussion on this
+    # topic.
+    try:
+        fp.flush()
+        os.fsync(fp.fileno())
+    except (OSError, IOError):
+        # do nothing since this is just best effort
+        pass
+
+
+def syncdir(dirpath):
+    """Makes best effort attempt to make sure previously issued
+    renames where target is a file immediately inside the specified
+    dirpath is persisted to a permanent storage device."""
+    # See comments in posix implementation for discussion on this topic.
+    # Do nothing.
+
+
 def getfstype(dirpath):
     try:
         return win32.getfstype(dirpath)