diff --git a/src/orchestration/repositories/inventory_repo.py b/src/orchestration/repositories/inventory_repo.py index 262f474a..6aaf7198 100644 --- a/src/orchestration/repositories/inventory_repo.py +++ b/src/orchestration/repositories/inventory_repo.py @@ -1384,6 +1384,36 @@ async def _mark_deprovision_failed_node_impl(self, *, node_id, reason=None): ) +async def _list_deprovision_failed_nodes_impl(self): + """Nodes whose external job may still be running because a deprovision + call failed, with the pool credential needed to retry it.""" + async with self.db.acquire() as conn: + rows = await conn.fetch( + """ + SELECT ci.id, ci.provider::text AS provider, ci.provider_instance_id, + cp.provider_credential_name + FROM compute_inventory ci + LEFT JOIN compute_pools cp ON cp.id = ci.pool_id + WHERE ci.metadata->>'deprovision_failed' = 'true' + """ + ) + return [dict(r) for r in rows] + + +async def _clear_deprovision_failed_impl(self, *, node_id): + nid = uuid.UUID(node_id) if isinstance(node_id, str) else node_id + async with self.db.acquire() as conn: + await conn.execute( + """ + UPDATE compute_inventory + SET metadata = metadata - 'deprovision_failed' - 'deprovision_error', + updated_at = now() + WHERE id = $1 + """, + nid, + ) + + async def _set_state_impl(self, *, node_id, state): """Set compute_inventory.state to an arbitrary node_state enum value. @@ -1469,3 +1499,7 @@ async def _clear_terminating_node_impl(self, *, node_id): InventoryRepository.mark_deprovision_failed_node = ( _mark_deprovision_failed_node_impl ) +InventoryRepository.list_deprovision_failed_nodes = ( + _list_deprovision_failed_nodes_impl +) +InventoryRepository.clear_deprovision_failed = _clear_deprovision_failed_impl diff --git a/src/orchestration/workers/depin_liveness_worker.py b/src/orchestration/workers/depin_liveness_worker.py index fcb5f5aa..52a7297b 100644 --- a/src/orchestration/workers/depin_liveness_worker.py +++ b/src/orchestration/workers/depin_liveness_worker.py @@ -111,6 +111,7 @@ async def reconcile_once(self) -> None: await self._sweep("RUNNING", self._check_running) await self._sweep("DEPLOYING", self._check_deploying) await self._sweep("PENDING_NODE", self._check_pending) + await self._retry_failed_deprovisions() async def _sweep(self, state: str, handler) -> None: rows = await self.deploys.list_by_state(state) @@ -204,7 +205,11 @@ async def _fail_and_teardown( return False if pii and node: try: - await _deprovision_direct_node(node, pool_credential_name=cred) + ok, err = await _deprovision_direct_node(node, pool_credential_name=cred) + if not ok and node_id: + await self.inventory.mark_deprovision_failed_node( + node_id=node_id, reason=err + ) except Exception: logger.exception( "DePIN reconcile: deprovision failed for %s", deploy_id @@ -218,6 +223,39 @@ async def _fail_and_teardown( ) return True + async def _retry_failed_deprovisions(self) -> None: + """Retry stopping external jobs whose deprovision call failed. The + node row is already terminated, so without this the job keeps + billing until someone stops it by hand.""" + for n in await self.inventory.list_deprovision_failed_nodes() or []: + if not is_direct_provision_provider(n.get("provider")): + continue + node_id = n["id"] + cred = n.get("provider_credential_name") + try: + status = await self._get_adapter(n["provider"]).get_node_status( + provider_instance_id=n["provider_instance_id"], + provider_credential_name=cred, + ) + if status not in _TERMINAL_STATES: + ok, err = await _deprovision_direct_node(n, pool_credential_name=cred) + if not ok: + logger.warning( + "DePIN reconcile: deprovision retry failed for node %s: %s", + node_id, err, + ) + continue + if status != "unknown": + # The provider still reports the job alive, and Nosana's stop + # ignores the HTTP status: confirm on a later tick. + continue + await self.inventory.clear_deprovision_failed(node_id=node_id) + logger.info("DePIN reconcile: external job for node %s is stopped", node_id) + except Exception: + logger.exception( + "DePIN reconcile: deprovision retry errored for node %s", node_id + ) + # ------------------------------------------------------------------ checks async def _check_running(self, d: dict) -> None: """Dead-RUNNING: a RUNNING deploy whose external job is terminal.""" diff --git a/src/tests/orchestration/workers/test_depin_liveness.py b/src/tests/orchestration/workers/test_depin_liveness.py index 882778da..cd64f72b 100644 --- a/src/tests/orchestration/workers/test_depin_liveness.py +++ b/src/tests/orchestration/workers/test_depin_liveness.py @@ -541,3 +541,94 @@ async def test_get_logs_pending_returns_running(monkeypatch): monkeypatch.setattr(na.aiohttp, "ClientSession", lambda *a, **k: _FakeSeqSession(seq)) out = await NosanaAdapter().get_logs(provider_instance_id="job1") assert any("running" in str(x).lower() for x in out["logs"]) + + +def _leak_mocks(rows): + deploys = AsyncMock(spec=ModelDeploymentRepository) + deploys.list_by_state.return_value = [] + inventory = AsyncMock(spec=InventoryRepository) + inventory.list_deprovision_failed_nodes.return_value = rows + return deploys, inventory, AsyncMock(spec=ComputePoolRepository) + + +_LEAKED = {"id": "n9", "provider": "nosana", "provider_instance_id": "job-x", + "provider_credential_name": "c1"} + + +async def test_failed_deprovision_is_retried_and_cleared_once_stopped(monkeypatch): + deploys, inventory, pool_repo = _leak_mocks([_LEAKED]) + dep = AsyncMock(return_value=(True, None)) + monkeypatch.setattr(mod, "_deprovision_direct_node", dep) + statuses = iter(["RUNNING", "STOPPED"]) + w = mod.DepinLivenessWorker(deploys=deploys, inventory=inventory, pool_repo=pool_repo, + get_adapter_fn=lambda p: _adapter(next(statuses))) + + await w.reconcile_once() + dep.assert_awaited_once() + assert dep.await_args.kwargs["pool_credential_name"] == "c1" + inventory.clear_deprovision_failed.assert_not_awaited() + + await w.reconcile_once() + dep.assert_awaited_once() + inventory.clear_deprovision_failed.assert_awaited_once_with(node_id="n9") + + +async def test_failed_deprovision_with_unknown_status_clears_after_stop(monkeypatch): + # k8s and Akash report no status, so a successful stop is all there is to go on. + deploys, inventory, pool_repo = _leak_mocks([{**_LEAKED, "provider": "k8s"}]) + dep = AsyncMock(return_value=(True, None)) + monkeypatch.setattr(mod, "_deprovision_direct_node", dep) + w = mod.DepinLivenessWorker(deploys=deploys, inventory=inventory, pool_repo=pool_repo, + get_adapter_fn=lambda p: _adapter("unknown")) + await w.reconcile_once() + + dep.assert_awaited_once() + inventory.clear_deprovision_failed.assert_awaited_once_with(node_id="n9") + + +async def test_failed_deprovision_still_failing_keeps_marker(monkeypatch): + deploys, inventory, pool_repo = _leak_mocks([_LEAKED]) + monkeypatch.setattr(mod, "_deprovision_direct_node", AsyncMock(return_value=(False, "down"))) + w = mod.DepinLivenessWorker(deploys=deploys, inventory=inventory, pool_repo=pool_repo, + get_adapter_fn=lambda p: _adapter("RUNNING")) + await w.reconcile_once() + + inventory.clear_deprovision_failed.assert_not_awaited() + + +async def test_failed_deprovision_job_already_gone_clears_marker(monkeypatch): + deploys, inventory, pool_repo = _leak_mocks([_LEAKED]) + dep = AsyncMock(return_value=(False, "already closed")) + monkeypatch.setattr(mod, "_deprovision_direct_node", dep) + w = mod.DepinLivenessWorker(deploys=deploys, inventory=inventory, pool_repo=pool_repo, + get_adapter_fn=lambda p: _adapter("COMPLETED")) + await w.reconcile_once() + + dep.assert_not_awaited() + inventory.clear_deprovision_failed.assert_awaited_once_with(node_id="n9") + + +async def test_failed_deprovision_retry_skips_cloud_nodes(monkeypatch): + deploys, inventory, pool_repo = _leak_mocks([{**_LEAKED, "provider": "aws"}]) + dep = AsyncMock(return_value=(True, None)) + monkeypatch.setattr(mod, "_deprovision_direct_node", dep) + w = mod.DepinLivenessWorker(deploys=deploys, inventory=inventory, pool_repo=pool_repo, + get_adapter_fn=lambda p: _adapter("RUNNING")) + await w.reconcile_once() + + dep.assert_not_awaited() + + +async def test_teardown_failure_stamps_marker(monkeypatch): + deploys, inventory, pool_repo = _mocks( + deploy_rows=[{"deployment_id": "d1", "pool_id": "p1", "target_node_id": "n1"}], + node={"provider_instance_id": "job-abc", "provider": "nosana"}, + pool={"provider": "nosana", "provider_credential_name": "cred1"}, + ) + inventory.list_deprovision_failed_nodes.return_value = [] + monkeypatch.setattr(mod, "_deprovision_direct_node", AsyncMock(return_value=(False, "sidecar down"))) + w = mod.DepinLivenessWorker(deploys=deploys, inventory=inventory, pool_repo=pool_repo, + get_adapter_fn=lambda p: _adapter("COMPLETED")) + await w.reconcile_once() + + inventory.mark_deprovision_failed_node.assert_awaited_once_with(node_id="n1", reason="sidecar down")