Startup reconciliation of stranded app installation states (#124)

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.

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:

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)

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)