mirror of
https://github.com/kevinveenbirkenbach/computer-playbook.git
synced 2026-11-06 02:54:26 +00:00
kevinveenbirkenbachandClaude Opus 5
56036be2a1
A single anonymous GET for the pinned hadolint binary completed its TLS
handshake and its request, then got a clean FIN with no status line after
3.6s. download_file was a bare urlopen with no retry, and make install-lint
is a prerequisite of autoformat, lint and every lint-* target -- so one shed
socket took down the whole quality gate. The same asset downloaded fine five
times elsewhere in that run, 15s after the failure on another distro.
Retry four times with exponential backoff, catching (OSError, HTTPException).
Two names, not the obvious four: ssl.SSLError is an OSError but NOT a
URLError, so the narrow tuple would have missed it, while OSError subsumes
URLError, ConnectionError, TimeoutError and gaierror in one token.
HTTPException is needed separately because IncompleteRead is not an OSError.
HTTPError is caught FIRST and re-raised immediately for any status outside
{408,429,500,502,503,504}. That ordering is load-bearing -- HTTPError is
itself an OSError, so a broader catch first would swallow it -- and it is
what keeps this robustness rather than masking: a 404 is exactly what a
stale version pin looks like after roles/sys-lint is bumped, and it still
fails on the first attempt with the URL. The final attempt re-raises the
original with its traceback, and write_bytes stays in the else branch so a
full disk is never retried.
apt was the other uncached, unretried input in the same target (its default
Acquire::Retries is 0), so give both invocations -o Acquire::Retries=3,
matching what eight Dockerfiles in this repo already do.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
91 lines
2.5 KiB
Python
91 lines
2.5 KiB
Python
"""Shared low-level install helpers."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from http.client import HTTPException
|
|
from pathlib import Path
|
|
from typing import TYPE_CHECKING
|
|
from urllib.error import HTTPError
|
|
from urllib.request import urlopen
|
|
|
|
if TYPE_CHECKING:
|
|
from collections.abc import Sequence
|
|
|
|
|
|
def log(msg: str) -> None:
|
|
print(msg, flush=True)
|
|
|
|
|
|
def warn(msg: str) -> None:
|
|
print(msg, file=sys.stderr, flush=True)
|
|
|
|
|
|
def which(cmd: str) -> bool:
|
|
return shutil.which(cmd) is not None
|
|
|
|
|
|
def run_privileged(cmd: Sequence[str]) -> None:
|
|
argv: list[str] = list(cmd)
|
|
if os.geteuid() != 0 and shutil.which("sudo") is not None:
|
|
argv = ["sudo", *argv]
|
|
subprocess.run(argv, check=True)
|
|
|
|
|
|
_DOWNLOAD_ATTEMPTS = 4
|
|
_RETRYABLE_STATUS = frozenset({408, 429, 500, 502, 503, 504})
|
|
|
|
|
|
def ensure_dir_on_path(directory: str) -> None:
|
|
if not directory:
|
|
return
|
|
current = os.environ.get("PATH", "")
|
|
parts = current.split(os.pathsep) if current else []
|
|
if directory in parts:
|
|
return
|
|
os.environ["PATH"] = directory + (os.pathsep + current if current else "")
|
|
|
|
|
|
def download_file(url: str, output: str, *, timeout: float = 60.0) -> None:
|
|
"""Fetch *url* into *output*, retrying only transient transport failures.
|
|
|
|
Args:
|
|
url: source to read.
|
|
output: path the body is written to.
|
|
timeout: per-attempt socket timeout in seconds.
|
|
|
|
Raises:
|
|
HTTPError: immediately for any status outside _RETRYABLE_STATUS, so a
|
|
stale version pin fails on the first attempt instead of after the
|
|
backoff.
|
|
"""
|
|
for attempt in range(1, _DOWNLOAD_ATTEMPTS + 1):
|
|
try:
|
|
with urlopen(url, timeout=timeout) as response: # noqa: S310 - trusted release URLs only
|
|
data = response.read()
|
|
except HTTPError as err:
|
|
if err.code not in _RETRYABLE_STATUS or attempt == _DOWNLOAD_ATTEMPTS:
|
|
raise
|
|
reason = f"HTTP {err.code}"
|
|
except (OSError, HTTPException) as err:
|
|
if attempt == _DOWNLOAD_ATTEMPTS:
|
|
raise
|
|
reason = type(err).__name__
|
|
else:
|
|
Path(output).write_bytes(data)
|
|
return
|
|
warn(f"{url}: {reason}; retrying in {2**attempt}s")
|
|
time.sleep(2**attempt)
|
|
|
|
|
|
def install_with_optional_sudo(cmd: Sequence[str]) -> None:
|
|
argv = list(cmd)
|
|
try:
|
|
subprocess.run(argv, check=True)
|
|
except subprocess.CalledProcessError:
|
|
run_privileged(argv)
|