fix #9872: refresh repository lock of an idle mount

src/borg/archiver/lock_cmds.py

@@ -3,8 +3,9 @@
 from ._common import with_repository
 from ..cache import Cache
 from ..constants import *  # NOQA
-from ..helpers import prepare_subprocess_env, set_ec, CommandError, ThreadRunner
+from ..helpers import prepare_subprocess_env, set_ec, CommandError
 from ..helpers.argparsing import ArgumentParser, REMAINDER
+from ..storelocking import LockRefresher
 
 from ..logger import create_logger
 
@@ -17,7 +18,7 @@ def do_with_lock(self, args, repository):
         """Runs a user-specified command with the repository lock held."""
         # the repository lock needs to get refreshed regularly, or it will be killed as stale.
         # refreshing the lock is not part of the repository API, so we do it indirectly via repository.info.
-        lock_refreshing_thread = ThreadRunner(sleep_interval=60, target=repository.info)
+        lock_refreshing_thread = LockRefresher(repository.info, sleep_interval=60)
         lock_refreshing_thread.start()
         env = prepare_subprocess_env(system=True)
         try:

src/borg/fuse.py

@@ -12,6 +12,11 @@
   asynchronous manner. However, as long as we do not use any asynchronous
   operations (like using "await") in this code, it is still de facto
   synchronous, and we are safe.
