Handle grant webhook and retry failure edges

Author: ishaanxgupta

src/api/routes/billing.py

@@ -341,6 +341,9 @@ async def razorpay_webhook(request: Request) -> dict[str, str]:
         elif package_id in billing_config.TOP_UP_PACKS:
             logger.warning("Razorpay top-up webhook missing order id: %s", event_name)
             raise HTTPException(status_code=400, detail="Webhook order id is required for credit grant")
+        else:
+            logger.warning("Razorpay webhook has unknown grant package id: %s", package_id)
+            raise HTTPException(status_code=400, detail="Webhook package id is not configured for credit grant")
 
     first_seen = await asyncio.to_thread(
         service.store.mark_payment_event,

src/api/routes/v2/jobs.py

@@ -141,9 +141,15 @@ async def retry_job(job_id: str, request: Request, user: dict = Depends(require_
     except InsufficientCredits as exc:
         return _error(request, str(exc), 402, elapsed_ms(start))
     except Exception as exc:
+        release_error = None
         if billing_reservation_created and billing_account_id:
-            await asyncio.to_thread(release_job_reservation, billing_account_id, job_id)
+            try:
+                await asyncio.to_thread(release_job_reservation, billing_account_id, job_id)
+            except Exception as release_exc:
+                release_error = str(release_exc) or release_exc.__class__.__name__
         error = str(exc) or exc.__class__.__name__
+        if release_error:
+            error = f"{error}; billing reservation release failed: {release_error}"
         await asyncio.to_thread(get_default_job_store().mark_failed, job_id, error)
         return _error(request, f"Retry failed to start workflow: {error}", 503, elapsed_ms(start))
     job = await asyncio.to_thread(get_default_job_store().get, job_id)

tests/api/test_memory_versioning.py

@@ -303,6 +303,56 @@ def fail_update_payload(job_id, payload):
     assert store.jobs["job-1"]["error"] == "payload write failed"
 
 
+def test_v2_retry_release_failure_still_marks_job_failed(monkeypatch):
+    app, _ = _build_app(monkeypatch)
+    store = FakeJobStore()
+    store.jobs["job-1"] = {
+        "job_id": "job-1",
+        "job_type": "memory_ingest",
+        "payload": {"billing_account_id": "acct-1", "user_id": "hunter"},
+        "user_id": "hunter",
+        "status": "failed",
+        "timeout_seconds": 30,
+        "max_attempts": 3,
+        "retry_count": 1,
+        "attempt_count": 1,
+        "workflow_id": "old-workflow",
+    }
+
+    class FakeEstimate:
+        reserved_credits = 100
+
+        def model_dump(self):
+            return {"reserved_credits": self.reserved_credits}
+
+    class FakeBillingService:
+        def estimate_required_credits(self, job_type, payload):
+            return FakeEstimate()
+
+        def reserve_credits(self, account_id, job_id, estimated_credits):
+            return SimpleNamespace(reservation_id="reservation-1", created=True)
+
+    async def fake_start_job_workflow(job):
+        raise RuntimeError("temporal unavailable")
+
+    def fail_release(account_id, job_id):
+        raise RuntimeError("mongo unavailable")
+
+    monkeypatch.setattr(jobs_v2, "get_default_job_store", lambda: store)
+    monkeypatch.setattr(durable, "get_default_job_store", lambda: store)
+    monkeypatch.setattr(jobs_v2, "get_default_billing_service", lambda: FakeBillingService())
+    monkeypatch.setattr(jobs_v2, "release_job_reservation", fail_release)
+    monkeypatch.setattr(jobs_v2, "start_job_workflow", fake_start_job_workflow)
+
+    response = TestClient(app).post("/v2/jobs/job-1/retry")
+
+    assert response.status_code == 503
+    assert store.jobs["job-1"]["status"] == "failed"
+    assert store.jobs["job-1"]["error"] == (
+        "temporal unavailable; billing reservation release failed: mongo unavailable"
+    )
+
+
 def test_v2_retry_start_failure_keeps_reused_billing_reservation(monkeypatch):
     app, _ = _build_app(monkeypatch)
     store = FakeJobStore()