diff --git a/agents.md b/agents.md index f867d73..15d92cf 100644 --- a/agents.md +++ b/agents.md @@ -112,13 +112,24 @@ re-enqueues an uninstall for every row still in `UNINSTALLATION_QUEUED` or `UNINSTALLING`. `_uninstall_app` asserts no status and tolerates missing files, so resuming it is idempotent — the row is the tombstone. The step only enqueues; the worker starts later in the lifespan, and awaiting task completion before then -deadlocks. Install and reinstall have no equivalent reconciliation yet. +deadlocks. Install is made crash-safe the same way by +`reconcile_interrupted_installs()` (run right before `write_traefik_dyn_config` so +a row it resets to a non-routable status is excluded from the boot's Traefik +config): it re-enqueues an interrupted install when its source still exists (zip +on disk → "install from zip", otherwise a store/config/unknown install → +re-download), resetting the status to `INSTALLATION_QUEUED` first so the worker's +status assertion passes, and marks the row `ERROR` when the source is gone (a +custom upload whose zip is lost). Reinstall is settled to `ERROR` rather than +resumed: `_reinstall_app` deletes the app's files before re-downloading, so +re-running it unattended could destroy a working install if the store is briefly +unreachable at boot — the user retries it manually. Apps in `NOT_ROUTABLE_STATUS` (`app_installation/util.py`) get no Traefik router: `INSTALLATION_QUEUED`, `ERROR`, `UNINSTALLATION_QUEUED`, `UNINSTALLING`. Their files -are either not there yet or already being removed, and `write_traefik_dyn_config` -runs inside the lifespan — an unfiltered status whose metadata is missing raises -`MetadataNotFound` and takes down the boot. +are either not there yet or already being removed. For every other status +`write_traefik_dyn_config` loads the app's metadata in a `try/except` and skips the +app with a warning if it is missing — so a half-installed row can no longer raise +`MetadataNotFound` out of the lifespan and boot-loop core. ### Async Fire-and-Forget Long operations use `asyncio.create_task()` with done callbacks. No thread pools. diff --git a/shard_core/app_factory.py b/shard_core/app_factory.py index 8db51cb..e114061 100644 --- a/shard_core/app_factory.py +++ b/shard_core/app_factory.py @@ -80,11 +80,12 @@ async def lifespan(_): await database.init_database() await identity.init_default_identity() + await app_installation.reconcile_interrupted_uninstalls() + await app_installation.reconcile_interrupted_installs() await write_traefik_dyn_config() await render_all_docker_compose_templates() await app_installation.login_docker_registries() await migration.migrate() - await app_installation.reconcile_interrupted_uninstalls() await app_installation.refresh_init_apps() await backup.ensure_backup_passphrase() try: diff --git a/shard_core/service/app_installation/__init__.py b/shard_core/service/app_installation/__init__.py index 5b7084e..8c6fefe 100644 --- a/shard_core/service/app_installation/__init__.py +++ b/shard_core/service/app_installation/__init__.py @@ -127,6 +127,71 @@ async def reconcile_interrupted_uninstalls(): log.info(f"resuming interrupted uninstallation of {app['name']}") +async def reconcile_interrupted_installs(): + """Resume or fail installs and reinstalls a previous process left in flight. + + Like uninstalls, the task queue lives only in memory: a stop while a row is + in an installing status strands it with nothing left to resume it. Re-enqueue + the work when its source still exists, otherwise mark the row ERROR so it + stops looking half-installed and no longer blocks a fresh install. Enqueue + only — the worker starts later in the lifespan. + """ + async with db_conn() as conn: + all_apps = await db_installed_apps.get_all(conn) + + for record in all_apps: + app = InstalledApp.model_validate(record) + if app.status in (Status.INSTALLATION_QUEUED, Status.INSTALLING): + await _resume_interrupted_install(app) + elif app.status in (Status.REINSTALLATION_QUEUED, Status.REINSTALLING): + await _resume_interrupted_reinstall(app) + + +async def _resume_interrupted_install(app: InstalledApp): + zip_file = get_installed_apps_path() / app.name / f"{app.name}.zip" + if zip_file.exists(): + task_type = "install from zip" + elif app.installation_reason in ( + InstallationReason.STORE, + InstallationReason.CONFIG, + InstallationReason.UNKNOWN, + ): + task_type = "install from store" + else: + await util.update_app_status( + app.name, + Status.ERROR, + message="installation interrupted by a restart and its source is gone", + ) + log.warning( + f"cannot resume interrupted installation of {app.name}, marking ERROR" + ) + return + + # the worker asserts INSTALLATION_QUEUED, so a row stranded in INSTALLING must + # be reset before it is re-enqueued. + await util.update_app_status(app.name, Status.INSTALLATION_QUEUED) + worker.installation_worker.enqueue( + worker.InstallationTask(app_name=app.name, task_type=task_type) + ) + log.info(f"resuming interrupted installation of {app.name}") + + +async def _resume_interrupted_reinstall(app: InstalledApp): + # a reinstall removes the app's files before re-downloading, so re-running it + # unattended could destroy a working install if the store is momentarily + # unreachable at boot. Settle it to ERROR instead — a non-destructive terminal + # state the user can retry from once the store is reachable. + await util.update_app_status( + app.name, + Status.ERROR, + message="reinstallation interrupted by a restart; retry it manually", + ) + log.warning( + f"interrupted reinstallation of {app.name} cannot be safely resumed, marking ERROR" + ) + + async def refresh_init_apps(): try: await database.get_value(STORE_KEY_INITIAL_APPS_INSTALLED) diff --git a/tests/test_app_install_crash_recovery.py b/tests/test_app_install_crash_recovery.py new file mode 100644 index 0000000..4ff560d --- /dev/null +++ b/tests/test_app_install_crash_recovery.py @@ -0,0 +1,208 @@ +import importlib +import shutil + +import pytest +from asgi_lifespan import LifespanManager + +from shard_core import app_factory +from shard_core.data_model.app_meta import InstallationReason, InstalledApp, Status +from shard_core.database import database +from shard_core.database import installed_apps as db_installed_apps +from shard_core.database.connection import db_conn +from shard_core.service import app_installation, telemetry, websocket +from shard_core.service.app_installation import worker +from shard_core.service.app_tools import get_installed_apps_path +from tests.conftest import mock_app_store +from tests.util import docker_network_portal, mock_app_store_path, retry_async + +pytest_plugins = ("pytest_asyncio",) +pytestmark = pytest.mark.asyncio + +config_override = {"apps": {"initial_apps": []}} + +APP_NAME = "mock_app" + + +async def _seed_app(status: Status, reason: InstallationReason): + async with db_conn() as conn: + await db_installed_apps.insert( + conn, + InstalledApp( + name=APP_NAME, + installation_reason=reason, + status=status, + ).model_dump(), + ) + + +async def _status(app_name: str) -> str: + async with db_conn() as conn: + record = await db_installed_apps.get_by_name(conn, app_name) + return record["status"] + + +def _place_zip(app_name: str): + zip_file = get_installed_apps_path() / app_name / f"{app_name}.zip" + zip_file.parent.mkdir(parents=True, exist_ok=True) + zip_file.write_bytes(b"not a real zip, only its presence matters here") + return zip_file + + +# --- decision matrix (db-tier, worker not started) ------------------------ + + +@pytest.mark.parametrize("status", [Status.INSTALLATION_QUEUED, Status.INSTALLING]) +async def test_custom_install_without_zip_is_marked_error(db, mocker, status): + enqueue = mocker.patch.object(worker.installation_worker, "enqueue") + await _seed_app(status, InstallationReason.CUSTOM) + + await app_installation.reconcile_interrupted_installs() + + assert await _status(APP_NAME) == Status.ERROR + enqueue.assert_not_called() + + +@pytest.mark.parametrize("status", [Status.INSTALLATION_QUEUED, Status.INSTALLING]) +@pytest.mark.parametrize( + "reason", [InstallationReason.CUSTOM, InstallationReason.STORE] +) +async def test_install_with_zip_is_requeued_from_zip(db, mocker, status, reason): + # a zip on disk takes precedence over installation_reason + enqueue = mocker.patch.object(worker.installation_worker, "enqueue") + await _seed_app(status, reason) + zip_file = _place_zip(APP_NAME) + try: + await app_installation.reconcile_interrupted_installs() + finally: + zip_file.unlink(missing_ok=True) + + assert await _status(APP_NAME) == Status.INSTALLATION_QUEUED + enqueue.assert_called_once() + assert enqueue.call_args.args[0].task_type == "install from zip" + + +@pytest.mark.parametrize("status", [Status.INSTALLATION_QUEUED, Status.INSTALLING]) +@pytest.mark.parametrize( + "reason", + [InstallationReason.STORE, InstallationReason.CONFIG, InstallationReason.UNKNOWN], +) +async def test_store_install_without_zip_is_requeued_from_store( + db, mocker, status, reason +): + enqueue = mocker.patch.object(worker.installation_worker, "enqueue") + await _seed_app(status, reason) + + await app_installation.reconcile_interrupted_installs() + + assert await _status(APP_NAME) == Status.INSTALLATION_QUEUED + enqueue.assert_called_once() + assert enqueue.call_args.args[0].task_type == "install from store" + + +@pytest.mark.parametrize("status", [Status.REINSTALLATION_QUEUED, Status.REINSTALLING]) +async def test_interrupted_reinstall_is_marked_error(db, mocker, status): + # a reinstall is destructive (rmtree then re-download), so it is settled to + # ERROR rather than resumed unattended — see _resume_interrupted_reinstall. + enqueue = mocker.patch.object(worker.installation_worker, "enqueue") + await _seed_app(status, InstallationReason.STORE) + + await app_installation.reconcile_interrupted_installs() + + assert await _status(APP_NAME) == Status.ERROR + enqueue.assert_not_called() + + +@pytest.mark.parametrize("status", [Status.STOPPED, Status.RUNNING, Status.ERROR]) +async def test_settled_apps_are_left_untouched(db, mocker, status): + enqueue = mocker.patch.object(worker.installation_worker, "enqueue") + await _seed_app(status, InstallationReason.STORE) + + await app_installation.reconcile_interrupted_installs() + + assert await _status(APP_NAME) == status + enqueue.assert_not_called() + + +# --- full-lifespan reconciliation (worker runs) --------------------------- + + +async def _seed_via_own_connection(status: Status, reason: InstallationReason): + """Seed a stranded row before the app's lifespan opens its own pool.""" + await database.init_database() + try: + await _seed_app(status, reason) + finally: + await database.shutdown_database() + + +def _reload_stateful_modules(): + importlib.reload(websocket) + importlib.reload(app_installation.worker) + importlib.reload(telemetry) + + +@pytest.mark.parametrize( + "status", + [ + Status.INSTALLATION_QUEUED, + Status.INSTALLING, + Status.REINSTALLATION_QUEUED, + Status.REINSTALLING, + ], +) +async def test_unrecoverable_stranded_row_boots_and_settles_to_error( + requests_mock, mocker, status +): + """A row with no recoverable source boots the shard and ends in ERROR. + + The store is mocked to serve mock_app, so a row misclassified as a store + install would be driven to STOPPED — the ERROR assertion therefore fails if + reconcile picks the wrong branch, giving it teeth. + """ + _reload_stateful_modules() + mock_app_store(mocker) + + async def noop(): + pass + + mocker.patch("shard_core.service.app_installation.login_docker_registries", noop) + + await _seed_via_own_connection(status, InstallationReason.CUSTOM) + + async with docker_network_portal(): + app = app_factory.create_app() + async with LifespanManager(app, startup_timeout=20, shutdown_timeout=20): + + async def app_row_is_error(): + assert await _status(APP_NAME) == Status.ERROR + + await retry_async(app_row_is_error, timeout=20, frequency=1) + + +@pytest.mark.parametrize("status", [Status.INSTALLATION_QUEUED, Status.INSTALLING]) +async def test_reconciled_install_completes_on_boot(requests_mock, mocker, status): + """A stranded install whose zip is on disk is re-queued by reconcile and the + worker drives it to STOPPED — proving reconcile's task_type and status reset + line up with the worker's status assertion end to end.""" + _reload_stateful_modules() + mock_app_store(mocker) + + async def noop(): + pass + + mocker.patch("shard_core.service.app_installation.login_docker_registries", noop) + + await _seed_via_own_connection(status, InstallationReason.CUSTOM) + src = mock_app_store_path() / APP_NAME / f"{APP_NAME}.zip" + dst = get_installed_apps_path() / APP_NAME / f"{APP_NAME}.zip" + dst.parent.mkdir(parents=True, exist_ok=True) + shutil.copy(src, dst) + + async with docker_network_portal(): + app = app_factory.create_app() + async with LifespanManager(app, startup_timeout=20, shutdown_timeout=20): + + async def app_is_installed(): + assert await _status(APP_NAME) == Status.STOPPED + + await retry_async(app_is_installed, timeout=60, frequency=2)