diff --git a/docs/faq.rst b/docs/faq.rst index e4fd87e59a..ab15618c77 100644 --- a/docs/faq.rst +++ b/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? -------------------------------------------------- diff --git a/docs/internals/data-structures.rst b/docs/internals/data-structures.rst index a90be6672d..b85f100abd 100644 --- a/docs/internals/data-structures.rst +++ b/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 ------------------ diff --git a/pyproject.toml b/pyproject.toml index 9a83f254ee..7fdfb35047 100644 --- a/pyproject.toml +++ b/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 = [] diff --git a/src/borg/constants.py b/src/borg/constants.py index 6103eb630e..7cbb5b291c 100644 --- a/src/borg/constants.py +++ b/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 diff --git a/src/borg/storelocking.py b/src/borg/storelocking.py index c6b90d89c5..c3fdf5e06f 100644 --- a/src/borg/storelocking.py +++ b/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,10 +140,67 @@ 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 (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 @@ -133,13 +208,44 @@ def _is_stale_lock(self, lock): # 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. diff --git a/src/borg/testsuite/storelocking_test.py b/src/borg/testsuite/storelocking_test.py index cd05af0100..4058f8ced1 100644 --- a/src/borg/testsuite/storelocking_test.py +++ b/src/borg/testsuite/storelocking_test.py @@ -1,3 +1,5 @@ +import datetime +import os import time from pathlib import Path @@ -5,6 +7,8 @@ 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
(simulating a writer whose clock + is skewed against ours). The object's store-side mtime is "now" (a real store write), unless + 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 @@ def test_lock_refresh_stale_removal(self, lockstore): 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): @@ -174,6 +197,142 @@ def load_vanished(name, *args, **kwargs): 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_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())