Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/filelock/_lease.py: 37%

1from __future__ import annotations 

2 

3import os 

4import secrets 

5import time 

6from contextlib import suppress 

7from dataclasses import dataclass 

8from math import isfinite 

9from threading import Event, Thread, current_thread, local 

10from typing import TYPE_CHECKING, Literal 

11 

12from ._error import LeaseSettingsMismatch 

13from ._identity import owner_is_stale 

14from ._marker import MarkerSoftFileLock, OwnerMode, OwnerRecord, parse_marker 

15from ._soft import _read_lock_file 

16from ._util import break_lock_file, touch 

17 

18if TYPE_CHECKING: 

19 import sys 

20 from collections.abc import Callable 

21 

22 from ._api import LockOptions 

23 

24 if sys.version_info >= (3, 11): # pragma: no cover (py311+) 

25 from typing import Unpack 

26 else: # pragma: no cover (<py311) 

27 from typing_extensions import Unpack 

28 

29CompromiseReason = Literal["marker-missing", "owner-changed", "refresh-failed"] 

30 

31_RefreshOutcome = Literal["ok", "lost", "transient"] 

32 

33 

34@dataclass(frozen=True) 

35class LeaseCompromise: 

36 """Why a held lease stopped being this process's to hold.""" 

37 

38 lock_file: str 

39 token: str 

40 reason: CompromiseReason 

41 error: OSError | None = None 

42 

43 

44@dataclass(frozen=True) 

45class _Heartbeat: 

46 """A running heartbeat and the event that stops it, which only ever exist together.""" 

47 

48 thread: Thread 

49 stop: Event 

50 

51 

52@dataclass 

53class _LeaseClaim: 

54 """The state of one claim: its token, its heartbeat, and how that claim was lost.""" 

55 

56 token: str | None = None 

57 compromise: LeaseCompromise | None = None 

58 heartbeat: _Heartbeat | None = None 

59 

60 

61class _LeaseClaimHolder: 

62 """Holds the claim its owning context acquired.""" 

63 

64 # Only the holder is thread-local, mirroring FileLockContext: the claim stays an ordinary object, so the heartbeat 

65 # thread records a compromise where the thread that acquired the lease reads it. 

66 

67 def __init__(self) -> None: 

68 self.claim = _LeaseClaim() 

69 

70 

71class _ThreadLocalLeaseClaimHolder(_LeaseClaimHolder, local): 

72 """A thread local version of the ``_LeaseClaimHolder`` class.""" 

73 

74 

75class SoftFileLease(MarkerSoftFileLock): 

76 """ 

77 Existence lock whose claim expires, so a peer may take it while the previous holder still runs. 

78 

79 A lease trades mutual exclusion for progress. The holder publishes a claim and refreshes it every 

80 ``heartbeat_interval`` seconds; a contender takes the marker once it is ``lease_duration`` seconds stale. Nothing 

81 stops the expired holder: it keeps running, and it keeps using whatever the lock protects. Treat the lease as a hint 

82 about who *should* be working, not as a guarantee that only one worker is. 

83 

84 To make a protected resource reject a superseded holder, that resource must be linearizable and must fence on a 

85 monotonic generation it controls. :attr:`token` names a claim; it does not fence one. Where overlap is unacceptable, 

86 use :class:`StrictSoftFileLock <filelock.StrictSoftFileLock>` instead. 

87 

88 Every contender for a path must agree on ``lease_duration``. A contender that finds a claim published under a 

89 different duration raises :class:`LeaseSettingsMismatch <filelock.LeaseSettingsMismatch>` rather than apply its own 

90 expiry to a peer that never agreed to it. 

91 

92 Expiry reclaims less on Windows, which refuses to rename or delete a file another process holds open. A peer there 

93 takes an expired claim only once the previous holder's process exits and its handle closes; a holder that lives on 

94 but stops refreshing keeps the marker. Unix reclaims the marker either way. 

95 

96 ``on_compromise`` fires from the heartbeat thread when a refresh fails, or when the marker vanishes or names another 

97 owner. The holder should stop touching the protected resource when it runs. Because it runs on that thread, a 

98 ``release()`` inside it only takes effect when the lease was built with ``thread_local=False``; the default 

99 thread-local context hides the claim from every thread but the one that acquired it, so the release does nothing. 

100 Signal the acquiring thread instead when the context stays thread-local. 

101 

102 .. versionadded:: 3.30.0 

103 

104 """ 

