Retry failed DePIN deprovisions

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

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."""

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")