Merge pull request #10077 from ThomasWaldmann/clock-skew-9870

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

@@ -1170,17 +1170,33 @@ 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-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: 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. to never
+  kill a healthy lock just because its writer's clock is skewed against ours
+  (see :issue:`9870`), a lock is only expired by age if it looks stale both by
+  the clients' clocks (content timestamp) and by the storage's clock
+  (store-side mtime); store-side timestamps can veto an expiry, but never
+  cause one.
+- a warning if the clocks of concurrently active clients differ by more than
+  a few minutes.
+
+See the module docstring of ``src/borg/storelocking.py`` for the details
+(clock domains, how store "now" is derived, what happens without store-side
+mtimes).
 
 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.
@@ -57,9 +57,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]
@@ -155,6 +155,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

@@ -93,6 +93,12 @@ MAX_OBJECT_SIZE = MAX_DATA_SIZE + 41  # see assertion at end of repository modul
 # 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

@@ -1,18 +1,99 @@
+"""
+Repository locking on top of borgstore.
+
+Lock objects
+------------
+Each client holding a lock owns one small object below locks/ in the repository store. Its content
+(JSON) records the lock type (exclusive or shared), the owner's host / process / thread id and a
+timestamp (the "content timestamp"), stamped by the *owner's* clock when the object was written.
+Lock objects are immutable: to refresh a lock, the owner writes a new object and deletes the old one.
+Where the storage backend provides object timestamps (file, sftp, s3, current rest servers - not
+rclone), a lock object additionally carries a store-side mtime, stamped by the *storage's* clock at
+the same write instant; borgstore reports it as ItemInfo.mtime (0 if unavailable).
+
+Acquiring
+---------
+Shared locks may coexist, an exclusive lock must be alone. acquire() lists the lock objects, creates
+its own lock object if nothing forbids it, and lists again to detect a race with other clients
+creating theirs at the same time: an exclusive acquirer backs off if another exclusive lock showed
+up (and otherwise waits for remaining shared locks to go away), a shared acquirer backs off if an
+exclusive lock showed up. This is retried until the timeout.
+
+Staleness
+---------
+A lock whose owner died (crash, power loss, suspended laptop, ...) must not block others forever, so
+every listing judges each lock object and deletes it if it is stale:
+
+- Our own lock object (and, during a refresh, the one we are just replacing) is never stale: we are
+  obviously alive and will refresh or release it.
+- If the owner is a process on this machine and it is dead, the lock is stale. This is local
+  knowledge, independent of any clock, so it is checked first and can not be vetoed by anything the
+  storage says (a storage serving bogus, always-fresh mtimes must not be able to keep an abandoned
+  local lock alive forever).
+- Otherwise, a lock is stale by age if it was not refreshed for longer than the stale timeout
+  (default 30 minutes; owners refresh after half of it). But owners write the content timestamp
+  with *their* clock and we compare it with *ours*: if the owner's clock runs more than the stale
+  timeout behind ours, its just refreshed lock already looks stale to us, and killing it would e.g.
+  enable a compact to delete chunks a running backup still references (see #9870). Thus a lock is
+  only expired by age if it looks stale in BOTH clock domains:
+
+  * writer / local clock domain: our "now" vs. the lock's content timestamp, and
+  * store clock domain: store "now" vs. the lock object's store-side mtime.
+
+  Store "now" is extrapolated from our own lock object: its store-side mtime (harvested from a
+  listing after we created it) plus the time.monotonic() elapsed since its creation (LockAnchor
+  keeps these together, see there). So the store domain comparison involves no client's clock at
+  all, and the storage's absolute clock error cancels out - the storage only serves as a common
+  reference. Its clock should run steadily, though: borg warns if it detects that it jumped between
+  two of its own lock writes (see _check_store_clock_step), and refresh() creates the new lock
+  object before listing, so the stale sweep always judges with a fresh anchor.
+
+  Store-side mtimes are advisory only: they can veto an expiry, but never cause one on their own,
+  so a hostile or broken storage gains no new capabilities (it can not make us kill a healthy lock;
+  blocking us was possible for it before, anyway). A client without an own lock object can not
+  compute store "now" and defers the decision until acquire() has created one - the listing right
+  afterwards then confirms or vetoes. If the backend provides no store-side mtimes (mtime == 0),
+  staleness is judged by the content timestamp alone.
+
+Clock skew warning
+------------------
+As every 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. Every listing compares the other
+writers' offsets with ours and warns (once per Lock instance) if the clocks of concurrently active
+clients differ by more than MAX_MUTUAL_CLOCK_SKEW. This is diagnosis only, never an abort.
+
+Refreshing
+----------
+Lock holders must call refresh() regularly (LockRefresher does that from a background thread). It is
+a no-op until the lock is older than half the stale timeout, then it writes a new lock object and
+deletes the old one. If the old one turns out to be gone, another client killed it as stale (and
+might have acquired its own lock meanwhile), so there is no safe way to continue: LockTimeout.
+"""
+
 import datetime
 import hashlib
 import json
 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 +161,14 @@ class Lock:
         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
