mount: thread safety

There is actual concurrency when using `pyfuse3` in Borg, which necessitates thread safety mechanisms like the recently added `threading.Lock`.

Borg's `pyfuse3` implementation is built on top of the **`trio`** async framework. When Borg is started in `pyfuse3` mode, it calls `trio.run(llfuse.main)`.

1.  **Async/Await Model**: Unlike the classic `llfuse` (which Borg runs with `workers=1` to remain single-threaded), `pyfuse3` uses an asynchronous event loop. While it often runs on a single OS thread, `trio` allows multiple tasks to be "in-flight" simultaneously.
2.  **Context Switching**: When an operation (like reading metadata or data) hits an `await` point—such as during network I/O with a remote repository or disk I/O—`trio` can suspend that task and switch to another one.
3.  **Parallel Archive Loading**: In `borg mount` (without a specific archive specified), archives are loaded lazily. If a user or a process (like `find` or a file manager) triggers a `lookup` or `opendir` on multiple archive directories nearly simultaneously, multiple `check_pending_archive` calls can be active at once.
4.  **Race Conditions**: Because `iter_archive_items` is a generator that yields control back to the caller (which, in the `pyfuse3` case, happens within an async wrapper), and because it performs I/O via `get_many`, it creates windows where one archive's loading process can be suspended while another begins.

Even if running on a single OS thread, the interleaved execution of async tasks can lead to the same data corruption issues as multi-threading:
-   **Shared State**: Both tasks access the same `ItemCache` instance.
-   **Inode Collisions**: If two tasks read the `write_offset` before either has incremented it, they will assign the same inode numbers to different files and overwrite each other's metadata in the `meta` bytearray.
-   **Global Interpreter Lock (GIL)**: While the GIL protects Python's internal memory integrity, it does not prevent logical race conditions at the application level when tasks are interleaved.

The use of `threading.Lock` in `check_pending_archive` and the direct instance variable access in `ItemCache` ensure that even when `trio` switches between concurrent FUSE requests, the internal state remains consistent and inode assignment remains unique across all archives.

Author: ThomasWaldmann

src/borg/fuse.py

@@ -6,6 +6,7 @@
 import struct
 import sys
 import tempfile
+import threading
 import time
 from collections import defaultdict
 from signal import SIGINT
@@ -155,18 +156,15 @@ def iter_archive_items(self, archive_item_ids, filter=None, consider_part_files=
         last_chunk_length = 0
         msgpacked_bytes = b''
 
-        write_offset = self.write_offset
-        meta = self.meta
         pack_indirect_into = self.indirect_entry_struct.pack_into
 
         for key, (csize, data) in zip(archive_item_ids, self.decrypted_repository.get_many(archive_item_ids)):
             # Store the chunk ID in the meta-array
-            if write_offset + 32 >= len(meta):
-                self.meta = meta = meta + bytes(self.GROW_META_BY)
-            meta[write_offset:write_offset + 32] = key
-            current_id_offset = write_offset
-            write_offset += 32
-            self.write_offset = write_offset
+            if self.write_offset + 32 >= len(self.meta):
+                self.meta = self.meta + bytes(self.GROW_META_BY)
+            self.meta[self.write_offset:self.write_offset + 32] = key
+            current_id_offset = self.write_offset
+            self.write_offset += 32
 
             chunk_begin += last_chunk_length
             last_chunk_length = len(data)
@@ -200,8 +198,8 @@ def iter_archive_items(self, archive_item_ids, filter=None, consider_part_files=
                 current_spans_chunks = stream_offset - current_item_length < chunk_begin
                 msgpacked_bytes = b''
 
-                if write_offset + 9 >= len(meta):
-                    self.meta = meta = meta + bytes(self.GROW_META_BY)
+                if self.write_offset + 9 >= len(self.meta):
+                    self.meta = self.meta + bytes(self.GROW_META_BY)
 
                 # item entries in the meta-array come in two different flavours, both nine bytes long.
                 # (1) for items that span chunks:
@@ -222,15 +220,14 @@ def iter_archive_items(self, archive_item_ids, filter=None, consider_part_files=
                 if current_spans_chunks:
                     pos = self.fd.seek(0, io.SEEK_END)
                     self.fd.write(current_item)
-                    meta[write_offset:write_offset + 9] = b'S' + pos.to_bytes(8, 'little')
+                    self.meta[self.write_offset:self.write_offset + 9] = b'S' + pos.to_bytes(8, 'little')
                     self.direct_items += 1
                 else:
                     item_offset = stream_offset - current_item_length - chunk_begin
-                    pack_indirect_into(meta, write_offset, b'I', write_offset - current_id_offset, item_offset)
+                    pack_indirect_into(self.meta, self.write_offset, b'I', self.write_offset - current_id_offset, item_offset)
                     self.indirect_items += 1
-                inode = write_offset + self.offset
-                write_offset += 9
-                self.write_offset = write_offset
+                inode = self.write_offset + self.offset
+                self.write_offset += 9
 
                 yield inode, item
 
@@ -269,6 +266,7 @@ def __init__(self, key, manifest, repository, args, decrypted_repository):
         self.uid_forced = None
         self.gid_forced = None
         self.umask = 0
+        self.lock = threading.Lock()
 
     def _create_filesystem(self):
         self._create_dir(parent=1)  # first call, create root dir (inode == 1)
@@ -304,9 +302,10 @@ def get_item(self, inode):
 
     def check_pending_archive(self, inode):
         # Check if this is an archive we need to load
-        archive_name = self.pending_archives.pop(inode, None)
-        if archive_name is not None:
-            self._process_archive(archive_name, [os.fsencode(archive_name)])
+        with self.lock:
+            archive_name = self.pending_archives.pop(inode, None)
+            if archive_name is not None:
+                self._process_archive(archive_name, [os.fsencode(archive_name)])
 
     def _allocate_inode(self):
         self.inode_count += 1