105 

106 _owner_mode: OwnerMode = "lease" 

107 

108 #: lease_duration replaces the legacy age-based lifetime, so accepting both would give one lock two expiry clocks. 

109 _lifetime_supported: bool = False 

110 _lifetime_unsupported_reason: str = "lease_duration sets when a lease expires" 

111 

112 def __init__( 

113 self, 

114 lock_file: str | os.PathLike[str], 

115 *, 

116 lease_duration: float = 30.0, 

117 heartbeat_interval: float | None = None, 

118 on_compromise: Callable[[LeaseCompromise], None] | None = None, 

119 **kwargs: Unpack[LockOptions], 

120 ) -> None: 

121 """ 

122 Create a lease. 

123 

124 :param lease_duration: seconds of marker staleness after which a contender may take the claim. Every contender 

125 for the path must pass the same value. 

126 :param heartbeat_interval: seconds between refreshes. Defaults to a third of ``lease_duration``, leaving room 

127 for two missed refreshes before a peer may take the claim. Must be shorter than ``lease_duration``. 

128 :param on_compromise: called from the heartbeat thread with a :class:`LeaseCompromise` when the claim is lost. 

129 :param kwargs: every other :class:`BaseFileLock <filelock.BaseFileLock>` option, ``timeout`` and ``mode`` among 

130 them. The metaclass passes them all by keyword, and taking them here lets 

131 :class:`AsyncSoftFileLease <filelock.AsyncSoftFileLease>` add the async plumbing a fixed signature would 

132 hide. 

133 

134 """ 

135 if isinstance(lease_duration, bool) or not isinstance(lease_duration, (int, float)): 

136 msg = f"lease_duration must be a finite positive number, not {type(lease_duration).__name__}" 

137 raise TypeError(msg) 

138 if not isfinite(lease_duration) or lease_duration <= 0: 

139 msg = f"lease_duration must be positive and finite, got {lease_duration!r}" 

140 raise ValueError(msg) 

141 if heartbeat_interval is None: 

142 heartbeat_interval = lease_duration / 3 

143 if not 0 < heartbeat_interval < lease_duration: 

144 msg = f"heartbeat_interval must be positive and below lease_duration, got {heartbeat_interval!r}" 

145 raise ValueError(msg) 

146 super().__init__(lock_file, **kwargs) 

147 self._lease_duration = lease_duration 

148 self._heartbeat_interval = heartbeat_interval 

149 self._on_compromise = on_compromise 

150 # Sharing one claim across a thread-local lock lets a second thread's failed acquisition stop the heartbeat of 

151 # the thread holding the lease, leaving its marker unrefreshed until a peer reclaims it. 

152 self._claims: _LeaseClaimHolder = ( 

153 _ThreadLocalLeaseClaimHolder if self.is_thread_local() else _LeaseClaimHolder 

154 )() 

155 

156 @property 

157 def _claim(self) -> _LeaseClaim: 

158 return self._claims.claim 

159 

160 @property 

161 def lease_duration(self) -> float: 

162 """The staleness in seconds after which a contender may take this claim.""" 

163 return self._lease_duration 

164 

165 @property 

166 def token(self) -> str | None: 

167 """ 

168 The token naming the claim this process published. 

169 

170 :returns: the token while the lease is held, ``None`` otherwise. It identifies a claim; it does not fence one. 

171 

172 """ 

173 return self._claim.token 

174 

175 @property 

176 def compromise(self) -> LeaseCompromise | None: 