+        self.my_old_lock_key = None  # store key of the lock we are replacing while a refresh is in progress
+        # 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.prev_lock_anchor = None  # the anchor before the current one, see _check_store_clock_step()
+        self.skew_warned = False  # emit the clock-skew warning only once per Lock instance
+        self.store_clock_step_warned = False  # emit the storage-clock-step 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 +185,10 @@ class Lock:
     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 +199,11 @@ class Lock:
             # 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.prev_lock_anchor = self.my_lock_anchor
+            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 +217,141 @@ class Lock:
             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_store_clock_step(self, anchor, mtime):
+        """
+        Warn (once) if the storage's clock jumped between the writes of our previous and our current
+        lock object: compare the new object's store-side mtime with what the previous anchor
+        extrapolates for the new object's creation instant. Diagnostic only: a backward step larger
+        than the stale timeout can defeat the store-domain cross-check in _is_stale_lock (see there),
+        this at least names the cause.
+        """
+        prev = self.prev_lock_anchor  # single read - it gets replaced atomically as a whole
+        if prev is None or prev.mtime is None or self.store_clock_step_warned:
+            return
+        elapsed_monotonic = anchor.monotonic - prev.monotonic
+        elapsed_wall = (anchor.dt - prev.dt).total_seconds()
+        if abs(elapsed_wall - elapsed_monotonic) > MAX_MUTUAL_CLOCK_SKEW:
+            # our own two clocks disagree about the elapsed time (suspend: time.monotonic() stood
+            # still, or our wall clock was stepped): then we can not judge the storage's clock.
+            return
+        step = (mtime - prev.mtime) - elapsed_monotonic
+        if abs(step) > MAX_MUTUAL_CLOCK_SKEW:
+            self.store_clock_step_warned = True
+            logger.warning(
+                f"The clock of the repository storage jumped by ~{step:+.0f}s between two lock writes of ours. "
+                f"Storage clock steps interfere with stale lock detection, the storage's clock should run steadily."
+            )
+
+    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 self._is_our_lock(lock):
+                # our own lock object(s): e.g. during a refresh, our old and our new lock object -
+                # a storage clock step between their writes would make them look skewed against
+                # each other, but that is no peer with a skewed clock (see _check_store_clock_step).
+                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.
+        if lock["key"] in (self.my_lock_key, self.my_old_lock_key):
+            # the lock we are currently holding (or the one we are just replacing by it, see
+            # refresh): 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 +371,32 @@ class Lock:
             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:
+                if anchor.mtime is None:
+                    # first harvest for this anchor: cross-check the storage clock's continuity.
+                    self._check_store_clock_step(anchor, mtime)
+                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 +418,11 @@ class Lock:
         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.
@@ -271,29 +504,36 @@ class Lock:
         """Refreshes the lock; call this frequently, but not later than every <stale> seconds."""
         now = datetime.datetime.now(datetime.UTC)
         if self.last_refresh_dt is not None and now > self.last_refresh_dt + self.refresh_td:
-            old_locks = self._find_locks(only_mine=True)
-            if len(old_locks) == 0:
-                # crap, my lock has been removed. :-(
-                # this can happen e.g. if my machine has been suspended while doing a backup, so that the
-                # lock became stale and a borg client on another machine killed it.
-                # if my machine then wakes up again, the lock will have vanished and we get here.
-                # note: if our lock became stale, but is still present (no other client killed it),
-                # we do not get here - we never consider our own lock stale (see _is_stale_lock),
-                # so it is found above and simply refreshed below.
-                # in this case, we need to abort the operation, because the other borg might have removed
-                # repo objects we have written, but the referential tree was not yet full present, e.g.
-                # no archive has been added yet to the manifest, thus all objects looked unused/orphaned.
-                # another scenario when this can happen is a careless user running break-lock on another
-                # machine without making sure there is no borg activity in that repo.
-                logger.debug("LOCK-REFRESH: our lock was killed, there is no safe way to continue.")
-                raise LockTimeout(str(self.store))
-            assert len(old_locks) == 1  # there shouldn't be more than 1
-            old_lock = old_locks[0]
-            if now > old_lock["dt"] + self.refresh_td:
-                logger.debug(f"LOCK-REFRESH: lock needs a refresh. lock: {old_lock}.")
-                new_key = self._create_lock(exclusive=old_lock["exclusive"], update_last_refresh=True)
+            old_key = self.my_lock_key
+            logger.debug(f"LOCK-REFRESH: lock needs a refresh. key: {old_key}.")
+            # create the new lock object BEFORE listing: the listing then harvests a fresh (seconds
+            # old) store-clock anchor before its stale sweep judges other locks with it. listing
+            # first would judge with the anchor of the previous refresh (up to refresh_td old) -
+            # a window in which a storage clock that stepped back meanwhile could make the store-
+            # domain cross-check wrongly confirm a skewed peer's healthy lock as stale, see #9870.
+            new_key = self._create_lock(exclusive=self.is_exclusive, update_last_refresh=True)
+            self.my_old_lock_key = old_key  # exempt our old lock from the stale sweep meanwhile
+            try:
+                locks = self._find_locks(only_mine=True)
+                if old_key not in {lock["key"] for lock in locks}:
+                    # crap, my lock has been removed. :-(
+                    # this can happen e.g. if my machine has been suspended while doing a backup, so that the
+                    # lock became stale and a borg client on another machine killed it.
+                    # if my machine then wakes up again, the lock will have vanished and we get here.
+                    # note: if our lock became stale, but is still present (no other client killed it),
+                    # we do not get here - we never consider our own lock stale (see _is_stale_lock),
+                    # so it is found above and simply replaced below.
+                    # in this case, we need to abort the operation, because the other borg might have removed
+                    # repo objects we have written, but the referential tree was not yet full present, e.g.
+                    # no archive has been added yet to the manifest, thus all objects looked unused/orphaned.
+                    # another scenario when this can happen is a careless user running break-lock on another
+                    # machine without making sure there is no borg activity in that repo.
+                    # clean up the new lock (so it does not needlessly block others until it expires).
+                    logger.debug("LOCK-REFRESH: our lock was killed, there is no safe way to continue.")
+                    self._delete_lock(new_key, ignore_not_found=True, update_last_refresh=True)
+                    raise LockTimeout(str(self.store))
                 try:
-                    self._delete_lock(old_lock["key"], update_last_refresh=False)
+                    self._delete_lock(old_key, update_last_refresh=False)
                 except ObjectNotFound:
                     # our old lock vanished between listing and deleting it: another client considered
                     # it stale, killed it and (not having seen our new lock in its recheck) might have
