locking: deal with clock skew

docs/faq.rst

@@ -29,6 +29,17 @@ Can I back up from multiple servers into a single repository?
 
 Yes, you can! Even simultaneously.
 
+The clocks of machines sharing a repository should be roughly synchronized
+(e.g. via NTP): repository locks and archive/manifest timestamps are based on
+the clients' clocks, so big clock differences between clients can cause
+trouble. Where the storage backend provides object timestamps (file, sftp, s3
+and current rest servers - but not rclone), borg cross-checks lock staleness
+against the storage's clock (so a client with a wrong clock can not break
+another client's healthy lock) and logs a warning when it detects that the
+clocks of concurrently active clients differ by more than a few minutes.
+The storage's own clock does not need to be correct - it is only used as a
+common reference between the clients.
+
 Can I back up to multiple swapped backup targets?
 --------------------------------------------------
 

docs/internals/data-structures.rst

@@ -1136,17 +1136,48 @@ To implement locking based on ``borgstore``, borg stores objects below locks/.
 
 The objects contain:
 
-- a timestamp when lock was created (or refreshed)
+- a timestamp when lock was created (or refreshed), stamped by the clock of
+  the machine writing the lock
 - host / process / thread information about lock owner
 - lock type: exclusive or shared
 
+Where the storage backend provides object timestamps (file, sftp, s3 and
+current rest servers - but not rclone), borg additionally uses the lock
+object's store-side mtime, which is stamped by the storage's clock.
+
 Using that information, borg implements:
 
+- lock auto-removal if the owner process is dead. the primary purpose of this
+  is to quickly get rid of stale locks by borg processes on the same machine.
+  process liveness is local knowledge, independent of any clock, so this check
+  runs first and can not be vetoed by store timestamps.
 - lock auto-expiry: if a lock is old and has not been refreshed in time,
   it will be automatically ignored and deleted. the primary purpose of this
   is to get rid of stale locks by borg processes on other machines.
-- lock auto-removal if the owner process is dead. the primary purpose of this
-  is to quickly get rid of stale locks by borg processes on the same machine.
+
+Lock auto-expiry must never kill a healthy lock just because its writer's
+clock is skewed against ours (see :issue:`9870`), thus a lock may only be
+expired by age if it looks stale in both clock domains:
+
+- writer / local clock domain: local "now" vs. the lock's content timestamp.
+- store clock domain: store "now" vs. the lock object's store-side mtime.
+  store "now" is computed from the mtime of our own lock object plus the
+  monotonic time elapsed since we created it, so this comparison stays
+  entirely within the storage's clock domain - neither the clients' nor the
+  storage's absolute clock error matters.
+
+Store-side mtimes are advisory only: they can veto an expiry, but they can
+never cause one on their own, so a hostile or broken store gains no new
+capabilities. A client that has no own lock object yet can not compute store
+"now" and defers the expiry decision until it has created one. If the backend
+can not provide store-side mtimes (mtime is 0), staleness is judged by the
+content timestamp alone.
+
+As each lock object carries two timestamps of the same write instant (content
+timestamp: writer's clock, store-side mtime: storage's clock), the writers'
+clock offsets relative to the storage are comparable, with the storage's
+absolute clock error cancelled out. borg uses this to warn (once) if the
+clocks of concurrently active clients differ by more than a few minutes.
 
 Breaking the locks
 ------------------

pyproject.toml