177 """ 

178 The loss of claim the heartbeat observed. 

179 

180 :returns: the :class:`LeaseCompromise`, or ``None`` while the claim still holds 

181 

182 """ 

183 return self._claim.compromise 

184 

185 def _acquire(self) -> None: 

186 claim = self._claim 

187 self._stop_heartbeat() # no earlier claim's heartbeat outlives the acquisition of the next one 

188 # The published record reads the token, so it exists before the marker is written and goes back on failure. 

189 claim.token = token = secrets.token_hex(16) 

190 claim.compromise = None 

191 try: 

192 super()._acquire() 

193 except BaseException: 

194 claim.token = None 

195 raise 

196 # The context is thread-local by default, so the heartbeat thread cannot read the descriptor this one just 

197 # published, nor the claim this one owns. Hand it the fd, the inode it verified and the claim instead. 

198 if (fd := self._context.lock_file_fd) is not None and ( 

199 identity := self._context.lock_file_fd_identity 

200 ) is not None: 

201 self._start_heartbeat(claim, fd, identity, token) 

202 else: 

203 claim.token = None 

204 

205 def _release(self) -> None: 

206 self._stop_heartbeat() 

207 self._claim.token = None 

208 super()._release() 

209 

210 def _published_record(self) -> OwnerRecord: 

211 return super()._published_record()._replace(token=self._claim.token, lease_duration=self._lease_duration) 

212 

213 def _try_break_stale_lock(self) -> None: 

214 if (peer := self._read_peer()) is None: 

215 # Not a readable protocol 2 lease record: a partial write, a foreign or legacy protocol 1 marker, or the 

216 # strict sentinel. The base self-heal evicts a genuinely malformed marker once it ages past the grace 

217 # window and leaves a legitimate legacy or strict holder in place, so a corrupt marker no longer wedges 

218 # every lease contender until its own timeout. 

219 super()._try_break_stale_lock() 

220 return 

221 owner, mtime, ino = peer 

222 # Only a peer that published a lease agreed to be superseded by one, so a record stating any other contract is 

223 # never reclaimed by age. Raise the mismatch outside the read so the suppression cannot swallow it. 

224 if owner.mode != "lease": 

225 return 

226 if owner.lease_duration != self._lease_duration: 

227 msg = ( 

228 f"{self.lock_file} holds a lease of {owner.lease_duration!r}s but this contender configured " 

229 f"{self._lease_duration!r}s; every contender for a path must agree on lease_duration" 

230 ) 

231 raise LeaseSettingsMismatch(msg) 

232 # A break can fail for reasons a contender must ride out rather than raise on: a peer broke the marker first, 

233 # or Windows refuses to rename a file whose holder still has it open. Poll again instead. 

234 with suppress(OSError): 

235 # A dead or recycled owner is reclaimed at once; a live owner past its lease duration is superseded on the 

236 # schedule every contender agreed to. 

237 if owner_is_stale(owner.pid, owner.hostname, owner.start): 

238 break_lock_file(self.lock_file, mtime, ino) 

239 return 

240 if time.time() - mtime >= self._lease_duration: 

241 break_lock_file(self.lock_file, mtime, ino) 

242 

243 def _read_peer(self) -> tuple[OwnerRecord, float, int] | None: 

244 with suppress(OSError, ValueError): 

245 content, mtime, ino = _read_lock_file(self.lock_file) 

246 if (owner := parse_marker(content)) is not None: 

247 return owner, mtime, ino 

248 return None 

249 

250 def _start_heartbeat(self, claim: _LeaseClaim, fd: int, identity: tuple[int, int], token: str) -> None: 

251 # The thread watches the event it was handed rather than whatever the claim names later: a heartbeat that 

252 # outlives its join timeout would otherwise adopt the next acquisition's event and never stop. 

253 stop = Event() 

254 thread = Thread( 

255 target=self._refresh_until_stopped, 

256 args=(claim, fd, identity, token, stop), 

257 name=f"filelock-lease-{os.getpid()}", 

258 daemon=True, 

259 ) 