@@ -302,6 +542,8 @@ class Lock:
                     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))
+            finally:
+                self.my_old_lock_key = None
 
 
 class LockRefresher:

src/borg/testsuite/storelocking_test.py

@@ -1,3 +1,5 @@
+import datetime
+import os
 import time
 from pathlib import Path
 
@@ -5,6 +7,8 @@ import pytest
 
 from borgstore.store import ObjectNotFound, Store
 
+from .fslocking_test import free_pid  # NOQA
+from ..platform import get_process_id
 from ..storelocking import Lock, NotLocked, LockTimeout
 
 ID1 = "foo", 1, 1
@@ -20,6 +24,19 @@ def lockstore(tmp_path):
     store.destroy()
 
 
+def write_raw_lock(store, id, *, exclusive, dt, mtime=None):
+    """
+    Write a lock object with an arbitrary content timestamp <dt> (simulating a writer whose clock
+    is skewed against ours). The object's store-side mtime is "now" (a real store write), unless
+    <mtime> is given (then the file's mtime is set to it).
+    """
+    key = Lock(store, id=id)._create_lock(exclusive=exclusive, dt=dt)
+    if mtime is not None:
+        path = store.backend.base_path / "locks" / key  # posixfs, locks/ nesting levels [0]
+        os.utime(path, (mtime, mtime))
+    return key
+
+
 class TestLock:
     def test_cm(self, lockstore):
         with Lock(lockstore, exclusive=True, id=ID1) as lock:
@@ -90,14 +107,20 @@ class TestLock:
         lock_keys_b00 = set(lock._get_locks())
         time.sleep(2.1)
         # now the lock is stale. we never consider the lock we hold ourselves stale,
-        # but another client (== another Lock instance) does:
+        # but another client (== another Lock instance) does. a client without a lock object
+        # of its own can not confirm staleness in the store's clock domain yet (see #9870),
+        # so a plain listing defers the kill:
         other_lock = Lock(lockstore, exclusive=True, id=ID2, stale=2)
-        lock_keys_b21 = set(other_lock._get_locks())  # now the lock should be stale & gone.
+        lock_keys_b21 = set(other_lock._get_locks())
         assert lock_keys_a00 == lock_keys_a05  # was too young, no refresh done
         assert len(lock_keys_a00) == 1
         assert lock_keys_a00 != lock_keys_b00  # refresh done, new lock has different key
         assert len(lock_keys_b00) == 1
-        assert len(lock_keys_b21) == 0  # stale lock was ignored
+        assert lock_keys_b21 == lock_keys_b00  # stale, but kill deferred (no store time reference yet)
+        # acquire() creates other_lock's own lock object first, then confirms the staleness
+        # in the store's clock domain and kills the stale lock:
+        other_lock.acquire()
+        other_lock.release()
         assert len(list(lock.store.list("locks"))) == 0  # stale lock was removed from store
 
     def test_release_stale_lock(self, lockstore):
@@ -121,6 +144,7 @@ class TestLock:
         new_keys = set(lock._get_locks())
         assert len(old_keys) == len(new_keys) == 1
         assert old_keys != new_keys  # refresh done, new lock has different key
+        assert not lock.store_clock_step_warned  # the storage clock ran steadily, no warning
         lock.release()
 
     def test_refresh_killed_lock(self, lockstore):
@@ -174,6 +198,191 @@ class TestLock:
         lock.acquire()  # the vanished exclusive lock must not block us
         lock.release()
 
