From f19d8aa657608248c2143e2759bfbce0a063d78b Mon Sep 17 00:00:00 2001 From: Dhirender Choudhary Date: Fri, 23 Oct 2026 03:47:13 +0530 Subject: [PATCH 1/3] Add deprovision retry tests --- .../workers/test_depin_liveness.py | 73 +++++++++++++++++++ 1 file changed, 73 insertions(+) diff --git a/src/tests/orchestration/workers/test_depin_liveness.py b/src/tests/orchestration/workers/test_depin_liveness.py index 882778da..0475b4e3 100644 --- a/src/tests/orchestration/workers/test_depin_liveness.py +++ b/src/tests/orchestration/workers/test_depin_liveness.py @@ -541,3 +541,76 @@ 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(monkeypatch): + deploys, inventory, pool_repo = _leak_mocks([_LEAKED]) + 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_awaited_once() + assert dep.await_args.kwargs["pool_credential_name"] == "c1" + 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") From 2a8c6325c6d372635a5c63794eeb1b77f8f8a3e4 Mon Sep 17 00:00:00 2001 From: Dhirender Choudhary Date: Fri, 23 Oct 2026 03:48:23 +0530 Subject: [PATCH 2/3] Retry failed DePIN deprovisions --- .../repositories/inventory_repo.py | 34 ++++++++++++++++++ .../workers/depin_liveness_worker.py | 36 ++++++++++++++++++- 2 files changed, 69 insertions(+), 1 deletion(-) 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..6bb056f4 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,35 @@ 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 + 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.""" From c0dcc0a354ef13da2c4978d7db6288149d7d46f4 Mon Sep 17 00:00:00 2001 From: Priyanshu-u07 Date: Sat, 24 Oct 2026 18:55:10 +0530 Subject: [PATCH 3/3] Clear the deprovision marker only once the job is confirmed stopped Nosana's stop ignores the sidecar's HTTP status, so a successful stop call does not prove the job stopped. When the provider still reports the job alive, keep the marker and clear it on a later tick that sees a terminal status. Providers that report no status (k8s, Akash) still clear after a successful stop. Signed-off-by: Priyanshu-u07 --- .../workers/depin_liveness_worker.py | 4 ++++ .../workers/test_depin_liveness.py | 24 ++++++++++++++++--- 2 files changed, 25 insertions(+), 3 deletions(-) diff --git a/src/orchestration/workers/depin_liveness_worker.py b/src/orchestration/workers/depin_liveness_worker.py index 6bb056f4..52a7297b 100644 --- a/src/orchestration/workers/depin_liveness_worker.py +++ b/src/orchestration/workers/depin_liveness_worker.py @@ -245,6 +245,10 @@ async def _retry_failed_deprovisions(self) -> None: 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: diff --git a/src/tests/orchestration/workers/test_depin_liveness.py b/src/tests/orchestration/workers/test_depin_liveness.py index 0475b4e3..cd64f72b 100644 --- a/src/tests/orchestration/workers/test_depin_liveness.py +++ b/src/tests/orchestration/workers/test_depin_liveness.py @@ -555,16 +555,34 @@ def _leak_mocks(rows): "provider_credential_name": "c1"} -async def test_failed_deprovision_is_retried_and_cleared(monkeypatch): +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("RUNNING")) - await w.reconcile_once() + 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")