fix(install,catalog): recover from interrupted installs, don't strand sync on network/disk errors
- an unfinished install-state.json whose runner is gone (free lock, >60 s old — reboot, OOM kill, runner never started) is marked 'error' instead of blocking every later install with 409 until someone fixes it over SSH - read_state in install_runner/updater tolerates unreadable and non-object files - Forgejo API + release downloads wrap OSError and HTTPException (socket timeouts, resets, disk full) — the update/catalog check endpoints used to drop the connection on a flaky network - catalog sync records any failure as the 'error' stage, not just CatalogError Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Wq1hzUp4ZmCKHDuLKCnu73
This commit is contained in:
parent
22c2245ac4
commit
9f9e627678
6 changed files with 118 additions and 6 deletions
|
|
@ -15,6 +15,7 @@ neutral ``ReleaseError`` for use in tests or standalone scripts.
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
import hashlib
|
import hashlib
|
||||||
|
import http.client
|
||||||
import json
|
import json
|
||||||
import shutil
|
import shutil
|
||||||
import tarfile
|
import tarfile
|
||||||
|
|
@ -33,7 +34,9 @@ def forgejo_api(host: str, repo: str, path: str, *, error_cls: type = ReleaseErr
|
||||||
try:
|
try:
|
||||||
with urllib.request.urlopen(req, timeout=15) as resp:
|
with urllib.request.urlopen(req, timeout=15) as resp:
|
||||||
return json.loads(resp.read())
|
return json.loads(resp.read())
|
||||||
except (urllib.error.URLError, json.JSONDecodeError) as e:
|
except (OSError, http.client.HTTPException, json.JSONDecodeError) as e:
|
||||||
|
# OSError covers URLError plus socket timeouts and resets while
|
||||||
|
# reading the body, which urllib doesn't wrap.
|
||||||
raise error_cls(f"forgejo api {url}: {e}") from e
|
raise error_cls(f"forgejo api {url}: {e}") from e
|
||||||
|
|
||||||
|
|
||||||
|
|
@ -43,7 +46,8 @@ def download(url: str, dest: Path, *, error_cls: type = ReleaseError) -> None:
|
||||||
try:
|
try:
|
||||||
with urllib.request.urlopen(req, timeout=60) as resp, dest.open("wb") as f:
|
with urllib.request.urlopen(req, timeout=60) as resp, dest.open("wb") as f:
|
||||||
shutil.copyfileobj(resp, f)
|
shutil.copyfileobj(resp, f)
|
||||||
except urllib.error.URLError as e:
|
except (OSError, http.client.HTTPException) as e:
|
||||||
|
# Includes disk full while writing `dest` and mid-body disconnects.
|
||||||
raise error_cls(f"download {url}: {e}") from e
|
raise error_cls(f"download {url}: {e}") from e
|
||||||
|
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -776,7 +776,7 @@ def _do_get_settings(name):
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
_INSTALL_TERMINAL_STAGES = frozenset({"done", "error"})
|
_INSTALL_TERMINAL_STAGES = install_runner.TERMINAL_STAGES
|
||||||
|
|
||||||
|
|
||||||
def _do_install_plan(name):
|
def _do_install_plan(name):
|
||||||
|
|
@ -830,8 +830,9 @@ def _do_install(name, settings=None, confirm_dependencies=False):
|
||||||
# POST can sneak in during that tiny window. Reading the state
|
# POST can sneak in during that tiny window. Reading the state
|
||||||
# first closes that gap: as long as a previous install hasn't
|
# first closes that gap: as long as a previous install hasn't
|
||||||
# written "done" or "error", we refuse.
|
# written "done" or "error", we refuse.
|
||||||
|
install_runner.clear_stale_state()
|
||||||
current_state = install_runner.read_state()
|
current_state = install_runner.read_state()
|
||||||
current_stage = current_state.get("stage", "") if isinstance(current_state, dict) else ""
|
current_stage = current_state.get("stage", "")
|
||||||
if current_stage and current_stage not in _INSTALL_TERMINAL_STAGES:
|
if current_stage and current_stage not in _INSTALL_TERMINAL_STAGES:
|
||||||
return 409, {
|
return 409, {
|
||||||
"error": (
|
"error": (
|
||||||
|
|
@ -927,6 +928,7 @@ def _do_install(name, settings=None, confirm_dependencies=False):
|
||||||
|
|
||||||
def _do_install_status():
|
def _do_install_status():
|
||||||
"""Return the current install-state.json contents (or {})."""
|
"""Return the current install-state.json contents (or {})."""
|
||||||
|
install_runner.clear_stale_state()
|
||||||
return 200, install_runner.read_state()
|
return 200, install_runner.read_state()
|
||||||
|
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -216,6 +216,9 @@ def sync_catalog() -> CatalogCheck:
|
||||||
except CatalogError as e:
|
except CatalogError as e:
|
||||||
write_state("error", error=str(e))
|
write_state("error", error=str(e))
|
||||||
raise
|
raise
|
||||||
|
except Exception as e:
|
||||||
|
write_state("error", error=str(e) or type(e).__name__)
|
||||||
|
raise CatalogError(str(e) or type(e).__name__) from e
|
||||||
|
|
||||||
|
|
||||||
def _sync_locked() -> CatalogCheck:
|
def _sync_locked() -> CatalogCheck:
|
||||||
|
|
|
||||||
|
|
@ -91,9 +91,53 @@ def write_state(stage: str, **extra) -> None:
|
||||||
|
|
||||||
def read_state() -> dict:
|
def read_state() -> dict:
|
||||||
try:
|
try:
|
||||||
return json.loads(state_path().read_text())
|
data = json.loads(state_path().read_text())
|
||||||
except (FileNotFoundError, json.JSONDecodeError):
|
except (OSError, UnicodeDecodeError, json.JSONDecodeError):
|
||||||
return {}
|
return {}
|
||||||
|
return data if isinstance(data, dict) else {}
|
||||||
|
|
||||||
|
|
||||||
|
TERMINAL_STAGES = frozenset({"done", "error"})
|
||||||
|
|
||||||
|
# How old an unfinished state must be before we believe the job is gone.
|
||||||
|
# Covers the gap between the API writing the first stage and the detached
|
||||||
|
# runner taking the lock.
|
||||||
|
_STALE_AFTER_SECONDS = 60
|
||||||
|
|
||||||
|
|
||||||
|
def clear_stale_state(now: float | None = None) -> bool:
|
||||||
|
"""Mark an unfinished install as failed if no runner is alive any more.
|
||||||
|
|
||||||
|
install-state.json lives in /var/lib and survives reboots, OOM kills and
|
||||||
|
a runner that never started — each of which used to block every later
|
||||||
|
install with "another install is in progress" until someone fixed it
|
||||||
|
over SSH. The runner holds the flock for its whole run, so a free lock
|
||||||
|
plus an old timestamp means nobody is coming back to finish. Returns
|
||||||
|
True when it reset the state.
|
||||||
|
"""
|
||||||
|
state = read_state()
|
||||||
|
stage = state.get("stage", "")
|
||||||
|
if not stage or stage in TERMINAL_STAGES:
|
||||||
|
return False
|
||||||
|
try:
|
||||||
|
updated = time.mktime(time.strptime(state.get("updated_at", ""), "%Y-%m-%dT%H:%M:%S%z"))
|
||||||
|
except (TypeError, ValueError):
|
||||||
|
updated = 0.0
|
||||||
|
if (now if now is not None else time.time()) - updated < _STALE_AFTER_SECONDS:
|
||||||
|
return False
|
||||||
|
try:
|
||||||
|
fh = acquire_lock()
|
||||||
|
except (InstallRunnerError, OSError):
|
||||||
|
return False # a runner is still working
|
||||||
|
try:
|
||||||
|
write_state(
|
||||||
|
"error",
|
||||||
|
app=state.get("app", ""),
|
||||||
|
error=f"the install was interrupted at '{stage}' (restart or crash) — try again",
|
||||||
|
)
|
||||||
|
finally:
|
||||||
|
fh.close()
|
||||||
|
return True
|
||||||
|
|
||||||
|
|
||||||
def _read_plan(target: str) -> dict:
|
def _read_plan(target: str) -> dict:
|
||||||
|
|
|
||||||
|
|
@ -349,3 +349,30 @@ def test_sync_catalog_failure_writes_error_state(catalog, monkeypatch):
|
||||||
state = catalog.read_state()
|
state = catalog.read_state()
|
||||||
assert state["stage"] == "error"
|
assert state["stage"] == "error"
|
||||||
assert "Network is unreachable" in state["error"]
|
assert "Network is unreachable" in state["error"]
|
||||||
|
|
||||||
|
|
||||||
|
def test_sync_catalog_unexpected_error_lands_as_error_stage(catalog, monkeypatch):
|
||||||
|
"""Disk full / mid-body disconnect used to leave the stage at
|
||||||
|
'downloading' forever because only CatalogError was caught."""
|
||||||
|
_stub_forgejo_release(monkeypatch, catalog, tag="26.6")
|
||||||
|
|
||||||
|
def boom(*a, **kw):
|
||||||
|
raise OSError(28, "No space left on device")
|
||||||
|
|
||||||
|
monkeypatch.setattr(catalog._rc, "download", boom)
|
||||||
|
with pytest.raises(catalog.CatalogError, match="No space left"):
|
||||||
|
catalog.sync_catalog()
|
||||||
|
assert catalog.read_state()["stage"] == "error"
|
||||||
|
|
||||||
|
|
||||||
|
def test_release_download_wraps_socket_errors(tmp_path, monkeypatch):
|
||||||
|
from furtka import _release_common as rc
|
||||||
|
|
||||||
|
def reset(*a, **kw):
|
||||||
|
raise ConnectionResetError("reset by peer")
|
||||||
|
|
||||||
|
monkeypatch.setattr(rc.urllib.request, "urlopen", reset)
|
||||||
|
with pytest.raises(rc.ReleaseError, match="reset by peer"):
|
||||||
|
rc.download("https://x/y", tmp_path / "y")
|
||||||
|
with pytest.raises(rc.ReleaseError, match="reset by peer"):
|
||||||
|
rc.forgejo_api("x", "a/b", "/releases")
|
||||||
|
|
|
||||||
|
|
@ -478,3 +478,35 @@ def test_parse_hook_output_furtka_json_rejects_bad_payload(runner):
|
||||||
def test_parse_hook_output_furtka_json_invalid_json(runner):
|
def test_parse_hook_output_furtka_json_invalid_json(runner):
|
||||||
with pytest.raises(runner.InstallRunnerError, match="invalid FURTKA_JSON"):
|
with pytest.raises(runner.InstallRunnerError, match="invalid FURTKA_JSON"):
|
||||||
runner._parse_hook_output("FURTKA_JSON: {not json}\n")
|
runner._parse_hook_output("FURTKA_JSON: {not json}\n")
|
||||||
|
|
||||||
|
|
||||||
|
def test_clear_stale_state_resets_abandoned_install(runner):
|
||||||
|
import time
|
||||||
|
|
||||||
|
runner.write_state("pulling_image", app="fileshare")
|
||||||
|
# Fresh state: could still be the runner starting up — leave it alone.
|
||||||
|
assert runner.clear_stale_state() is False
|
||||||
|
assert runner.read_state()["stage"] == "pulling_image"
|
||||||
|
# Two minutes later with nobody holding the lock: interrupted.
|
||||||
|
assert runner.clear_stale_state(now=time.time() + 120) is True
|
||||||
|
state = runner.read_state()
|
||||||
|
assert state["stage"] == "error"
|
||||||
|
assert state["app"] == "fileshare"
|
||||||
|
assert "interrupted" in state["error"]
|
||||||
|
|
||||||
|
|
||||||
|
def test_clear_stale_state_keeps_running_install(runner):
|
||||||
|
import time
|
||||||
|
|
||||||
|
runner.write_state("pulling_image", app="fileshare")
|
||||||
|
fh = runner.acquire_lock()
|
||||||
|
try:
|
||||||
|
assert runner.clear_stale_state(now=time.time() + 3600) is False
|
||||||
|
assert runner.read_state()["stage"] == "pulling_image"
|
||||||
|
finally:
|
||||||
|
fh.close()
|
||||||
|
|
||||||
|
|
||||||
|
def test_read_state_tolerates_non_object(runner):
|
||||||
|
runner.state_path().write_text("[1, 2]")
|
||||||
|
assert runner.read_state() == {}
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue