| 1 | from __future__ import annotations |
| 2 | |
| 3 | import subprocess |
| 4 | import threading |
| 5 | import time |
| 6 | from typing import Callable |
| 7 | |
| 8 | |
| 9 | APT_LOCK_TIMEOUT_SECONDS = 240 |
| 10 | APT_LOCK_RETRY_SECONDS = 5 |
| 11 | |
| 12 | _apt_lock = threading.RLock() |
| 13 | |
| 14 | |
| 15 | def run_apt_with_retries( |
| 16 | runner: Callable[[], subprocess.CompletedProcess[str]], |
| 17 | *, |
| 18 | lock_timeout_seconds: int = APT_LOCK_TIMEOUT_SECONDS, |
| 19 | retry_seconds: int = APT_LOCK_RETRY_SECONDS, |
| 20 | ) -> subprocess.CompletedProcess[str]: |
| 21 | """Run an apt/dpkg command, serializing in-process callers and waiting out apt locks.""" |
| 22 | |
| 23 | with _apt_lock: |
| 24 | deadline = time.monotonic() + max(0, lock_timeout_seconds) |
| 25 | while True: |
| 26 | result = runner() |
| 27 | if result.returncode == 0 or not is_apt_lock_error(result): |
| 28 | return result |
| 29 | remaining = deadline - time.monotonic() |
| 30 | if remaining <= 0: |
| 31 | return result |
| 32 | time.sleep(min(max(1, retry_seconds), remaining)) |
| 33 | |
| 34 | |
| 35 | def is_apt_lock_error(result: subprocess.CompletedProcess[str]) -> bool: |
| 36 | output = f"{result.stderr or ''}\n{result.stdout or ''}".lower() |
| 37 | return ( |
| 38 | "could not get lock" in output |
| 39 | or "unable to lock directory" in output |
| 40 | or "unable to acquire the dpkg frontend lock" in output |
| 41 | or "is another process using it" in output |
| 42 | ) |