@@ -31,7 +31,7 @@ license = "BSD-3-Clause"
 license-files = ["LICENSE", "AUTHORS"]
 dependencies = [
   "borghash ~= 0.2.0",
-  "borgstore[rest,blake3] ~= 0.6.0",
+  "borgstore[rest,blake3] ~= 0.6.1",
   "msgpack >=1.0.3, <=1.2.1",
   "packaging",
   "platformdirs >=3.0.0, <5.0.0; sys_platform == 'darwin'",  # for macOS: breaking changes in 3.0.0.
@@ -53,9 +53,9 @@ mfusepy = ["mfusepy >= 3.1.0, <4.0.0"]  # fuse 2+3, high-level
 # a pypi release of borgbackup can't contain a dependency on github!
 # mfusepym = ["mfusepy @ git+https://github.com/mxmlnkn/mfusepy.git@master"]
 nofuse = []
-s3 = ["borgstore[rest,blake3,s3] ~= 0.6.0"]
-sftp = ["borgstore[rest,blake3,sftp] ~= 0.6.0"]
-rclone = ["borgstore[rest,blake3,rclone] ~= 0.6.0"]
+s3 = ["borgstore[rest,blake3,s3] ~= 0.6.1"]
+sftp = ["borgstore[rest,blake3,sftp] ~= 0.6.1"]
+rclone = ["borgstore[rest,blake3,rclone] ~= 0.6.1"]
 cockpit = ["textual>=6.8.0"]  # might also work with older versions, untested
 
 [project.urls]
@@ -151,6 +151,7 @@ dummy-variable-rgx = "^(_+|(_+[a-zA-Z0-9_]*[a-zA-Z0-9]+?))$"
 "src/borg/testsuite/archiver/return_codes_test.py" = ["F811"]
 "src/borg/testsuite/benchmark_test.py" = ["F811"]
 "src/borg/testsuite/platform/platform_test.py" = ["F811"]
+"src/borg/testsuite/storelocking_test.py" = ["F811"]
 
 [tool.pytest.ini_options]
 markers = []

src/borg/constants.py

@@ -81,6 +81,12 @@
 # this, the pack is re-verified.
 MAX_CLOCK_SKEW = 7200  # [s]
 
+# Maximum tolerated clock skew between the clocks of borg clients concurrently using the same
+# repository before a warning is emitted (see storelocking). Must be well below the lock stale
+# timeout (30 min) / refresh interval (15 min) so users get warned long before skew could
+# interfere with lock staleness judgment or manifest timestamps.
+MAX_MUTUAL_CLOCK_SKEW = 300  # [s]
+
 # How many segment files Borg puts into a single directory by default.
 DEFAULT_SEGMENTS_PER_DIR = 1000
 

src/borg/storelocking.py

@@ -4,15 +4,23 @@
 import random
 import threading
 import time
+from collections import namedtuple
 
 from borgstore.store import ObjectNotFound
 
 from . import platform
+from .constants import MAX_MUTUAL_CLOCK_SKEW
 from .helpers import Error, ErrorWithTraceback
 from .logger import create_logger
 
 logger = create_logger(__name__)
 
+# all we know about the lock object we most recently created: its store key, its content timestamp
+# (stamped by our clock) and its store-side mtime [s] (stamped by the store's clock, harvested from
+# lock listings, None until harvested), plus time.monotonic() at its creation. always replaced as a
+# whole, so concurrent readers (e.g. a LockRefresher thread) never see a torn mix of its fields.
+LockAnchor = namedtuple("LockAnchor", "key dt mtime monotonic")
+
 
 class LockError(Error):
     """Failed to acquire the lock {}."""
@@ -80,6 +88,11 @@ def __init__(self, store, exclusive=False, sleep=None, timeout=1.0, stale=30 * 6
         self.refresh_td = datetime.timedelta(seconds=stale // 2)  # don't refresh it if younger
         self.last_refresh_dt = None
         self.my_lock_key = None  # store key of the lock we currently hold, None if we hold none
+        # LockAnchor of the lock object we most recently created - its mtime and monotonic fields
+        # together let us compute the current time in the store's clock domain, see _store_now().
+        # it deliberately outlives its lock object: the calibration stays valid after deletion.
+        self.my_lock_anchor = None
+        self.skew_warned = False  # emit the clock-skew warning only once per Lock instance
         self.id = id or platform.get_process_id()
         assert len(self.id) == 3
         logger.debug(f"LOCK-INIT: initializing. store: {store}, stale: {stale}s, refresh: {stale // 2}s.")
@@ -96,9 +109,10 @@ def __exit__(self, exc_type, exc_val, exc_tb):
     def __repr__(self):
         return f"<{self.__class__.__name__}: {self.id!r}>"
 
-    def _create_lock(self, *, exclusive=None, update_last_refresh=False):
+    def _create_lock(self, *, exclusive=None, dt=None, update_last_refresh=False):
         assert exclusive is not None
-        now = datetime.datetime.now(datetime.UTC)
+        # dt: explicit content timestamp (default: now) - tests use it to simulate skewed clocks.
+        now = dt if dt is not None else datetime.datetime.now(datetime.UTC)
         timestamp = now.isoformat(timespec="milliseconds")
         lock = dict(exclusive=exclusive, hostid=self.id[0], processid=self.id[1], threadid=self.id[2], time=timestamp)
         value = json.dumps(lock).encode("utf-8")
@@ -109,6 +123,10 @@ def _create_lock(self, *, exclusive=None, update_last_refresh=False):
             # we parse the timestamp string to get *precisely* the datetime in the lock:
             self.last_refresh_dt = datetime.datetime.fromisoformat(timestamp)
             self.my_lock_key = key
+            # the store-side mtime of the new lock object is not known yet - it is harvested
+            # from the next locks listing. anchor the monotonic clock at creation time so the
+            # harvested mtime can be extrapolated to "now" later, see _store_now().
+            self.my_lock_anchor = LockAnchor(key, self.last_refresh_dt, None, time.monotonic())
         return key
 
     def _delete_lock(self, key, *, ignore_not_found=False, update_last_refresh=False):
@@ -122,24 +140,112 @@ def _delete_lock(self, key, *, ignore_not_found=False, update_last_refresh=False
             if update_last_refresh:
                 self.last_refresh_dt = None
                 self.my_lock_key = None
+                # my_lock_anchor is deliberately kept: it is a store-clock calibration, not a
+                # property of the deleted object. keeping it lets an acquire that is blocked by
+                # a healthy-but-skewed lock veto the kill on the first listing of every retry,
+                # instead of re-deferring and re-creating a transient lock each time.
 
     def _is_our_lock(self, lock):
         return self.id == (lock["hostid"], lock["processid"], lock["threadid"])
 
+    def _store_now(self):
+        """Return the current time in the store's clock domain [UNIX timestamp], or None if unknown."""
+        anchor = self.my_lock_anchor  # single read - it gets replaced atomically as a whole
+        if anchor is None or anchor.mtime is None:
+            return None
+        # note: on most platforms time.monotonic() does not advance while the machine is suspended,
+        # so after a suspend the extrapolation below lags behind store "now" by up to the suspend
+        # duration. that errs towards NOT considering other locks stale (the safe direction), but
+        # do not extrapolate from a too old anchor at all: its age is measured with our wall clock
+        # at both ends, so suspends count here. self-heals at our next lock creation/refresh.
+        if datetime.datetime.now(datetime.UTC) > anchor.dt + self.stale_td:
+            return None
+        return anchor.mtime + (time.monotonic() - anchor.monotonic)
+
+    def _mutual_skew(self, lock):
+        """
+        Return the clock skew [s] between us and the writer of <lock> (positive: their clock runs
+        ahead of ours), or None if it can not be determined.
+
+        Each lock object carries two timestamps of the same write instant: its content timestamp
+        (stamped by the writer's clock) and its store-side mtime (stamped by the store's clock).
+        Their difference is that writer's clock offset relative to the store; comparing two
+        writers' offsets yields their mutual skew, with the store's absolute clock error cancelled
+        out (the store's clock is only used as a common reference and may itself be wrong).
+        """
+        anchor = self.my_lock_anchor  # single read - it gets replaced atomically as a whole
+        if not lock.get("mtime") or anchor is None or anchor.mtime is None:
+            return None
+        offset_self = anchor.dt.timestamp() - anchor.mtime
+        offset_other = lock["dt"].timestamp() - lock["mtime"]
+        return offset_other - offset_self
+
+    def _warn_clock_skew(self, lock, skew):
+        if self.skew_warned:
+            return
+        self.skew_warned = True
+        logger.warning(
+            f"Clock skew of ~{abs(skew):.0f}s detected between this machine and the borg client on "
+            f"{lock['hostid']!r} (also using this repository). "
+            f"The clocks of machines sharing a repository should be synchronized (e.g. via NTP)."
+        )
+
+    def _check_clock_skew(self, locks):
+        """Warn (once) if another current lock writer's clock is skewed against ours."""
+        if self.skew_warned:
+            return
+        for lock in locks.values():
+            if lock["key"] == self.my_lock_key:
+                continue
+            skew = self._mutual_skew(lock)
+            if skew is not None and abs(skew) > MAX_MUTUAL_CLOCK_SKEW:
+                self._warn_clock_skew(lock, skew)
+
     def _is_stale_lock(self, lock):
         if lock["key"] == self.my_lock_key:
             # the lock we are currently holding: we are obviously alive and can refresh or
             # release it, so it must never be considered stale (and get deleted), no matter
             # how old it is. it can get old e.g. if the machine is suspended while doing a
             # backup or if there is a long stretch of work without repository access, see #9883.
             return False
+        if not platform.process_alive(lock["hostid"], lock["processid"], lock["threadid"]):
+            # the lock owner is a process on THIS machine and it is dead - local knowledge,
+            # independent of any clock and of the store. checked first (and never vetoed by
+            # store timestamps below), so a store serving bogus, always-fresh mtimes can not
+            # keep an abandoned lock alive forever and block us.
+            logger.debug(f"LOCK-STALE: we KNOW that the lock-owning process is dead. lock: {lock}.")
+            return True
         now = datetime.datetime.now(datetime.UTC)
         if now > lock["dt"] + self.stale_td:
+            # the lock looks stale, judging by its content timestamp (writer's clock) vs. our
+            # local clock. but that comparison breaks down if the writer's clock is skewed
+            # against ours, and we must never kill a healthy lock (data loss hazard, #9870).
+            # thus, cross-check in the store's clock domain: the lock object's store-side mtime
+            # vs. store "now". store timestamps are advisory only: they can veto a kill here,
+            # but they can never cause a kill on their own (the store might be hostile).
+            # residual risk: a store clock that steps BACK by more than the stale timeout during
+            # the lifetime of our anchor defeats the veto; accepted - pre-#9870 there was no
+            # cross-check at all, and re-anchoring at each lock creation bounds the window.
+            if lock["mtime"]:
+                store_now = self._store_now()
+                if store_now is None:
+                    # we do not have a store-written object of our own yet, so we can not compute
+                    # store "now". defer: the caller may create our lock first and list again -
+                    # then we get here again with store_now available. never kill unconfirmed.
+                    lock["maybe_stale"] = True
+                    logger.debug(f"LOCK-STALE: lock looks stale, deferring until store time is known. lock: {lock}.")
+                    return False
+                if store_now <= lock["mtime"] + self.stale_td.total_seconds():
+                    # the store saw this lock object being written recently: it is NOT stale.
+                    # either its writer's clock is skewed against ours (the skew check in
+                    # _get_locks warns about that) or our store "now" estimate lags behind
+                    # (e.g. time.monotonic() stood still while we were suspended). never kill it!
+                    logger.debug(f"LOCK-STALE: lock looks stale locally, but not to the store. lock: {lock}.")
+                    return False
+            # either both clock domains agree that the lock is stale, or the backend can not
+            # provide store-side mtimes (mtime == 0) and the content timestamp has to suffice.
             logger.debug(f"LOCK-STALE: lock is too old, it was not refreshed. lock: {lock}.")
             return True
-        if not platform.process_alive(lock["hostid"], lock["processid"], lock["threadid"]):
-            logger.debug(f"LOCK-STALE: we KNOW that the lock-owning process is dead. lock: {lock}.")
-            return True
         return False
 
     def _get_locks(self):
@@ -159,14 +265,29 @@ def _get_locks(self):
             lock = json.loads(content.decode("utf-8"))
             lock["key"] = key
             lock["dt"] = datetime.datetime.fromisoformat(lock["time"])
-            if self._is_stale_lock(lock):
+            lock["mtime"] = info.mtime  # store-side mtime [s], 0 if the backend can not provide it
+            locks[key] = lock
+        my_key = self.my_lock_key  # single read - a LockRefresher thread may rebind it concurrently
+        if my_key in locks:
+            # harvest the store-side mtime of our own lock object from this listing into the
+            # anchor set at its creation (see _create_lock / _store_now) - but only if the anchor
+            # still describes the same lock object (a concurrent refresh may have replaced it).
+            mtime = locks[my_key]["mtime"]
+            anchor = self.my_lock_anchor
+            if mtime and anchor is not None and anchor.key == my_key:
+                self.my_lock_anchor = anchor._replace(mtime=mtime)
+        for key in list(locks):
+            if self._is_stale_lock(locks[key]):
                 # ignore it and delete it (even if it is not from us).
                 # note: this is never the lock we currently hold (see _is_stale_lock), so this
                 # must not touch last_refresh_dt / my_lock_key - a stale lock matching our id
                 # can only be a leftover of a dead process (pid reuse), not our own lock.
                 self._delete_lock(key, ignore_not_found=True)
-            else:
-                locks[key] = lock
+                del locks[key]
+        # check for clock skew on every listing, not just on acquire success: the listing that
+        # satisfies an exclusive acquire only contains our own lock, so a skewed peer is only
+        # visible in earlier listings, e.g. while we wait for its healthy lock to go away.
+        self._check_clock_skew(locks)
         return locks
 
     def _find_locks(self, *, only_exclusive=False, only_mine=False):
@@ -188,8 +309,11 @@ def acquire(self):
         started = time.monotonic()
         while time.monotonic() - started < self.timeout:
             exclusive_locks = self._find_locks(only_exclusive=True)
-            if len(exclusive_locks) == 0:
-                # looks like there are no exclusive locks, create our lock.
+            if all(lock.get("maybe_stale") for lock in exclusive_locks):
+                # there are no exclusive locks (or only ones that look stale, but whose staleness
+                # could not be confirmed in the store's clock domain yet, see _is_stale_lock -
+                # creating our lock below gives the next listing a store time reference, so they
+                # get either confirmed (and deleted) or vetoed there). create our lock.
                 key = self._create_lock(exclusive=self.is_exclusive, update_last_refresh=True)
                 # obviously we have a race condition here: other client(s) might have created exclusive
                 # lock(s) at the same time in parallel. thus we have to check again.