+    def test_skewed_writer_healthy_lock_not_killed(self, lockstore):
+        # a lock whose content timestamp looks stale (its writer's clock runs >stale behind ours),
+        # but whose store-side mtime shows it was written just now: it must NOT be killed - killing
+        # a healthy lock enables compact to delete chunks a running backup references, see #9870.
+        dt = datetime.datetime.now(datetime.UTC) - datetime.timedelta(minutes=40)
+        foreign_key = write_raw_lock(lockstore, ID1, exclusive=False, dt=dt)  # store mtime: now
+        # a shared lock can coexist with it - and must warn about the skew:
+        lock = Lock(lockstore, exclusive=False, id=ID2)
+        lock.acquire()
+        assert lock.skew_warned
+        assert foreign_key in lock._get_locks()  # healthy foreign lock survived
+        lock.release()
+        # an exclusive lock must NOT be obtainable by killing the healthy shared lock:
+        with pytest.raises(LockTimeout):
+            Lock(lockstore, exclusive=True, id=ID2).acquire()
+        assert f"locks/{foreign_key}" in [f"locks/{k}" for k in Lock(lockstore, id=ID2)._get_locks()]
+
+    def test_stale_lock_killed_when_both_clock_domains_agree(self, lockstore):
+        # a lock that is stale in both clock domains (old content timestamp AND old store-side
+        # mtime) is really stale and must be killed during acquire().
+        dt = datetime.datetime.now(datetime.UTC) - datetime.timedelta(minutes=40)
+        foreign_key = write_raw_lock(lockstore, ID1, exclusive=True, dt=dt, mtime=dt.timestamp())
+        lock = Lock(lockstore, exclusive=True, id=ID2)
+        lock.acquire()  # must succeed: the stale exclusive lock gets confirmed stale and killed
+        assert not lock.skew_warned
+        locks = lock._get_locks()
+        assert foreign_key not in locks
+        assert lock.my_lock_key in locks
+        lock.release()
+
+    def test_skew_warning_below_stale_threshold(self, lockstore):
+        # a live lock whose writer's clock runs 10 minutes ahead of ours: far from the stale
+        # threshold, but still worth a warning (e.g. concurrent manifest writes could produce
+        # a spurious RepositoryReplay later), see #9870.
+        dt = datetime.datetime.now(datetime.UTC) + datetime.timedelta(minutes=10)
+        write_raw_lock(lockstore, ID1, exclusive=False, dt=dt)  # store mtime: now
+        lock = Lock(lockstore, exclusive=False, id=ID2)
+        lock.acquire()
+        assert lock.skew_warned
+        lock.release()
+
+    def test_no_skew_warning_for_small_offsets(self, lockstore):
+        # small clock differences (well below MAX_MUTUAL_CLOCK_SKEW) must not warn.
+        dt = datetime.datetime.now(datetime.UTC) + datetime.timedelta(seconds=60)
+        write_raw_lock(lockstore, ID1, exclusive=False, dt=dt)  # store mtime: now
+        lock = Lock(lockstore, exclusive=False, id=ID2)
+        lock.acquire()
+        assert not lock.skew_warned
+        lock.release()
+
+    def test_no_lock_churn_when_blocked_by_skewed_lock(self, lockstore, monkeypatch):
+        # a healthy exclusive lock of a writer whose clock runs >stale behind ours blocks us
+        # (correctly so - we must never kill it). the store-clock anchor survives the deletion
+        # of our transient lock object, so only the first acquire iteration needs to create
+        # one: later iterations veto the kill on their first listing instead of repeating the
+        # whole defer/create/veto/delete cycle, see #9870.
+        dt = datetime.datetime.now(datetime.UTC) - datetime.timedelta(minutes=40)
+        foreign_key = write_raw_lock(lockstore, ID1, exclusive=True, dt=dt)  # store mtime: now
+
+        lock_writes = []
+        orig_store = lockstore.store
+
+        def counting_store(name, *args, **kwargs):
+            lock_writes.append(name)
+            return orig_store(name, *args, **kwargs)
+
+        monkeypatch.setattr(lockstore, "store", counting_store)
+        lock = Lock(lockstore, exclusive=False, id=ID2)
+        lock.retry_delay_min = lock.retry_delay_max = 0.1  # several retry iterations within timeout
+        with pytest.raises(LockTimeout):
+            lock.acquire()  # the healthy exclusive lock does not go away
+        assert len(lock_writes) == 1  # only the first iteration created a transient lock
+        assert foreign_key in Lock(lockstore, id=ID2)._get_locks()  # the blocker survived
+
+    def test_no_skew_warning_when_only_our_store_time_lags(self, lockstore):
+        # while we are suspended, time.monotonic() stands still, so our store "now" estimate
+        # lags afterwards. a foreign lock that went genuinely stale meanwhile then gets vetoed
+        # (the safe direction), but that veto is no evidence of clock skew and must not
+        # produce a bogus "clock skew of ~0s" warning, see #9870.
+        lock = Lock(lockstore, exclusive=False, id=ID2)
+        lock.acquire()
+        anchor = lock.my_lock_anchor
+        assert anchor.mtime is not None  # store time reference was harvested
+        # simulate a 45 minute suspend after the anchor was set (time.monotonic() stood still):
+        lock.my_lock_anchor = anchor._replace(monotonic=anchor.monotonic + 45 * 60)
+        dt = datetime.datetime.now(datetime.UTC) - datetime.timedelta(minutes=40)
+        foreign_key = write_raw_lock(lockstore, ID1, exclusive=False, dt=dt, mtime=dt.timestamp())
+        locks = lock._get_locks()
+        assert foreign_key in locks  # genuinely stale, but vetoed: store "now" lags behind
+        assert not lock.skew_warned  # the writer's clock is not skewed - no warning
+        lock.release()
+
+    def test_refresh_judges_with_fresh_store_time(self, lockstore):
+        # refresh() creates the new lock object BEFORE listing, so the listing's stale sweep judges
+        # other locks with a seconds-old store-clock anchor, not with the one of the previous
+        # refresh (up to 15min old). if the storage's clock stepped back meanwhile (simulated by
+        # an anchor that extrapolates 1h ahead of the storage's clock), that old anchor would
+        # wrongly confirm a skewed peer's healthy lock as stale in both clock domains, see #9870.
+        lock = Lock(lockstore, exclusive=False, id=ID2)
+        lock.acquire()
+        anchor = lock.my_lock_anchor
+        assert anchor.mtime is not None
+        lock.my_lock_anchor = anchor._replace(monotonic=anchor.monotonic - 3600)  # storage clock stepped back 1h
+        # a healthy peer whose clock runs 40min behind ours (content looks stale, store mtime: now):
+        dt = datetime.datetime.now(datetime.UTC) - datetime.timedelta(minutes=40)
+        foreign_key = write_raw_lock(lockstore, ID1, exclusive=False, dt=dt)
+        lock.last_refresh_dt -= lock.refresh_td + datetime.timedelta(seconds=1)  # make refresh() act now
+        lock.refresh()
+        assert foreign_key in lock._get_locks()  # survived: judged with the fresh anchor
+        lock.release()
+
+    def test_storage_clock_step_warning(self, lockstore):
+        # when the mtime of a freshly created lock object is harvested, it is compared with what the
+        # previous anchor extrapolates for that instant: a mismatch means the storage's clock jumped
+        # between the two writes (simulated here by a previous anchor whose mtime is 1h ahead), see
+        # #9870. diagnostic only, but it names the cause of a defeated stale-lock cross-check.
+        lock = Lock(lockstore, exclusive=False, id=ID2)
+        lock.acquire()
+        anchor = lock.my_lock_anchor
+        assert anchor.mtime is not None
+        lock.my_lock_anchor = anchor._replace(mtime=anchor.mtime + 3600)  # as if the storage clock stepped back 1h
+        lock.last_refresh_dt -= lock.refresh_td + datetime.timedelta(seconds=1)  # make refresh() act now
+        lock.refresh()  # creates a new lock object and harvests its mtime: the step is detected
+        assert lock.store_clock_step_warned
+        lock.release()
+
+    def test_no_skew_warning_about_our_own_old_lock(self, lockstore):
+        # during a refresh, the listing contains our old and our new lock object. a storage clock
+        # step between their writes (simulated by bumping the old object's mtime by 1h) makes the
+        # old one look skewed against our new anchor - but that is not a peer with a skewed clock,
+        # so there must be no clock skew warning (about ourselves!), see #9870.
+        lock = Lock(lockstore, exclusive=False, id=ID2)
+        lock.acquire()
+        old_path = lockstore.backend.base_path / "locks" / lock.my_lock_key  # posixfs, levels [0]
+        mtime = os.stat(old_path).st_mtime + 3600
+        os.utime(old_path, (mtime, mtime))
+        lock.last_refresh_dt -= lock.refresh_td + datetime.timedelta(seconds=1)  # make refresh() act now
+        lock.refresh()
+        assert not lock.skew_warned
+        lock.release()
+
+    def test_skew_warning_during_exclusive_acquire(self, lockstore):
+        # an exclusive acquirer must warn about a skewed peer it sees while (unsuccessfully)
+        # waiting for the peer's healthy shared lock to go away: the listing that would satisfy
+        # the exclusive acquire can only contain our own lock, so the skewed peer is only
+        # visible in the intermediate listings, see #9870.
+        dt = datetime.datetime.now(datetime.UTC) + datetime.timedelta(minutes=10)
+        write_raw_lock(lockstore, ID1, exclusive=False, dt=dt)  # store mtime: now
+        lock = Lock(lockstore, exclusive=True, id=ID2)
+        with pytest.raises(LockTimeout):
+            lock.acquire()  # the healthy shared lock does not go away
+        assert lock.skew_warned
+
+    def test_dead_process_lock_killed_despite_fresh_store_mtime(self, lockstore, free_pid):
+        # knowing locally that the lock-owning process (on THIS machine) is dead must not be
+        # vetoable by store-side timestamps: a store serving bogus, always-fresh mtimes could
+        # otherwise keep an abandoned lock alive forever and block us, see #9870.
+        host, _, tid = get_process_id()
+        dead_id = (host, free_pid, tid)
+        dt = datetime.datetime.now(datetime.UTC) - datetime.timedelta(minutes=40)
+        dead_key = write_raw_lock(lockstore, dead_id, exclusive=True, dt=dt)  # store mtime: now (fresh)
+        lock = Lock(lockstore, exclusive=True, id=ID2)
+        # killed on a plain listing already - no store time reference of our own needed:
+        assert dead_key not in lock._get_locks()
+        lock.acquire()  # the exclusive lock must be obtainable
+        lock.release()
+
+    def test_stale_kill_legacy_behavior_without_mtime(self, lockstore, monkeypatch):
+        # if the backend can not provide store-side mtimes (e.g. rclone, mtime == 0), staleness
+        # is judged by the content timestamp alone, like before #9870 - even on a plain listing.
+        dt = datetime.datetime.now(datetime.UTC) - datetime.timedelta(minutes=40)
+        foreign_key = write_raw_lock(lockstore, ID1, exclusive=True, dt=dt)
+
+        orig_list = lockstore.list
+
+        def list_no_mtime(name, *args, **kwargs):
+            for info in orig_list(name, *args, **kwargs):
+                yield info._replace(mtime=0)
+
+        monkeypatch.setattr(lockstore, "list", list_no_mtime)
+        lock = Lock(lockstore, exclusive=True, id=ID2)
+        locks = lock._get_locks()  # legacy: killed right away, no store-domain cross-check possible
+        assert foreign_key not in locks
+        assert len(list(orig_list("locks"))) == 0
+
     def test_migrate_lock(self, lockstore):
         old_id, new_id = ID1, ID2
         assert old_id[1] != new_id[1]  # different PIDs (like when doing daemonize())