+
+The only exception is a background thread that periodically refreshes the
+repository lock while the mount is idle (see #9872). borgstore connections are
+not thread-safe, so all repository access from the FUSE handlers and the refresh
+thread is serialized via self._repo_lock.
 """
 
 import errno
@@ -22,6 +27,7 @@
 import struct
 import sys
 import tempfile
+import threading
 import time
 from collections import defaultdict, Counter
 from signal import SIGINT
@@ -65,6 +71,7 @@ def async_wrapper(fn):
 from .helpers import daemonizing, signal_handler, format_file_size, bin_to_hex, Error
 from .helpers import HardLinkManager
 from .helpers import msgpack
+from .storelocking import LockRefresher
 from .helpers.lrucache import LRUCache
 from .item import Item
 from .platform import uid2user, gid2group
@@ -296,6 +303,9 @@ def __init__(self, manifest, args, repository):
         # Archives to be loaded when first accessed, mapped by their placeholder inode
         self.pending_archives = {}
         self.cache = ItemCache(repository, self.repo_objs)
+        # serializes all repository access (FUSE handlers and the background lock-refresh
+        # thread), because borgstore connections are not thread-safe. see _lock_refresh.
+        self._repo_lock = threading.RLock()
         self.allow_damaged_files = False
         self.versions = False
         self.uid_forced = None
@@ -334,15 +344,17 @@ def get_item(self, inode):
             return self._items[inode]
         except KeyError:
             # while self.cache does some internal caching, it has still quite some overhead, so we cache the result.
-            item = self.cache.get(inode)
+            with self._repo_lock:
+                item = self.cache.get(inode)
             self._inode_cache[inode] = item
             return item
 
     def check_pending_archive(self, inode):
         # Check if this is an archive we need to load
         archive_info = self.pending_archives.pop(inode, None)
         if archive_info is not None:
-            self._process_archive(archive_info.id, [os.fsencode(self.archive_root_dir[archive_info.id])])
+            with self._repo_lock:
+                self._process_archive(archive_info.id, [os.fsencode(self.archive_root_dir[archive_info.id])])
 
     def _allocate_inode(self):
         self.inode_count += 1
@@ -602,12 +614,17 @@ def pop_option(options, key, present, not_present, wanted_type, int_base=0):
         # job - seeing the mountpoint empty, rsync would delete everything in the
         # mirror.
         umount = False
+        # keep the repository lock of an idle mount alive, so it is not killed as stale (see #9872).
+        # started here (after a possible daemonizing fork, as threads do not survive fork()).
+        lock_refreshing_thread = LockRefresher(self.repository_uncached.info, sleep_interval=60, lock=self._repo_lock)
+        lock_refreshing_thread.start()
         try:
             with signal_handler("SIGUSR1", self.sig_info_handler), signal_handler("SIGINFO", self.sig_info_handler):
                 signal = fuse_main()
             # no crash and no signal (or it's ^C and we're in the foreground) -> umount request
             umount = signal is None or (signal == SIGINT and foreground)
         finally:
+            lock_refreshing_thread.terminate()
             llfuse.close(umount)
 
     @async_wrapper
@@ -721,7 +738,8 @@ def read(self, fh, offset, size):
                     del self.data_cache[id]
             else:
                 try:
-                    cdata = self.repository_uncached.get(id)
+                    with self._repo_lock:
+                        cdata = self.repository_uncached.get(id)
                 except Repository.ObjectNotFound:
                     if self.allow_damaged_files:
                         data = zeros[:s]

src/borg/hlfuse.py

@@ -3,6 +3,7 @@
 import hashlib
 import os
 import stat
+import threading
 import time
 from collections import Counter
 from typing import TYPE_CHECKING
@@ -27,6 +28,7 @@
 from .helpers import daemonizing, signal_handler, bin_to_hex, Error
 from .helpers import HardLinkManager
 from .helpers import msgpack
+from .storelocking import LockRefresher
 from .helpers.lrucache import LRUCache
 from .item import Item
 from .platform import uid2user, gid2group
@@ -87,6 +89,9 @@ def __init__(self, manifest, args, repository):
         self._manifest = manifest
         self.repo_objs = manifest.repo_objs
         self.repository = repository
+        # serializes all repository access (FUSE handlers and the background lock-refresh
+        # thread), because borgstore connections are not thread-safe. see _lock_refresh.
+        self._repo_lock = threading.RLock()
 
         self.default_uid = os.getuid()
         self.default_gid = os.getgid()
@@ -171,7 +176,8 @@ def _create_filesystem(self):
     def check_pending_archive(self, node):
         archive_info = self.pending_archives.pop(node, None)
         if archive_info is not None:
-            self._process_archive(archive_info.id, node)
+            with self._repo_lock:
+                self._process_archive(archive_info.id, node)
 
     def _iter_archive_items(self, archive_item_ids, filter=None):
         unpacker = msgpack.Unpacker()
@@ -532,9 +538,16 @@ def pop_option(options, key, present, not_present, wanted_type, int_base=0):
                 logger.debug("fuse: mount repo, going to background: migrating lock.")
                 self.repository.migrate_lock(old_id, new_id)
 
-        # Run the FUSE main loop in foreground (we might be daemonized already or not)
-        with signal_handler("SIGUSR1", self.sig_info_handler), signal_handler("SIGINFO", self.sig_info_handler):
-            hlfuse.FUSE(self, mountpoint, options, foreground=True, use_ino=True)
+        # keep the repository lock of an idle mount alive, so it is not killed as stale (see #9872).
+        # started here (after a possible daemonizing fork, as threads do not survive fork()).
+        lock_refreshing_thread = LockRefresher(self.repository.info, sleep_interval=60, lock=self._repo_lock)
+        lock_refreshing_thread.start()
+        try:
+            # Run the FUSE main loop in foreground (we might be daemonized already or not)
+            with signal_handler("SIGUSR1", self.sig_info_handler), signal_handler("SIGINFO", self.sig_info_handler):
+                hlfuse.FUSE(self, mountpoint, options, foreground=True, use_ino=True)
+        finally:
+            lock_refreshing_thread.terminate()
 
     def statfs(self, path):
         debug_log(f"statfs(path={path!r})")
@@ -645,7 +658,8 @@ def read(self, path, size, offset, fi):
             else:
                 try:
                     # Direct repository access
-                    cdata = self.repository.get(id)
+                    with self._repo_lock:
+                        cdata = self.repository.get(id)
                 except Repository.ObjectNotFound:
                     if self.allow_damaged_files:
                         data = zeros[:s]

src/borg/storelocking.py

@@ -2,6 +2,7 @@
 import hashlib
 import json
 import random
+import threading
 import time
 
 from borgstore.store import ObjectNotFound
@@ -296,3 +297,57 @@ def refresh(self):
                     logger.debug("LOCK-REFRESH: our lock was killed while refreshing it, no safe way to continue.")
                     self._delete_lock(new_key, ignore_not_found=True, update_last_refresh=True)
                     raise LockTimeout(str(self.store))
+
+
+class LockRefresher:
+    def __init__(self, refresh_callable, sleep_interval=60, lock=None):
+        """
+        Periodically refreshes the repository lock in a background thread.
+
+        :param refresh_callable: Callable (e.g. repository.info) to refresh the lock.
+        :param sleep_interval: Frequency of refresh attempts (in seconds).
+        :param lock: Optional reentrant lock (RLock) to serialize repository access.
+        """
+        self.refresh_callable = refresh_callable
+        self.sleep_interval = sleep_interval
+        self.lock = lock
+        self._thread = None
+        self._keep_running = threading.Event()
+        self._keep_running.set()
+
+    def _run(self):
+        while self._keep_running.is_set():
+            try:
+                if self.lock is not None:
+                    with self.lock:
+                        self.refresh_callable()
+                else:
+                    self.refresh_callable()
+            except LockTimeout:
+                logger.error("Lock refresh thread: lock has been killed/timed out. Terminating refresh.")
+                break
+            except Exception as e:
+                logger.warning(f"Lock refresh thread: temporary error during lock refresh: {e}. Will retry.")
+
+            # sleep up to self.sleep_interval in small steps to allow quick shutdown
+            count = 1000
+            micro_sleep = float(self.sleep_interval) / count
+            while self._keep_running.is_set() and count > 0:
+                time.sleep(micro_sleep)
+                count -= 1
+
+    def start(self):
+        # daemon=True so a refresh that is stuck in a blocking refresh_callable() (e.g. network
+        # I/O to a remote repo that never returns) can never keep the process alive on exit.
+        self._thread = threading.Thread(target=self._run, name="LockRefresher", daemon=True)
+        self._thread.start()
+
+    def terminate(self):
+        if self._thread is not None:
+            self._keep_running.clear()
+            # bounded join: if refresh_callable() is wedged (e.g. a hung remote-repo request),
+            # don't let it stall shutdown - e.g. a FUSE unmount waiting for this. the thread is a
+            # daemon, so it is torn down at interpreter exit even if it is still alive here.
+            self._thread.join(timeout=5.0)
+            if self._thread.is_alive():
+                logger.warning("Lock refresh thread did not stop within 5s; continuing shutdown.")

src/borg/testsuite/archiver/mount_cmds_test.py

@@ -380,3 +380,36 @@ def wrapper(self, old_id, new_id):
     finally:
         # Undecorate
         Lock.migrate_lock = Lock.migrate_lock.__wrapped__
+
+
+def test_fuse_lock_refresh_calls_repository_info():
+    # regression test for #9872: an idle mount must keep its repository lock alive via a
+    # background refresh. LockRefresher periodically calls repository.info() while holding
+    # the serialization lock (the FUSE handlers use the same lock), so repo access from the
+    # refresh thread and the FUSE handlers never runs concurrently.
+    import threading
+
+    from ...storelocking import LockRefresher
+
+    # a plain Lock is enough here and, unlike RLock, has .locked() on all Python versions.
+    # in the FUSE code the lock is an RLock (it needs to be reentrant), but LockRefresher
+    # only ever acquires/releases it, so a plain Lock exercises the same code path.
+    lock = threading.Lock()
+    called = threading.Event()
+    held_while_calling = []
+
+    class FakeRepository:
+        def info(self):
+            held_while_calling.append(lock.locked())  # the lock must be held while we run
+            called.set()
+            return dict(id=b"\x00" * 32, version=3)
+
+    refresher = LockRefresher(FakeRepository().info, sleep_interval=60, lock=lock)
+    refresher.start()
+    try:
+        assert called.wait(timeout=10), "LockRefresher did not call repository.info()"
+    finally:
+        refresher.terminate()
+
+    assert held_while_calling and all(held_while_calling), "repository.info() must run while holding the lock"
+    assert not lock.locked(), "lock must be released again after refreshing"