diff -r 2fbc109fd58a -r a6b8b1ab9116 mercurial/dirstatemap.py
--- a/mercurial/dirstatemap.py	Thu Mar 02 04:16:47 2023 +0100
+++ b/mercurial/dirstatemap.py	Thu Mar 02 19:02:52 2023 +0100
@@ -10,6 +10,7 @@
     error,
     pathutil,
     policy,
+    testing,
     txnutil,
     util,
 )
@@ -31,6 +32,13 @@
 
 rangemask = 0x7FFFFFFF
 
+WRITE_MODE_AUTO = 0
+WRITE_MODE_FORCE_NEW = 1
+WRITE_MODE_FORCE_APPEND = 2
+
+
+V2_MAX_READ_ATTEMPTS = 5
+
 
 class _dirstatemapcommon:
     """
@@ -54,6 +62,16 @@
         self._parents = None
         self._dirtyparents = False
         self._docket = None
+        write_mode = ui.config(b"devel", b"dirstate.v2.data_update_mode")
+        if write_mode == b"auto":
+            self._write_mode = WRITE_MODE_AUTO
+        elif write_mode == b"force-append":
+            self._write_mode = WRITE_MODE_FORCE_APPEND
+        elif write_mode == b"force-new":
+            self._write_mode = WRITE_MODE_FORCE_NEW
+        else:
+            # unknown value, fallback to default
+            self._write_mode = WRITE_MODE_AUTO
 
         # for consistent view between _pl() and _read() invocations
         self._pendingmode = None
@@ -132,11 +150,31 @@
                 raise error.ProgrammingError(
                     b'dirstate only has a docket in v2 format'
                 )
+            self._set_identity()
             self._docket = docketmod.DirstateDocket.parse(
                 self._readdirstatefile(), self._nodeconstants
             )
         return self._docket
 
+    def _read_v2_data(self):
+        data = None
+        attempts = 0
+        while attempts < V2_MAX_READ_ATTEMPTS:
+            attempts += 1
+            try:
+                # TODO: use mmap when possible
+                data = self._opener.read(self.docket.data_filename())
+            except FileNotFoundError:
+                # read race detected between docket and data file
+                # reload the docket and retry
+                self._docket = None
+        if data is None:
+            assert attempts >= V2_MAX_READ_ATTEMPTS
+            msg = b"dirstate read race happened %d times in a row"
+            msg %= attempts
+            raise error.Abort(msg)
+        return self._opener.read(self.docket.data_filename())
+
     def write_v2_no_append(self, tr, st, meta, packed):
         old_docket = self.docket
         new_docket = docketmod.DirstateDocket.with_new_uuid(
@@ -290,14 +328,15 @@
     ### disk interaction
 
     def read(self):
-        # ignore HG_PENDING because identity is used only for writing
-        self._set_identity()
+        testing.wait_on_cfg(self._ui, b'dirstate.pre-read-file')
+        if self._use_dirstate_v2:
 
-        if self._use_dirstate_v2:
             if not self.docket.uuid:
                 return
-            st = self._opener.read(self.docket.data_filename())
+            testing.wait_on_cfg(self._ui, b'dirstate.post-docket-read-file')
+            st = self._read_v2_data()
         else:
+            self._set_identity()
             st = self._readdirstatefile()
 
         if not st:
@@ -556,19 +595,39 @@
             # ignore HG_PENDING because identity is used only for writing
             self._set_identity()
 
+            testing.wait_on_cfg(self._ui, b'dirstate.pre-read-file')
             if self._use_dirstate_v2:
-                if self.docket.uuid:
-                    # TODO: use mmap when possible
-                    data = self._opener.read(self.docket.data_filename())
+                self.docket  # load the data if needed
+                inode = (
+                    self.identity.stat.st_ino
+                    if self.identity is not None
+                    and self.identity.stat is not None
+                    else None
+                )
+                testing.wait_on_cfg(self._ui, b'dirstate.post-docket-read-file')
+                if not self.docket.uuid:
+                    data = b''
+                    self._map = rustmod.DirstateMap.new_empty()
                 else:
-                    data = b''
-                self._map = rustmod.DirstateMap.new_v2(
-                    data, self.docket.data_size, self.docket.tree_metadata
-                )
+                    data = self._read_v2_data()
+                    self._map = rustmod.DirstateMap.new_v2(
+                        data,
+                        self.docket.data_size,
+                        self.docket.tree_metadata,
+                        self.docket.uuid,
+                        inode,
+                    )
                 parents = self.docket.parents
             else:
+                self._set_identity()
+                inode = (
+                    self.identity.stat.st_ino
+                    if self.identity is not None
+                    and self.identity.stat is not None
+                    else None
+                )
                 self._map, parents = rustmod.DirstateMap.new_v1(
-                    self._readdirstatefile()
+                    self._readdirstatefile(), inode
                 )
 
             if parents and not self._dirtyparents:
@@ -638,8 +697,10 @@
                 return
 
             # We can only append to an existing data file if there is one
-            can_append = self.docket.uuid is not None
-            packed, meta, append = self._map.write_v2(can_append)
+            write_mode = self._write_mode
+            if self.docket.uuid is None:
+                write_mode = WRITE_MODE_FORCE_NEW
+            packed, meta, append = self._map.write_v2(write_mode)
             if append:
                 docket = self.docket
                 data_filename = docket.data_filename()