stream: create a `e.preserve_volatiles` method directly on StoreEntry

For cache, we will need an atomic operation that detect the file and preserve
it, so we start by creating an API that make such atomic operation possible. We
will actually make things atomic in the next changesets.

mercurial/store.py

@@ -524,6 +524,16 class BaseStoreEntry:
         assert vfs is not None
         return [f.get_stream(vfs, volatiles) for f in self.files()]
 
+    def preserve_volatiles(self, vfs, volatiles):
+        """Use a VolatileManager to preserve the state of any volatile file
+
+        This is useful for code that need a consistent view of the content like stream clone.
+        """
+        if self.maybe_volatile:
+            for f in self.files():
+                if f.is_volatile:
+                    volatiles(vfs.join(f.unencoded_path))
+
 
 @attr.s(slots=True, init=False)
 class SimpleStoreEntry(BaseStoreEntry):

mercurial/streamclone.py

@@ -714,9 +714,8 def _emit2(repo, entries):
         with util.nogc():
             # record the expected size of every file
             for k, vfs, e in entries:
+                e.preserve_volatiles(vfs, volatiles)
                 for f in e.files():
-                    if f.is_volatile:
-                        volatiles(vfs.join(f.unencoded_path))
                     file_count += 1
                     totalfilesize += f.file_size(vfs)
 
@@ -777,11 +776,11 def _emit3(repo, entries):
         # make sure we preserve volatile files
         for k, vfs, e in entries:
             if e.maybe_volatile:
+                e.preserve_volatiles(vfs, volatiles)
                 for f in e.files():
                     if f.is_volatile:
                         # record the expected size under lock
                         f.file_size(vfs)
-                        volatiles(vfs.join(f.unencoded_path))
 
         total_entry_count = len(entries)