260 # Record the heartbeat before starting the thread so a release racing this acquire on a shared, 

261 # non-thread-local claim always sees it and sets the stop event; the thread then exits at its first wait 

262 # instead of outliving the release. A start that raises leaves the unstarted thread for _stop_heartbeat. 

263 claim.heartbeat = _Heartbeat(thread, stop) 

264 thread.start() 

265 

266 def _stop_heartbeat(self) -> None: 

267 claim = self._claim 

268 if (heartbeat := claim.heartbeat) is None: 

269 return 

270 heartbeat.stop.set() 

271 claim.heartbeat = None 

272 # thread.ident is None until start() runs: a heartbeat recorded before its thread started (a start that 

273 # raised, or a release racing acquire on a shared claim) has nothing to join, and the stop above makes it 

274 # exit at once. on_compromise runs on the heartbeat thread and may release the lease, landing back here. 

275 if heartbeat.thread.ident is not None and heartbeat.thread is not current_thread(): 

276 heartbeat.thread.join(timeout=self._heartbeat_interval) 

277 

278 def _refresh_until_stopped( 

279 self, 

280 claim: _LeaseClaim, 

281 fd: int, 

282 identity: tuple[int, int], 

283 token: str, 

284 stop: Event, 

285 ) -> None: 

286 # The loop ends at the first loss of the claim, so the holder hears about it once. A transient filesystem 

287 # error (ESTALE / EIO on the NFS-style filesystems a lease targets) is not a loss: retry rather than raise a 

288 # false compromise. Report the claim unrefreshable only once failures have run long enough that a contender 

289 # could take it before the next success would land, a margin before the marker actually ages out, the way 

290 # restic declares a lock unrefreshable ahead of its stale time. 

291 last_success = time.monotonic() 

292 while not stop.wait(self._heartbeat_interval): 

293 outcome, error = self._refresh_claim(claim, fd, identity, token) 

294 if outcome == "lost": 

295 return 

296 if outcome == "ok": 

297 last_success = time.monotonic() 

298 elif time.monotonic() - last_success >= self._lease_duration - self._heartbeat_interval: 

299 self._report_compromise(claim, "refresh-failed", error, token) 

300 return 

301 

302 def _refresh_claim( 

303 self, 

304 claim: _LeaseClaim, 

305 fd: int, 

306 identity: tuple[int, int], 

307 token: str, 

308 ) -> tuple[_RefreshOutcome, OSError | None]: 

309 try: 

310 st = os.lstat(self.lock_file) 

311 except FileNotFoundError as error: 

312 self._report_compromise(claim, "marker-missing", error, token) 

313 return "lost", None 

314 except OSError as error: 

315 return "transient", error 

316 # A peer that took the expired claim replaced the marker, so the pathname now names its inode, not ours. 

317 if (st.st_dev, st.st_ino) != identity: 

318 self._report_compromise(claim, "owner-changed", None, token) 

319 return "lost", None 

320 try: 

321 touch(self.lock_file, fd=fd) 

322 except OSError as error: 

323 return "transient", error 

324 return "ok", None 

325 

326 def _report_compromise( 

327 self, 

328 claim: _LeaseClaim, 

329 reason: CompromiseReason, 

330 error: OSError | None, 

331 token: str, 

332 ) -> None: 

333 # Record it on the claim this heartbeat serves, not on self._claim: a thread-local claim read from the 

334 # heartbeat thread is a different, empty one, so the holder would never see the loss it is being told about. 

335 # The token is the one this thread published, not claim.token, which a release may already have cleared. 

336 claim.compromise = LeaseCompromise(lock_file=self.lock_file, token=token, reason=reason, error=error) 

337 if self._on_compromise is not None: 

338 self._on_compromise(claim.compromise) 

339 

340 

341__all__ = [ 

342 "CompromiseReason", 

343 "LeaseCompromise", 

344 "SoftFileLease", 

345]