from __future__ import annotations import importlib.util import sys import types from datetime import UTC, datetime from pathlib import Path from threading import Event, Thread, current_thread import pytest ROOT = Path(__file__).resolve().parents[2] FLEET_PATH = ROOT / "ops" / "gitea-deploy-broker" / "runner_fleet.py" @pytest.fixture def fleet_module(monkeypatch): if sys.platform == "win32" and "fcntl" not in sys.modules: fcntl = types.ModuleType("fcntl") fcntl.LOCK_EX = 2 fcntl.LOCK_UN = 8 fcntl.LOCK_NB = 4 fcntl.flock = lambda *_args, **_kwargs: None monkeypatch.setitem(sys.modules, "fcntl", fcntl) spec = importlib.util.spec_from_file_location("appops_runner_fleet_scheduler_test", FLEET_PATH) assert spec and spec.loader module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) module._host_guard_state_unpatched = module._host_guard_state monkeypatch.setitem( sys.modules, "runner_fleet_specialize", types.SimpleNamespace(specialization_state=lambda: "legacy"), ) monkeypatch.setattr(module, "_host_guard_state", lambda: ("clear", "")) monkeypatch.setattr( module.RunnerCapacityController, "snapshot", lambda _self: _healthy_capacity_snapshot(), ) return module def _healthy_capacity_snapshot(): return { "status": "ok", "decision": {"pressure": "headroom"}, "host": { "swapCritical": False, "loadRatio": 0.1, "availableMemoryBytes": 12 * 1024**3, "cacheFreeBytes": 100 * 1024**3, "swapSampleAt": datetime.now(UTC).timestamp(), }, } def _healthy_snapshot(): runners = [ { "name": name, "running": True, "health": "healthy", "restarts": 0, "startedAt": "2026-09-16T18:00:00Z", "oomKilled": False, "restartPolicy": "unless-stopped", } for name in ( "itworx-security-runner", "itworx-validation-runner-2", ) ] return { "status": "healthy", "generatedAt": "2026-10-01T10:00:00Z", "capacity": _healthy_capacity_snapshot(), "networks": {"count": 0, "capacity": None, "cleanupCandidates": []}, "storage": { "status": "healthy", "freeBytes": 30 * 1024**3, "totalBytes": 40 * 1024**3, }, "scheduler": { "status": "healthy", "candidate": None, "stallThresholdSeconds": 600, }, "topology": { "dind": { "name": "gitea-runner-dind", "running": True, "health": "healthy", "restarts": 0, "startedAt": "2026-09-16T18:00:00Z", "oomKilled": False, "restartPolicy": "unless-stopped", }, "runners": runners, "generalSlots": 2, "runnerCount": len(runners), "specializedRunners": [], "burstRunner": None, }, } def _jobs_page(jobs=()): rows = list(jobs) return {"total_count": len(rows), "jobs": rows} def _scheduler_api(state, error_type): def fake_gitea(path, params=None): params = params or {} if path.endswith("/runners"): runners = state.get( "runners", [ { "id": 17, "name": "itworx-validation-runner-2", "status": "online", "busy": False, "disabled": False, "labels": ["linux-validation"], } ], ) return {"total_count": len(runners), "runners": runners} status = params.get("status") if status == state.get("error_status"): raise error_type("runner_fleet_gitea_test_error") rows = state.get( {"queued": "queued", "pending": "pending", "in_progress": "active"}.get( status, "" ), [], ) if status == state.get("incomplete_status"): return {"total_count": len(rows) + 1, "jobs": rows} return _jobs_page(rows) return fake_gitea def test_container_projection_includes_restart_policy(fleet_module, monkeypatch): service = fleet_module.RunnerFleetService() payload = [ { "Config": {"Labels": {"com.docker.compose.project": "gitea-deploy-runners"}}, "HostConfig": {"RestartPolicy": {"Name": "unless-stopped", "MaximumRetryCount": 0}}, "State": { "Running": True, "StartedAt": "2026-09-21T16:00:00Z", "OOMKilled": False, "Health": {"Status": "healthy"}, }, "RestartCount": 2, } ] monkeypatch.setattr(service, "_run", lambda _args, timeout=20: __import__("json").dumps(payload)) result = service._container("gitea-runner-dind") assert result["restartPolicy"] == "unless-stopped" assert result["running"] is True assert result["health"] == "healthy" assert result["restarts"] == 2 def test_snapshot_keeps_exited_dind_actionable_without_querying_dind_networks( fleet_module, monkeypatch ): service = fleet_module.RunnerFleetService() states = { "gitea-runner-dind": { "name": "gitea-runner-dind", "running": False, "health": "none", "restarts": 0, "startedAt": "", "oomKilled": False, "restartPolicy": "no", }, "itworx-validation-runner-2": { "name": "itworx-validation-runner-2", "running": True, "health": "healthy", "restarts": 0, "startedAt": "2026-09-16T18:00:00Z", "oomKilled": False, "restartPolicy": "unless-stopped", }, "itworx-security-runner": { "name": "itworx-security-runner", "running": True, "health": "healthy", "restarts": 0, "startedAt": "2026-09-16T18:00:00Z", "oomKilled": False, "restartPolicy": "unless-stopped", }, } monkeypatch.setattr(service, "_container", lambda name: states[name]) monkeypatch.setattr(service, "_optional_container", lambda _name: None) monkeypatch.setattr(service, "_storage", lambda: {"status": "healthy", "freeBytes": 20_000_000_000, "totalBytes": 40_000_000_000}) monkeypatch.setattr( service, "_network_inventory", lambda: pytest.fail("network inventory must not query an exited DinD daemon"), ) monkeypatch.setattr( service, "cleanup_candidates", lambda: pytest.fail("cleanup must not query an exited DinD daemon"), ) monkeypatch.setattr(service, "_scheduler_recovery_candidate", lambda: None) result = service.snapshot() assert result["status"] == "critical" assert result["topology"]["dind"]["running"] is False assert result["networks"] == {"count": 0, "capacity": None, "cleanupCandidates": []} assert result["policy"]["hostDockerMutationTargets"][0] == "gitea-runner-dind" def test_snapshot_does_not_query_dind_api_during_health_start_grace( fleet_module, monkeypatch ): service = fleet_module.RunnerFleetService() now = 1_800_000_000 monkeypatch.setattr(fleet_module.time, "time", lambda: now) states = { "gitea-runner-dind": { "name": "gitea-runner-dind", "running": True, "health": "starting", "restarts": 0, "startedAt": datetime.fromtimestamp(now - 10, UTC).isoformat(), "oomKilled": False, "restartPolicy": "unless-stopped", }, "itworx-validation-runner-2": { "name": "itworx-validation-runner-2", "running": True, "health": "healthy", "restarts": 0, "startedAt": datetime.fromtimestamp(now - 300, UTC).isoformat(), "oomKilled": False, "restartPolicy": "unless-stopped", }, "itworx-security-runner": { "name": "itworx-security-runner", "running": True, "health": "healthy", "restarts": 0, "startedAt": datetime.fromtimestamp(now - 300, UTC).isoformat(), "oomKilled": False, "restartPolicy": "unless-stopped", }, } monkeypatch.setattr(service, "_container", lambda name: states[name]) monkeypatch.setattr(service, "_optional_container", lambda _name: None) monkeypatch.setattr(service, "_storage", lambda: {"status": "healthy", "freeBytes": 20_000_000_000, "totalBytes": 40_000_000_000}) monkeypatch.setattr( service, "_network_inventory", lambda: pytest.fail("DinD API must not be queried during health start grace"), ) monkeypatch.setattr( service, "cleanup_candidates", lambda: pytest.fail("cleanup must not query DinD during health start grace"), ) monkeypatch.setattr(service, "_scheduler_recovery_candidate", lambda: None) result = service.snapshot() assert result["status"] == "healthy" assert result["topology"]["dind"]["health"] == "starting" assert result["networks"] == {"count": 0, "capacity": None, "cleanupCandidates": []} def test_automatic_cycle_defers_every_other_recovery_while_dind_is_starting( fleet_module, monkeypatch ): service = fleet_module.RunnerFleetService() snapshot = _healthy_snapshot() snapshot["topology"]["dind"] = { "name": "gitea-runner-dind", "running": True, "health": "starting", "startedAt": "2026-09-21T16:00:00Z", "restartPolicy": "unless-stopped", } commands: list[list[str]] = [] events: list[dict[str, object]] = [] monkeypatch.setattr(service, "snapshot", lambda: snapshot) monkeypatch.setattr(service, "_run", lambda args, timeout=20: commands.append(args) or "") monkeypatch.setattr(service, "_record", lambda **event: events.append(event)) result = service.automatic_cycle() assert commands == [] assert events == [] assert result["dindRecovery"] == { "status": "starting", "target": "gitea-runner-dind", } assert result["schedulerRecovery"] == {"status": "deferred_for_dind_recovery"} def test_automatic_cycle_recovers_critical_exited_dind_with_safe_host_capacity( fleet_module, monkeypatch ): service = fleet_module.RunnerFleetService() snapshot = _healthy_snapshot() snapshot["status"] = "critical" snapshot["topology"]["dind"].update( { "running": False, "health": "none", "startedAt": "", "restartPolicy": "no", } ) commands: list[tuple[list[str], int]] = [] events: list[dict[str, object]] = [] monkeypatch.setattr(service, "snapshot", lambda: snapshot) monkeypatch.setattr(service, "_recovery_available", lambda target: target == "gitea-runner-dind") monkeypatch.setattr( service, "_run", lambda args, timeout=20: commands.append((args, timeout)) or "", ) monkeypatch.setattr(service, "_record", lambda **event: events.append(event)) result = service.automatic_cycle() assert commands == [ (["docker", "update", "--restart", "unless-stopped", "gitea-runner-dind"], 30), (["docker", "start", "gitea-runner-dind"], 30), ] assert [event["action"] for event in events] == [ "restart-policy.update", "start", ] assert result["status"] == "ok" assert result["dindRecovery"] == { "status": "requested", "restartPolicy": "requested", "target": "gitea-runner-dind", } def test_automatic_cycle_repairs_fixed_dind_policy_when_healthy_and_guard_clear( fleet_module, monkeypatch ): service = fleet_module.RunnerFleetService() snapshot = _healthy_snapshot() snapshot["topology"]["dind"]["restartPolicy"] = "no" commands: list[tuple[list[str], int]] = [] events: list[dict[str, object]] = [] monkeypatch.setattr(service, "snapshot", lambda: snapshot) monkeypatch.setattr( service, "_run", lambda args, timeout=20: commands.append((args, timeout)) or "", ) monkeypatch.setattr(service, "_record", lambda **event: events.append(event)) result = service.automatic_cycle() assert commands == [ (["docker", "update", "--restart", "unless-stopped", "gitea-runner-dind"], 30) ] assert result["dindRecovery"] == { "status": "not_needed", "restartPolicy": "requested", } assert len(events) == 1 assert events[0]["action"] == "restart-policy.update" def test_automatic_cycle_never_restarts_running_unhealthy_dind( fleet_module, monkeypatch ): service = fleet_module.RunnerFleetService() snapshot = _healthy_snapshot() snapshot["status"] = "healthy" snapshot["topology"]["dind"] = { "name": "gitea-runner-dind", "running": True, "health": "unhealthy", "startedAt": "2026-09-16T18:00:00Z", "restartPolicy": "unless-stopped", } commands: list[list[str]] = [] events: list[dict[str, object]] = [] monkeypatch.setattr(service, "snapshot", lambda: snapshot) monkeypatch.setattr(service, "_run", lambda args, timeout=20: commands.append(args) or "") monkeypatch.setattr(service, "_record", lambda **event: events.append(event)) result = service.automatic_cycle() assert commands == [] assert result["dindRecovery"] == { "status": "deferred", "target": "gitea-runner-dind", } assert any( event.get("action") == "restart.deferred" and event.get("target") == "gitea-runner-dind" for event in events ) @pytest.mark.parametrize( ("guard_state", "guard_reason"), [ (("paused", "host_resource_guard_paused"), "host_resource_guard_paused"), (("unavailable", "host_guard_state_untrusted"), "host_guard_state_untrusted"), ], ) def test_automatic_cycle_never_mutates_when_host_guard_is_paused_or_untrusted( fleet_module, monkeypatch, guard_state, guard_reason ): service = fleet_module.RunnerFleetService() snapshot = _healthy_snapshot() snapshot["topology"]["dind"].update( {"running": False, "health": "none", "restartPolicy": "no"} ) commands: list[list[str]] = [] events: list[dict[str, object]] = [] monkeypatch.setattr(service, "snapshot", lambda: snapshot) monkeypatch.setattr(service, "_run", lambda args, timeout=20: commands.append(args) or "") monkeypatch.setattr(service, "_dind", lambda args, timeout=20: commands.append(args) or "") monkeypatch.setattr(service, "_record", lambda **event: events.append(event)) monkeypatch.setattr(fleet_module, "_host_guard_state", lambda: guard_state) result = service.automatic_cycle() assert result["status"] == "deferred" assert result["reason"] == guard_reason assert commands == [] assert events == [] @pytest.mark.parametrize( "action", [ "restart-policy-update", "dind-start", "scheduler-restart", "network-remove", "cache-prune", "runner-start", ], ) def test_guard_pause_appearing_after_cycle_entry_blocks_each_mutation( fleet_module, monkeypatch, action ): service = fleet_module.RunnerFleetService() snapshot = _healthy_snapshot() dind = snapshot["topology"]["dind"] monkeypatch.setattr(service, "_scheduler_recovery_candidate", lambda: None) if action == "restart-policy-update": dind["restartPolicy"] = "no" elif action == "dind-start": dind.update({"running": False, "health": "none"}) elif action == "scheduler-restart": snapshot["topology"]["runners"] = [ { "name": "itworx-validation-runner-2", "running": True, "health": "healthy", } ] monkeypatch.setattr( service, "_scheduler_recovery_candidate", lambda: { "target": "itworx-validation-runner-2", "jobId": 22, "ageSeconds": 900, "reason": "stale_scheduler_assignment", }, ) monkeypatch.setattr( service, "_container", lambda name: { "name": name, "running": True, "health": "healthy", }, ) elif action == "network-remove": snapshot["networks"]["cleanupCandidates"] = [ { "id": "a" * 64, "name": "GITEA-ACTIONS-TASK-22-WORKFLOW-ci-JOB-test-network", "ageSeconds": 3600, "reason": "empty_gitea_actions_task_network_in_dedicated_dind", } ] elif action == "cache-prune": snapshot["storage"]["status"] = "critical" elif action == "runner-start": snapshot["topology"]["runners"] = [ { "name": "itworx-validation-runner-2", "running": False, "health": "unhealthy", } ] guard_states = iter( [ ("clear", ""), ("paused", "host_resource_guard_paused"), ] ) commands: list[list[str]] = [] events: list[dict[str, object]] = [] monkeypatch.setattr(service, "snapshot", lambda: snapshot) monkeypatch.setattr(fleet_module, "_host_guard_state", lambda: next(guard_states)) monkeypatch.setattr(service, "_recovery_available", lambda _target: True) monkeypatch.setattr(service, "_run", lambda args, timeout=20: commands.append(args) or "") monkeypatch.setattr(service, "_dind", lambda args, timeout=20: commands.append(args) or "") monkeypatch.setattr(service, "_record", lambda **event: events.append(event)) result = service.automatic_cycle() assert result["status"] == "deferred" assert result["reason"] == "host_resource_guard_paused" assert commands == [] assert events == [] def test_critical_current_pressure_after_cycle_entry_blocks_mutation( fleet_module, monkeypatch ): service = fleet_module.RunnerFleetService() snapshot = _healthy_snapshot() snapshot["topology"]["dind"]["restartPolicy"] = "no" current_samples = iter( [ { "status": "ok", "decision": {"pressure": "critical"}, "host": {"swapCritical": True}, } ] ) commands: list[list[str]] = [] monkeypatch.setattr(service, "snapshot", lambda: snapshot) monkeypatch.setattr(fleet_module, "_host_guard_state", lambda: ("clear", "")) monkeypatch.setattr( fleet_module.RunnerCapacityController, "snapshot", lambda _self: next(current_samples), ) monkeypatch.setattr( service, "_run", lambda args, timeout=20: commands.append(args) or "", ) result = service.automatic_cycle() assert result["status"] == "deferred" assert result["reason"] == "runner_capacity_pressure_critical" assert commands == [] def test_stale_capacity_sample_blocks_automatic_mutation(fleet_module, monkeypatch): service = fleet_module.RunnerFleetService() snapshot = _healthy_snapshot() snapshot["topology"]["dind"]["restartPolicy"] = "no" stale = _healthy_capacity_snapshot() stale["host"]["swapSampleAt"] = ( datetime.now(UTC).timestamp() - fleet_module.CAPACITY_SAMPLE_MAX_AGE_SECONDS - 1 ) commands: list[list[str]] = [] monkeypatch.setattr(service, "snapshot", lambda: snapshot) monkeypatch.setattr(fleet_module, "_host_guard_state", lambda: ("clear", "")) monkeypatch.setattr( fleet_module.RunnerCapacityController, "snapshot", lambda _self: stale, ) monkeypatch.setattr( service, "_run", lambda args, timeout=20: commands.append(args) or "", ) result = service.automatic_cycle() assert result["status"] == "deferred" assert result["reason"] == "runner_capacity_telemetry_unavailable" assert commands == [] @pytest.mark.parametrize("capacity", [{"status": "unavailable"}, {"status": "not_configured"}]) def test_unavailable_or_missing_current_pressure_blocks_mutation( fleet_module, monkeypatch, capacity ): service = fleet_module.RunnerFleetService() snapshot = _healthy_snapshot() snapshot["topology"]["dind"]["restartPolicy"] = "no" commands: list[list[str]] = [] monkeypatch.setattr(service, "snapshot", lambda: snapshot) monkeypatch.setattr(fleet_module, "_host_guard_state", lambda: ("clear", "")) monkeypatch.setattr( fleet_module.RunnerCapacityController, "snapshot", lambda _self: capacity, ) monkeypatch.setattr( service, "_run", lambda args, timeout=20: commands.append(args) or "", ) result = service.automatic_cycle() assert result["status"] == "deferred" assert result["reason"] == "runner_capacity_telemetry_unavailable" assert commands == [] def test_automatic_cycle_rejects_incomplete_critical_snapshot( fleet_module, monkeypatch ): service = fleet_module.RunnerFleetService() snapshot = _healthy_snapshot() snapshot["status"] = "critical" snapshot["topology"]["dind"].update( {"running": False, "health": "none", "restartPolicy": "no"} ) snapshot["topology"].pop("runnerCount") monkeypatch.setattr(service, "snapshot", lambda: snapshot) monkeypatch.setattr( service, "_run", lambda *_args, **_kwargs: pytest.fail("critical fleet must not mutate") ) monkeypatch.setattr( service, "_record", lambda **_event: pytest.fail("critical fleet must not audit a mutation") ) monkeypatch.setattr(fleet_module, "_host_guard_state", lambda: ("clear", "")) result = service.automatic_cycle() assert result["status"] == "deferred" assert result["reason"] == "runner_fleet_snapshot_untrusted" assert result["dindRecovery"]["status"] == "deferred" def test_automatic_cycle_recovers_stalled_scheduler_in_critical_snapshot( fleet_module, monkeypatch ): service = fleet_module.RunnerFleetService() snapshot = _healthy_snapshot() snapshot["status"] = "critical" snapshot["scheduler"] = { "status": "stalled", "candidate": { "target": "itworx-validation-runner-2", "jobId": 22931, "label": "linux-validation", "ageSeconds": 900, "reason": "stale_scheduler_assignment", }, "stallThresholdSeconds": 600, } commands: list[tuple[list[str], int]] = [] events: list[dict[str, object]] = [] monkeypatch.setattr(service, "snapshot", lambda: snapshot) monkeypatch.setattr( service, "_scheduler_recovery_candidate", lambda: snapshot["scheduler"]["candidate"], ) monkeypatch.setattr( service, "_container", lambda name: { "name": name, "running": True, "health": "healthy", "startedAt": "2026-09-16T18:00:00Z", "restarts": 0, "oomKilled": False, "restartPolicy": "unless-stopped", }, ) monkeypatch.setattr(service, "_recovery_available", lambda _target: True) monkeypatch.setattr( service, "_run", lambda args, timeout=20: commands.append((args, timeout)) or "", ) monkeypatch.setattr(service, "_record", lambda **event: events.append(event) or True) result = service.automatic_cycle() assert commands == [ (["docker", "restart", "--timeout", "30", "itworx-validation-runner-2"], 45) ] assert result["status"] == "ok" assert result["schedulerRecovery"] == { "status": "requested", "target": "itworx-validation-runner-2", "jobId": 22931, } assert [event["action"] for event in events] == ["restart", "restart.result"] def test_automatic_cycle_allows_critical_storage_prune_with_safe_host_capacity( fleet_module, monkeypatch ): service = fleet_module.RunnerFleetService() snapshot = _healthy_snapshot() snapshot["status"] = "critical" snapshot["storage"] = { "status": "critical", "freeBytes": 8 * 1024**3, "totalBytes": 40 * 1024**3, } commands: list[list[str]] = [] events: list[dict[str, object]] = [] monkeypatch.setattr(service, "snapshot", lambda: snapshot) monkeypatch.setattr(service, "_scheduler_recovery_candidate", lambda: None) monkeypatch.setattr(service, "_dind", lambda args, timeout=20: commands.append(args) or "") monkeypatch.setattr(service, "_record", lambda **event: events.append(event)) result = service.automatic_cycle() assert commands == [ ["image", "prune", "--force"], ["builder", "prune", "--force", "--filter", "until=168h"], ] assert result["status"] == "ok" assert events == [ { "action": "cache.prune", "target": "gitea-runner-dind", "reason": "ci_storage_critical", "result": "completed", } ] @pytest.mark.parametrize( ("capacity", "reason"), [ ({"status": "ok", "decision": {"pressure": "critical"}, "host": {"swapCritical": False}}, "runner_capacity_pressure_critical"), ({"status": "unavailable"}, "runner_capacity_telemetry_unavailable"), ({"status": "ok", "decision": {}, "host": {}}, "runner_capacity_telemetry_untrusted"), ], ) def test_automatic_cycle_blocks_critical_or_untrusted_capacity_snapshot( fleet_module, monkeypatch, capacity, reason ): service = fleet_module.RunnerFleetService() snapshot = _healthy_snapshot() snapshot["status"] = "critical" snapshot["capacity"] = capacity snapshot["topology"]["dind"].update( {"running": False, "health": "none", "restartPolicy": "no"} ) monkeypatch.setattr(service, "snapshot", lambda: snapshot) monkeypatch.setattr( service, "_run", lambda *_args, **_kwargs: pytest.fail("unsafe capacity must not mutate") ) monkeypatch.setattr(fleet_module, "_host_guard_state", lambda: ("clear", "")) result = service.automatic_cycle() assert result["status"] == "deferred" assert result["reason"] == reason def test_host_guard_probe_fails_closed_when_path_is_not_posix_root(monkeypatch, fleet_module): monkeypatch.setattr(fleet_module.os, "name", "nt") assert fleet_module._host_guard_state_unpatched() == ( "unavailable", "host_guard_state_untrusted", ) def test_recovery_budget_counts_start_and_restart_attempts(fleet_module, monkeypatch): service = fleet_module.RunnerFleetService() now = 1_800_000_000 monkeypatch.setattr(fleet_module.time, "time", lambda: now) recent = datetime.fromtimestamp(now - 60, UTC).isoformat() monkeypatch.setattr( service, "_events", lambda: [ {"at": recent, "action": "start", "target": "gitea-runner-dind"}, {"at": recent, "action": "restart", "target": "gitea-runner-dind"}, {"at": recent, "action": "start", "target": "gitea-runner-dind"}, ], ) assert service._recovery_available("gitea-runner-dind") is False def test_scheduler_candidate_uses_live_gitea_queue_and_fixed_label_target( fleet_module, monkeypatch ): service = fleet_module.RunnerFleetService() monotonic = [1_000.0] created_at = datetime.fromtimestamp(1_800_000_000 - 7_200, UTC).isoformat() monkeypatch.setattr( fleet_module, "_scheduler_monotonic_now", lambda: monotonic[0] ) def fake_gitea(path, params=None): params = params or {} if path.endswith("/runners"): return { "total_count": 1, "runners": [ { "id": 17, "name": "itworx-validation-runner-2", "status": "online", "busy": False, "disabled": False, "labels": ["linux-validation", "ubuntu-latest"], } ] } if params.get("status") == "queued": return _jobs_page( [ { "id": 22931, "status": "queued", "run_id": 3001, "attempt": 1, "created_at": created_at, "labels": ["linux-validation"], } ] ) return _jobs_page() monkeypatch.setattr(service, "_gitea_request", fake_gitea) # An old created_at is not evidence of an old runnable/claimable interval. assert service._scheduler_recovery_candidate() is None candidate = None for _ in range(fleet_module.SCHEDULER_STALL_SECONDS // 60): monotonic[0] += 60 candidate = service._scheduler_recovery_candidate() assert candidate == { "target": "itworx-validation-runner-2", "jobId": 22931, "label": "linux-validation", "runnerId": 17, "labels": ("linux-validation",), "reason": "stale_scheduler_assignment", "runIdentity": (("run_id", 3001), ("attempt", 1)), "ageSeconds": fleet_module.SCHEDULER_STALL_SECONDS, } def test_pending_age_is_not_carried_into_newly_queued_claimability( fleet_module, monkeypatch ): service = fleet_module.RunnerFleetService() monotonic = [1_000.0] created_at = datetime.fromtimestamp(1_800_000_000 - 7_200, UTC).isoformat() pending_job = { "id": 22931, "status": "requested", "created_at": created_at, "labels": ["linux-validation"], } queued_job = {**pending_job, "status": "queued"} state = {"pending": [pending_job], "queued": [], "active": []} monkeypatch.setattr( fleet_module, "_scheduler_monotonic_now", lambda: monotonic[0] ) monkeypatch.setattr( service, "_gitea_request", _scheduler_api(state, fleet_module.RunnerFleetError) ) assert service._scheduler_recovery_candidate() is None state["pending"] = [] state["queued"] = [queued_job] monotonic[0] += fleet_module.SCHEDULER_STALL_SECONDS # The first queued observation starts the timer even though created_at is old. assert service._scheduler_recovery_candidate() is None for _ in range(fleet_module.SCHEDULER_STALL_SECONDS // 60): monotonic[0] += 60 candidate = service._scheduler_recovery_candidate() assert candidate is not None assert candidate["jobId"] == 22931 assert candidate["ageSeconds"] == fleet_module.SCHEDULER_STALL_SECONDS @pytest.mark.parametrize("uncertainty", ["error", "truncated"]) def test_scheduler_claimability_resets_after_unknown_or_incomplete_observation( fleet_module, monkeypatch, uncertainty ): service = fleet_module.RunnerFleetService() monotonic = [2_000.0] old_job = { "id": 22931, "status": "queued", "created_at": datetime.fromtimestamp(1_800_000_000 - 7_200, UTC).isoformat(), "labels": ["linux-validation"], } state = {"queued": [old_job], "pending": [], "active": []} monkeypatch.setattr( fleet_module, "_scheduler_monotonic_now", lambda: monotonic[0] ) monkeypatch.setattr( service, "_gitea_request", _scheduler_api(state, fleet_module.RunnerFleetError) ) assert service._scheduler_recovery_candidate() is None monotonic[0] += 60 if uncertainty == "error": state["error_status"] = "pending" else: state["incomplete_status"] = "queued" with pytest.raises(fleet_module.RunnerFleetError): service._scheduler_recovery_candidate() state.pop("error_status", None) state.pop("incomplete_status", None) monotonic[0] += 60 assert service._scheduler_recovery_candidate() is None for _ in range((fleet_module.SCHEDULER_STALL_SECONDS // 60) - 2): monotonic[0] += 60 candidate = service._scheduler_recovery_candidate() # The pre-error timer would have matured at this instant if unknown time # had been bridged instead of starting over at the first fresh sample. assert candidate is None for _ in range(2): monotonic[0] += 60 candidate = service._scheduler_recovery_candidate() assert candidate is not None assert candidate["ageSeconds"] == fleet_module.SCHEDULER_STALL_SECONDS @pytest.mark.parametrize("loss", ["busy", "active"]) def test_scheduler_claimability_resets_when_target_becomes_unavailable( fleet_module, monkeypatch, loss ): service = fleet_module.RunnerFleetService() monotonic = [3_000.0] job = { "id": 22931, "status": "queued", "labels": ["linux-validation"], } state = {"queued": [job], "pending": [], "active": []} monkeypatch.setattr( fleet_module, "_scheduler_monotonic_now", lambda: monotonic[0] ) monkeypatch.setattr( service, "_gitea_request", _scheduler_api(state, fleet_module.RunnerFleetError) ) assert service._scheduler_recovery_candidate() is None monotonic[0] += 60 if loss == "busy": state["runners"] = [ { "id": 17, "name": "itworx-validation-runner-2", "status": "online", "busy": True, "disabled": False, "labels": ["linux-validation"], } ] else: state["active"] = [ {"id": 22932, "status": "in_progress", "runner_id": 17} ] assert service._scheduler_recovery_candidate() is None state.pop("runners", None) state["active"] = [] monotonic[0] += 60 assert service._scheduler_recovery_candidate() is None for _ in range((fleet_module.SCHEDULER_STALL_SECONDS // 60) - 1): monotonic[0] += 60 candidate = service._scheduler_recovery_candidate() assert candidate is None monotonic[0] += 60 candidate = service._scheduler_recovery_candidate() assert candidate is not None assert candidate["ageSeconds"] == fleet_module.SCHEDULER_STALL_SECONDS def test_scheduler_claimability_gap_restarts_the_observation_window( fleet_module, monkeypatch ): service = fleet_module.RunnerFleetService() monotonic = [4_000.0] state = { "queued": [{"id": 22931, "status": "queued", "labels": ["linux-validation"]}], "pending": [], "active": [], } monkeypatch.setattr( fleet_module, "_scheduler_monotonic_now", lambda: monotonic[0] ) monkeypatch.setattr( service, "_gitea_request", _scheduler_api(state, fleet_module.RunnerFleetError) ) assert service._scheduler_recovery_candidate() is None monotonic[0] += fleet_module.SCHEDULER_OBSERVATION_MAX_GAP_SECONDS + 1 assert service._scheduler_recovery_candidate() is None monotonic[0] += fleet_module.SCHEDULER_STALL_SECONDS # A second long gap resets again instead of treating elapsed wall time as observed. assert service._scheduler_recovery_candidate() is None def test_older_eligible_sample_cannot_repopulate_after_newer_busy_sample( fleet_module, monkeypatch ): service = fleet_module.RunnerFleetService() monotonic = [1_000.0] queued_job = { "id": 22931, "status": "queued", "run_id": 3001, "attempt": 1, "labels": ["linux-validation"], } idle_state = {"queued": [queued_job], "pending": [], "active": []} live_state = {"queued": [queued_job], "pending": [], "active": []} stale_request = _scheduler_api(idle_state, fleet_module.RunnerFleetError) live_request = _scheduler_api(live_state, fleet_module.RunnerFleetError) older_sample_entered = Event() release_older_sample = Event() older_result: dict[str, object] = {} def fake_gitea(path, params=None): if current_thread().name == "older-idle-sample": if not older_sample_entered.is_set(): older_sample_entered.set() assert release_older_sample.wait(timeout=5) return stale_request(path, params) return live_request(path, params) monkeypatch.setattr( fleet_module, "_scheduler_monotonic_now", lambda: monotonic[0] ) monkeypatch.setattr(service, "_gitea_request", fake_gitea) assert service._scheduler_recovery_candidate() is None monotonic[0] = 1_010.0 older_sample = Thread( name="older-idle-sample", target=lambda: older_result.setdefault( "candidate", service._scheduler_recovery_candidate() ), ) older_sample.start() assert older_sample_entered.wait(timeout=5) monotonic[0] = 1_020.0 live_state["runners"] = [ { "id": 17, "name": "itworx-validation-runner-2", "status": "online", "busy": True, "disabled": False, "labels": ["linux-validation"], } ] assert service._scheduler_recovery_candidate() is None monotonic[0] = 1_030.0 release_older_sample.set() older_sample.join(timeout=5) assert not older_sample.is_alive() assert older_result["candidate"] is None live_state.pop("runners") monotonic[0] = 1_031.0 assert service._scheduler_recovery_candidate() is None for _ in range(9): monotonic[0] += 60 candidate = service._scheduler_recovery_candidate() # The rejected stale sample must not have started the ten-minute dwell. assert monotonic[0] == 1_571.0 assert candidate is None monotonic[0] += 59 assert service._scheduler_recovery_candidate() is None monotonic[0] += 1 candidate = service._scheduler_recovery_candidate() assert candidate is not None assert candidate["ageSeconds"] == fleet_module.SCHEDULER_STALL_SECONDS def test_scheduler_claimability_is_process_local_and_lost_on_restart( fleet_module, monkeypatch ): service = fleet_module.RunnerFleetService() monotonic = [5_000.0] state = { "queued": [{"id": 22931, "status": "queued", "labels": ["linux-validation"]}], "pending": [], "active": [], } monkeypatch.setattr( fleet_module, "_scheduler_monotonic_now", lambda: monotonic[0] ) monkeypatch.setattr( service, "_gitea_request", _scheduler_api(state, fleet_module.RunnerFleetError) ) assert service._scheduler_recovery_candidate() is None fleet_module.RunnerFleetService._clear_scheduler_claimability() monotonic[0] += fleet_module.SCHEDULER_STALL_SECONDS assert service._scheduler_recovery_candidate() is None @pytest.mark.parametrize("identity_change", ["origin", "job", "run", "labels", "runner"]) def test_scheduler_claimability_restarts_when_job_or_runner_identity_changes( fleet_module, monkeypatch, identity_change ): service = fleet_module.RunnerFleetService() monotonic = [6_000.0] job = { "id": 22931, "run_id": 3001, "attempt": 1, "status": "queued", "labels": ["linux-validation"], } state = {"queued": [job], "pending": [], "active": []} monkeypatch.setattr( fleet_module, "_scheduler_monotonic_now", lambda: monotonic[0] ) monkeypatch.setattr( service, "_gitea_request", _scheduler_api(state, fleet_module.RunnerFleetError) ) assert service._scheduler_recovery_candidate() is None for _ in range((fleet_module.SCHEDULER_STALL_SECONDS // 60) - 1): monotonic[0] += 60 assert service._scheduler_recovery_candidate() is None if identity_change == "origin": monkeypatch.setattr( fleet_module, "GITEA_API_BASE", "http://other-gitea:3000" ) elif identity_change == "job": job["id"] = 22932 elif identity_change == "run": job["run_id"] = 3002 elif identity_change == "labels": job["labels"] = ["ubuntu-latest"] state["runners"] = [ { "id": 17, "name": "itworx-validation-runner-2", "status": "online", "busy": False, "disabled": False, "labels": ["linux-validation", "ubuntu-latest"], } ] else: state["runners"] = [ { "id": 18, "name": "itworx-validation-runner-2", "status": "online", "busy": False, "disabled": False, "labels": ["linux-validation"], } ] monotonic[0] += 60 assert service._scheduler_recovery_candidate() is None def test_scheduler_candidate_pages_jobs_at_gitea_fifty_row_limit( fleet_module, monkeypatch ): service = fleet_module.RunnerFleetService() queued = [ {"id": job_id, "status": "queued", "labels": ["unmapped-label"]} for job_id in range(1, 51) ] queued.append( {"id": 51, "status": "queued", "labels": ["linux-validation"]} ) job_calls: list[dict[str, object]] = [] def fake_gitea(path, params=None): params = params or {} if path.endswith("/runners"): return { "total_count": 1, "runners": [ { "id": 17, "name": "itworx-validation-runner-2", "status": "online", "busy": False, "disabled": False, "labels": ["linux-validation"], } ], } status = params["status"] if status == "queued": job_calls.append(params) start = (int(params["page"]) - 1) * 50 return {"total_count": 51, "jobs": queued[start : start + 50]} return _jobs_page() monkeypatch.setattr(service, "_gitea_request", fake_gitea) assert service._scheduler_recovery_candidate() is None assert [(call["page"], call["limit"]) for call in job_calls] == [ (1, fleet_module.GITEA_JOB_PAGE_LIMIT), (2, fleet_module.GITEA_JOB_PAGE_LIMIT), ] def test_scheduler_candidate_rejects_job_inventory_over_one_thousand_rows( fleet_module, monkeypatch ): service = fleet_module.RunnerFleetService() def fake_gitea(path, params=None): if path.endswith("/runners"): return {"total_count": 0, "runners": []} return {"total_count": 1_001, "jobs": []} monkeypatch.setattr(service, "_gitea_request", fake_gitea) with pytest.raises( fleet_module.RunnerFleetError, match="runner_fleet_gitea_invalid_job_inventory", ): service._scheduler_recovery_candidate() def test_scheduler_candidate_fails_closed_when_job_appears_in_two_statuses( fleet_module, monkeypatch ): service = fleet_module.RunnerFleetService() monotonic = [7_000.0] job = {"id": 22931, "labels": ["linux-validation"]} state = { "queued": [{**job, "status": "queued"}], "pending": [{**job, "status": "requested"}], "active": [], } monkeypatch.setattr( fleet_module, "_scheduler_monotonic_now", lambda: monotonic[0] ) monkeypatch.setattr( service, "_gitea_request", _scheduler_api(state, fleet_module.RunnerFleetError) ) with pytest.raises( fleet_module.RunnerFleetError, match="runner_fleet_gitea_job_status_changed_during_read", ): service._scheduler_recovery_candidate() state["pending"] = [] monotonic[0] += fleet_module.SCHEDULER_STALL_SECONDS assert service._scheduler_recovery_candidate() is None def test_scheduler_candidate_does_not_classify_offline_target_as_claimable( fleet_module, monkeypatch ): service = fleet_module.RunnerFleetService() state = { "queued": [ { "id": 22931, "status": "queued", "created_at": datetime.fromtimestamp( 1_800_000_000 - 7_200, UTC ).isoformat(), "labels": ["linux-validation"], } ], "pending": [], "active": [], "runners": [ { "id": 17, "name": "itworx-validation-runner-2", "status": "offline", "busy": False, "disabled": False, "labels": ["linux-validation"], } ], } monkeypatch.setattr( service, "_gitea_request", _scheduler_api(state, fleet_module.RunnerFleetError) ) assert service._scheduler_recovery_candidate() is None @pytest.mark.parametrize( ("label_source", "expected_error"), [ ("job", "runner_fleet_gitea_invalid_queued_job"), ("runner", "runner_fleet_gitea_invalid_runner_inventory"), ], ) def test_scheduler_candidate_rejects_labels_that_would_change_when_trimmed( fleet_module, monkeypatch, label_source, expected_error ): service = fleet_module.RunnerFleetService() state = { "queued": [ {"id": 22931, "status": "queued", "labels": ["linux-validation"]} ], "pending": [], "active": [], } if label_source == "job": state["queued"][0]["labels"] = [" linux-validation"] else: state["runners"] = [ { "id": 17, "name": "itworx-validation-runner-2", "status": "online", "busy": False, "disabled": False, "labels": ["linux-validation "], } ] monkeypatch.setattr( service, "_gitea_request", _scheduler_api(state, fleet_module.RunnerFleetError) ) with pytest.raises(fleet_module.RunnerFleetError, match=expected_error): service._scheduler_recovery_candidate() def test_scheduler_candidate_never_targets_busy_runner(fleet_module, monkeypatch): service = fleet_module.RunnerFleetService() def fake_gitea(path, params=None): params = params or {} if path.endswith("/runners"): return { "total_count": 1, "runners": [ { "id": 17, "name": "itworx-validation-runner-2", "status": "online", "busy": True, "disabled": False, "labels": ["linux-validation"], } ] } if params.get("status") == "queued": return _jobs_page( [{"id": 22931, "status": "queued", "labels": ["linux-validation"]}] ) return _jobs_page() monkeypatch.setattr(service, "_gitea_request", fake_gitea) assert service._scheduler_recovery_candidate() is None def test_scheduler_candidate_never_restarts_runner_with_in_progress_job( fleet_module, monkeypatch ): service = fleet_module.RunnerFleetService() def fake_gitea(path, params=None): params = params or {} if path.endswith("/runners"): return { "total_count": 1, "runners": [{ "id": 17, "name": "itworx-validation-runner-2", "status": "online", "busy": False, "disabled": False, "labels": ["linux-validation"], }], } if params.get("status") == "queued": return _jobs_page( [{"id": 22931, "status": "queued", "labels": ["linux-validation"]}] ) if params.get("status") == "in_progress": return _jobs_page( [{"id": 22932, "runner_id": 17, "status": "in_progress"}] ) return _jobs_page() monkeypatch.setattr(service, "_gitea_request", fake_gitea) assert service._scheduler_recovery_candidate() is None def test_scheduler_candidate_never_hides_disabled_busy_runner(fleet_module, monkeypatch): service = fleet_module.RunnerFleetService() def fake_gitea(path, params=None): params = params or {} if path.endswith("/runners"): if params.get("disabled") == "false": return {"total_count": 0, "runners": []} return { "total_count": 1, "runners": [ { "id": 17, "name": "itworx-validation-runner-2", "status": "online", "busy": True, "disabled": True, "labels": ["linux-validation"], } ] } if params.get("status") == "queued": return _jobs_page( [{"id": 22931, "status": "queued", "labels": ["linux-validation"]}] ) return _jobs_page() monkeypatch.setattr(service, "_gitea_request", fake_gitea) assert service._scheduler_recovery_candidate() is None def test_scheduler_candidate_reads_later_runner_inventory_pages( fleet_module, monkeypatch ): service = fleet_module.RunnerFleetService() runner_calls: list[dict[str, object]] = [] def fake_gitea(path, params=None): params = params or {} if path.endswith("/runners"): runner_calls.append(params) if params.get("page") == 2: return { "total_count": 2, "runners": [ { "id": 18, "name": "itworx-validation-runner-2", "status": "online", "busy": True, "disabled": True, "labels": ["linux-validation"], } ], } return { "total_count": 2, "runners": [ { "id": 17, "name": "unrelated-runner", "status": "online", "busy": False, "disabled": False, "labels": ["unrelated"], } ], } if params.get("status") == "queued": return _jobs_page( [{"id": 22931, "status": "queued", "labels": ["linux-validation"]}] ) return _jobs_page() monkeypatch.setattr(service, "_gitea_request", fake_gitea) assert service._scheduler_recovery_candidate() is None assert [call.get("page") for call in runner_calls] == [1, 2] def test_scheduler_candidate_fails_closed_on_malformed_runner_inventory( fleet_module, monkeypatch ): service = fleet_module.RunnerFleetService() def fake_gitea(path, params=None): if path.endswith("/runners"): return {"total_count": 1} return _jobs_page() monkeypatch.setattr(service, "_gitea_request", fake_gitea) with pytest.raises( fleet_module.RunnerFleetError, match="runner_fleet_gitea_invalid_runner_inventory", ): service._scheduler_recovery_candidate() @pytest.mark.parametrize("unsafe_first", [True, False]) def test_scheduler_candidate_vetoes_any_unsafe_duplicate_fixed_runner( fleet_module, monkeypatch, unsafe_first ): service = fleet_module.RunnerFleetService() unsafe = { "id": 17, "name": "itworx-validation-runner-2", "status": "online", "busy": True, "disabled": True, "labels": ["linux-validation"], } idle = { "id": 18, "name": "itworx-validation-runner-2", "status": "online", "busy": False, "disabled": False, "labels": ["linux-validation"], } def fake_gitea(path, params=None): params = params or {} if path.endswith("/runners"): runners = [unsafe, idle] if unsafe_first else [idle, unsafe] return {"total_count": 2, "runners": runners} if params.get("status") == "queued": return _jobs_page( [{"id": 22931, "status": "queued", "labels": ["linux-validation"]}] ) return _jobs_page() monkeypatch.setattr(service, "_gitea_request", fake_gitea) assert service._scheduler_recovery_candidate() is None def test_automatic_cycle_rechecks_the_same_scheduler_candidate_before_restart( fleet_module, monkeypatch ): service = fleet_module.RunnerFleetService() commands: list[list[str]] = [] events: list[dict[str, object]] = [] candidate = { "target": "itworx-validation-runner-2", "jobId": 22931, "label": "linux-validation", "runnerId": 17, "labels": ("linux-validation",), "runIdentity": (("run_id", 3001), ("attempt", 1)), "ageSeconds": 600, "reason": "stale_scheduler_assignment", } changed = {**candidate, "runIdentity": (("run_id", 3002), ("attempt", 1))} candidates = iter([candidate, changed]) monkeypatch.setattr(service, "snapshot", _healthy_snapshot) monkeypatch.setattr( service, "_scheduler_recovery_candidate", lambda: next(candidates) ) monkeypatch.setattr(service, "_recovery_available", lambda _target: True) monkeypatch.setattr( service, "_container", lambda _target: { "name": "itworx-validation-runner-2", "running": True, "health": "healthy", }, ) monkeypatch.setattr( service, "_run", lambda args, timeout=20: commands.append(args) or "" ) monkeypatch.setattr(service, "_record", lambda **event: events.append(event) or True) result = service.automatic_cycle() assert commands == [] assert result["schedulerRecovery"] == { "status": "ineligible", "target": "itworx-validation-runner-2", "jobId": 22931, "reason": "scheduler_claimability_changed", } assert any( event.get("action") == "restart.result" and event.get("result") == "cancelled" for event in events ) @pytest.mark.parametrize( ("unsafe_source", "expected_reason"), [ ("host_guard", "host_resource_guard_paused"), ("pressure", "runner_capacity_pressure_critical"), ], ) def test_automatic_cycle_rechecks_host_safety_after_final_scheduler_read( fleet_module, monkeypatch, unsafe_source, expected_reason ): service = fleet_module.RunnerFleetService() commands: list[list[str]] = [] events: list[dict[str, object]] = [] safety_changed = [False] candidate = { "target": "itworx-validation-runner-2", "jobId": 22931, "label": "linux-validation", "runnerId": 17, "labels": ("linux-validation",), "runIdentity": (("run_id", 3001), ("attempt", 1)), "ageSeconds": 600, "reason": "stale_scheduler_assignment", } scheduler_reads = [0] def read_candidate(): scheduler_reads[0] += 1 if scheduler_reads[0] == 2: safety_changed[0] = True return candidate def read_host_guard(): if safety_changed[0] and unsafe_source == "host_guard": return "paused", "host_resource_guard_paused" return "clear", "" def read_capacity(_self): capacity = _healthy_capacity_snapshot() if safety_changed[0] and unsafe_source == "pressure": capacity["decision"]["pressure"] = "critical" return capacity monkeypatch.setattr(service, "snapshot", _healthy_snapshot) monkeypatch.setattr(service, "_scheduler_recovery_candidate", read_candidate) monkeypatch.setattr(service, "_recovery_available", lambda _target: True) monkeypatch.setattr( service, "_container", lambda _target: { "name": "itworx-validation-runner-2", "running": True, "health": "healthy", }, ) monkeypatch.setattr(fleet_module, "_host_guard_state", read_host_guard) monkeypatch.setattr( fleet_module.RunnerCapacityController, "snapshot", read_capacity ) monkeypatch.setattr( service, "_run", lambda args, timeout=20: commands.append(args) or "" ) monkeypatch.setattr( service, "_record", lambda **event: events.append(event) or True ) result = service.automatic_cycle() assert result["status"] == "deferred" assert result["reason"] == expected_reason assert commands == [] # The durable intent remains in the budget even though the final fresh # safety check cancelled the Docker mutation. assert any( event.get("action") == "restart" and event.get("result") == "intent" for event in events ) assert not any(event.get("action") == "restart.result" for event in events) def test_automatic_cycle_restarts_only_exact_known_idle_runner_for_stale_scheduler( fleet_module, monkeypatch ): service = fleet_module.RunnerFleetService() commands: list[tuple[list[str], int]] = [] events: list[dict[str, object]] = [] monkeypatch.setattr(service, "snapshot", _healthy_snapshot) monkeypatch.setattr( service, "_scheduler_recovery_candidate", lambda: { "target": "itworx-validation-runner-2", "jobId": 22931, "label": "linux-validation", "ageSeconds": 900, "reason": "stale_scheduler_assignment", }, ) monkeypatch.setattr(service, "_recovery_available", lambda _target: True) monkeypatch.setattr( service, "_container", lambda _target: { "name": "itworx-validation-runner-2", "running": True, "health": "healthy", }, ) monkeypatch.setattr( service, "_run", lambda args, timeout=20: commands.append((args, timeout)) or "", ) monkeypatch.setattr( service, "_record", lambda **event: events.append(event) or True ) result = service.automatic_cycle() assert commands == [ (["docker", "restart", "--timeout", "30", "itworx-validation-runner-2"], 45) ] assert any( event.get("action") == "restart" and event.get("target") == "itworx-validation-runner-2" and event.get("reason") == "stale_scheduler_assignment" for event in events ) assert result["schedulerRecovery"] == { "status": "requested", "target": "itworx-validation-runner-2", "jobId": 22931, } def test_scheduler_restart_persists_intent_before_invoking_docker( fleet_module, monkeypatch, tmp_path ): service = fleet_module.RunnerFleetService() monkeypatch.setattr(fleet_module, "STATE_ROOT", tmp_path) monkeypatch.setattr(service, "snapshot", _healthy_snapshot) monkeypatch.setattr( service, "_scheduler_recovery_candidate", lambda: { "target": "itworx-validation-runner-2", "jobId": 22931, "label": "linux-validation", "ageSeconds": 900, "reason": "stale_scheduler_assignment", }, ) monkeypatch.setattr(service, "_recovery_available", lambda _target: True) monkeypatch.setattr( service, "_container", lambda _target: { "name": "itworx-validation-runner-2", "running": True, "health": "healthy", }, ) def observe_durable_intent(args, timeout=20): assert args == [ "docker", "restart", "--timeout", "30", "itworx-validation-runner-2", ] assert timeout == 45 persisted = service._events() assert persisted[-1] == { "at": persisted[-1]["at"], "action": "restart", "target": "itworx-validation-runner-2", "reason": "stale_scheduler_assignment", "result": "intent", "ageSeconds": 900, } return "" monkeypatch.setattr(service, "_run", observe_durable_intent) result = service.automatic_cycle() assert result["schedulerRecovery"]["status"] == "requested" def test_scheduler_recovery_refuses_restart_when_intent_cannot_be_persisted( fleet_module, monkeypatch ): service = fleet_module.RunnerFleetService() commands: list[list[str]] = [] monkeypatch.setattr(service, "snapshot", _healthy_snapshot) monkeypatch.setattr( service, "_scheduler_recovery_candidate", lambda: { "target": "itworx-validation-runner-2", "jobId": 22931, "label": "linux-validation", "ageSeconds": 900, "reason": "stale_scheduler_assignment", }, ) monkeypatch.setattr(service, "_recovery_available", lambda _target: True) monkeypatch.setattr( service, "_container", lambda _target: { "name": "itworx-validation-runner-2", "running": True, "health": "healthy", }, ) monkeypatch.setattr(service, "_safe_state_root", lambda: False) monkeypatch.setattr( service, "_run", lambda args, timeout=20: commands.append(args) or "" ) result = service.automatic_cycle() assert commands == [] assert result["schedulerRecovery"] == { "status": "failed", "target": "itworx-validation-runner-2", "jobId": 22931, "reason": "restart_intent_not_persisted", } def test_scheduler_recovery_refuses_restart_when_intent_write_raises_oserror( fleet_module, monkeypatch, tmp_path ): service = fleet_module.RunnerFleetService() commands: list[list[str]] = [] monkeypatch.setattr(fleet_module, "STATE_ROOT", tmp_path) monkeypatch.setattr(service, "snapshot", _healthy_snapshot) monkeypatch.setattr( service, "_scheduler_recovery_candidate", lambda: { "target": "itworx-validation-runner-2", "jobId": 22931, "label": "linux-validation", "ageSeconds": 900, "reason": "stale_scheduler_assignment", }, ) monkeypatch.setattr(service, "_recovery_available", lambda _target: True) monkeypatch.setattr( service, "_container", lambda _target: { "name": "itworx-validation-runner-2", "running": True, "health": "healthy", }, ) monkeypatch.setattr( fleet_module.os, "open", lambda *_args, **_kwargs: (_ for _ in ()).throw( PermissionError("state storage is unwritable") ), ) monkeypatch.setattr( service, "_run", lambda args, timeout=20: commands.append(args) or "" ) result = service.automatic_cycle() assert commands == [] assert result["schedulerRecovery"] == { "status": "failed", "target": "itworx-validation-runner-2", "jobId": 22931, "reason": "restart_intent_not_persisted", } def test_scheduler_recovery_keeps_intent_when_outcome_write_fails( fleet_module, monkeypatch, tmp_path ): service = fleet_module.RunnerFleetService() commands: list[list[str]] = [] event_open_count = 0 real_open = fleet_module.os.open monkeypatch.setattr(fleet_module, "STATE_ROOT", tmp_path) monkeypatch.setattr(fleet_module, "RECOVERY_BUDGET", 1) monkeypatch.setattr(service, "snapshot", _healthy_snapshot) monkeypatch.setattr( service, "_scheduler_recovery_candidate", lambda: { "target": "itworx-validation-runner-2", "jobId": 22931, "label": "linux-validation", "ageSeconds": 900, "reason": "stale_scheduler_assignment", }, ) monkeypatch.setattr( service, "_container", lambda _target: { "name": "itworx-validation-runner-2", "running": True, "health": "healthy", }, ) def fail_outcome_open(path, flags, mode=0o777): nonlocal event_open_count if Path(path).name.startswith(".events."): event_open_count += 1 if event_open_count == 2: raise PermissionError("outcome storage is unwritable") return real_open(path, flags, mode) monkeypatch.setattr(fleet_module.os, "open", fail_outcome_open) monkeypatch.setattr( service, "_run", lambda args, timeout=20: commands.append(args) or "" ) result = service.automatic_cycle() assert commands == [ ["docker", "restart", "--timeout", "30", "itworx-validation-runner-2"] ] assert result["schedulerRecovery"] == { "status": "requested", "target": "itworx-validation-runner-2", "jobId": 22931, } assert [event["action"] for event in service._events()] == ["restart"] assert service._recovery_available("itworx-validation-runner-2") is False @pytest.mark.parametrize( ("running", "health"), [ (True, "starting"), (True, "none"), (True, "unhealthy"), (False, "healthy"), ], ) def test_scheduler_recovery_rechecks_strict_runner_health_before_restart( fleet_module, monkeypatch, running, health ): service = fleet_module.RunnerFleetService() commands: list[list[str]] = [] monkeypatch.setattr(service, "snapshot", _healthy_snapshot) monkeypatch.setattr( service, "_scheduler_recovery_candidate", lambda: { "target": "itworx-validation-runner-2", "jobId": 22931, "label": "linux-validation", "ageSeconds": 900, "reason": "stale_scheduler_assignment", }, ) monkeypatch.setattr(service, "_recovery_available", lambda _target: True) monkeypatch.setattr( service, "_container", lambda _target: { "name": "itworx-validation-runner-2", "running": running, "health": health, }, ) monkeypatch.setattr( service, "_run", lambda args, timeout=20: commands.append(args) or "" ) monkeypatch.setattr(service, "_record", lambda **_event: True) result = service.automatic_cycle() assert commands == [] assert result["schedulerRecovery"] == { "status": "ineligible", "target": "itworx-validation-runner-2", "jobId": 22931, "reason": "docker_health_not_healthy", } def test_scheduler_recovery_rejects_duplicate_fresh_inspect_results( fleet_module, monkeypatch ): service = fleet_module.RunnerFleetService() commands: list[tuple[list[str], int]] = [] monkeypatch.setattr(service, "snapshot", _healthy_snapshot) monkeypatch.setattr( service, "_scheduler_recovery_candidate", lambda: { "target": "itworx-validation-runner-2", "jobId": 22931, "label": "linux-validation", "ageSeconds": 900, "reason": "stale_scheduler_assignment", }, ) monkeypatch.setattr(service, "_recovery_available", lambda _target: True) def duplicate_inspect(args, timeout=20): commands.append((args, timeout)) return "[{}, {}]" monkeypatch.setattr(service, "_run", duplicate_inspect) monkeypatch.setattr(service, "_record", lambda **_event: True) result = service.automatic_cycle() assert commands == [ (["docker", "inspect", "itworx-validation-runner-2"], 20) ] assert result["schedulerRecovery"] == { "status": "ineligible", "target": "itworx-validation-runner-2", "jobId": 22931, "reason": "docker_health_not_healthy", } def test_scheduler_recovery_rejects_runner_without_healthy_docker_state( fleet_module, monkeypatch ): service = fleet_module.RunnerFleetService() snapshot = _healthy_snapshot() next( runner for runner in snapshot["topology"]["runners"] if runner["name"] == "itworx-validation-runner-2" )["health"] = "unhealthy" commands: list[list[str]] = [] monkeypatch.setattr(service, "snapshot", lambda: snapshot) monkeypatch.setattr( service, "_scheduler_recovery_candidate", lambda: { "target": "itworx-validation-runner-2", "jobId": 22931, "label": "linux-validation", "ageSeconds": 900, "reason": "stale_scheduler_assignment", }, ) monkeypatch.setattr(service, "_recovery_available", lambda _target: True) monkeypatch.setattr( service, "_run", lambda args, timeout=20: commands.append(args) or "" ) monkeypatch.setattr(service, "_record", lambda **_event: None) result = service.automatic_cycle() assert commands == [] assert result["schedulerRecovery"] == { "status": "ineligible", "target": "itworx-validation-runner-2", "jobId": 22931, "reason": "docker_health_not_healthy", } def test_scheduler_recovery_refuses_unknown_target(fleet_module, monkeypatch): service = fleet_module.RunnerFleetService() commands: list[list[str]] = [] monkeypatch.setattr(service, "snapshot", _healthy_snapshot) monkeypatch.setattr( service, "_scheduler_recovery_candidate", lambda: { "target": "Gitea", "jobId": 22931, "label": "linux-validation", "ageSeconds": 900, "reason": "stale_scheduler_assignment", }, ) monkeypatch.setattr(service, "_recovery_available", lambda _target: True) monkeypatch.setattr(service, "_run", lambda args, timeout=20: commands.append(args) or "") monkeypatch.setattr(service, "_record", lambda **_event: None) result = service.automatic_cycle() assert commands == [] assert result["schedulerRecovery"]["status"] == "rejected" def test_scheduler_recovery_obeys_existing_recovery_budget(fleet_module, monkeypatch): service = fleet_module.RunnerFleetService() commands: list[list[str]] = [] monkeypatch.setattr(service, "snapshot", _healthy_snapshot) monkeypatch.setattr( service, "_scheduler_recovery_candidate", lambda: { "target": "itworx-validation-runner-2", "jobId": 22931, "label": "linux-validation", "ageSeconds": 900, "reason": "stale_scheduler_assignment", }, ) monkeypatch.setattr(service, "_recovery_available", lambda _target: False) monkeypatch.setattr(service, "_run", lambda args, timeout=20: commands.append(args) or "") monkeypatch.setattr(service, "_record", lambda **_event: None) result = service.automatic_cycle() assert commands == [] assert result["schedulerRecovery"]["status"] == "budget_exhausted" def test_scheduler_recovery_does_nothing_without_stale_candidate(fleet_module, monkeypatch): service = fleet_module.RunnerFleetService() commands: list[list[str]] = [] monkeypatch.setattr(service, "snapshot", _healthy_snapshot) monkeypatch.setattr(service, "_scheduler_recovery_candidate", lambda: None) monkeypatch.setattr(service, "_run", lambda args, timeout=20: commands.append(args) or "") monkeypatch.setattr(service, "_record", lambda **_event: None) result = service.automatic_cycle() assert commands == [] assert result["schedulerRecovery"] == {"status": "not_needed"}