From 7fa0fb8a983bf23233ca5c76829489a27ed3f1f9 Mon Sep 17 00:00:00 2001 From: suguanYang Date: Tue, 12 May 2026 03:44:52 +0000 Subject: [PATCH] fix: reconcile qstash webhook delivery state --- .../api/app/api/v1/routes/qstash_callbacks.py | 67 +++++++-- .../contract/test_qstash_callback_contract.py | 137 +++++++++++++++++- apps/worker/app/core/tasks/webhook_tasks.py | 103 ++++++++++++- .../test_webhook_recovery_contract.py | 137 +++++++++++++++++- .../shared/core/config/qstash.py | 4 + .../services/webhook/qstash_publisher.py | 54 ++++++- 6 files changed, 483 insertions(+), 19 deletions(-) diff --git a/apps/api/app/api/v1/routes/qstash_callbacks.py b/apps/api/app/api/v1/routes/qstash_callbacks.py index cfa2b133d..84cae94f2 100644 --- a/apps/api/app/api/v1/routes/qstash_callbacks.py +++ b/apps/api/app/api/v1/routes/qstash_callbacks.py @@ -128,25 +128,67 @@ def _build_callback_log_idempotency_key( return event_id +def _get_response_status_code(value: Any) -> Optional[int]: + """Return the destination response status reported by QStash.""" + if value is None: + return None + + try: + return int(value) + except (TypeError, ValueError): + return None + + +def _is_success_response_status(status_code: Optional[int]) -> bool: + """Return whether a destination response status is successful.""" + return status_code is not None and 200 <= status_code < 300 + + +def _get_callback_event_status(data: Dict[str, Any]) -> str: + """Map a normal QStash callback to the current webhook event status.""" + response_status = _get_response_status_code(data.get("status")) + if _is_success_response_status(response_status): + return WebhookEventStatus.DELIVERED + + return WebhookEventStatus.DELIVERING + + +def _resolve_event_status(current_status: str, callback_status: str) -> str: + """Apply callback status without downgrading terminal delivery state.""" + if current_status in ( + WebhookEventStatus.DELIVERED, + WebhookEventStatus.FAILED, + WebhookEventStatus.CANCELED, + ): + return current_status + + return callback_status + + def _process_qstash_callback( data: Dict[str, Any], event_id: str, - terminal_status: str, + callback_status: str, log_label: str, ) -> Response: """Shared logic for both success and failure QStash callbacks. Fetches the WebhookEvent, updates its status, and writes a WebhookLog entry. """ - response_status = data.get("status") + response_status_code = _get_response_status_code(data.get("status")) response_body = data.get("body", "") qstash_message_id = data.get("sourceMessageId") retried = data.get("retried", 0) - error_message = ( - data.get("error", response_body) - if terminal_status == WebhookEventStatus.FAILED - else None + is_failed_delivery_attempt = ( + callback_status == WebhookEventStatus.FAILED + or ( + callback_status == WebhookEventStatus.DELIVERING + and not _is_success_response_status(response_status_code) + ) ) + error_message = None + if is_failed_delivery_attempt: + error_message = data.get("error") or response_body with get_sync_db_context() as db: event = db.execute( @@ -158,21 +200,23 @@ def _process_qstash_callback( return Response(status_code=200, content="OK (event not found)") now = datetime.now(timezone.utc).replace(tzinfo=None) - event.status = terminal_status - event.attempts = retried + 1 + event_status = _resolve_event_status(event.status, callback_status) + attempt_number = retried + 1 + event.status = event_status + event.attempts = max(event.attempts, attempt_number) event.updated_at = now log = WebhookLog( job_id=event.job_id, event_id=event.id, webhook_url=event.target_url, - attempt_number=retried + 1, + attempt_number=attempt_number, request_payload=event.payload, signature="", idempotency_key=_build_callback_log_idempotency_key( qstash_message_id, event.id ), - response_status_code=int(response_status) if response_status else None, + response_status_code=response_status_code, response_body=response_body[:4096] if response_body else None, error_message=str(error_message)[:4096] if error_message else None, duration_ms=0, @@ -210,8 +254,9 @@ async def handle_qstash_callback(request: Request) -> Response: f"retried={retried}, qstash_message_id={data.get('sourceMessageId')}" ) + event_status = _get_callback_event_status(data) return _process_qstash_callback( - data, event_id, WebhookEventStatus.DELIVERED, "callback" + data, event_id, event_status, "callback" ) diff --git a/apps/api/tests/contract/test_qstash_callback_contract.py b/apps/api/tests/contract/test_qstash_callback_contract.py index 73be267a5..9b6734d2d 100644 --- a/apps/api/tests/contract/test_qstash_callback_contract.py +++ b/apps/api/tests/contract/test_qstash_callback_contract.py @@ -11,7 +11,12 @@ from tests.support.contract_database import ContractDatabase -async def _insert_qstash_event() -> tuple[str, str]: +async def _insert_qstash_event( + *, + status: str = "pending", + attempts: int = 0, + qstash_message_id: str | None = None, +) -> tuple[str, str]: user_id = f"contract-qstash-user-{uuid4().hex[:12]}" job_id = f"job_{uuid4().hex[:12]}" event_id = str(uuid4()) @@ -29,6 +34,9 @@ async def _insert_qstash_event() -> tuple[str, str]: job_id=job_id, target_url="https://hooks.contract.test/qstash", payload={"job_id": job_id, "status": "done"}, + status=status, + attempts=attempts, + qstash_message_id=qstash_message_id, ) return job_id, event_id @@ -112,6 +120,133 @@ async def test_should_mark_the_matching_event_delivered_and_persist_a_webhook_lo } +@pytest.mark.asyncio +async def test_should_keep_the_matching_event_delivering_for_retry_callback_with_non_success_status( + api_client_factory: Callable[[], AbstractAsyncContextManager[AsyncClient]], + monkeypatch: MonkeyPatch, +) -> None: + job_id: str = "" + event_id: str = "" + + async with api_client_factory() as api_client: + job_id, event_id = await _insert_qstash_event( + status="delivering", + qstash_message_id="qstash-message-retry", + ) + qstash_module = importlib.import_module("app.api.v1.routes.qstash_callbacks") + monkeypatch.setattr(qstash_module, "_verify_qstash_signature", lambda *args: True) + response = await api_client.post( + "/api/v1/webhooks/qstash/callback", + json={ + "status": 503, + "body": "temporary unavailable", + "retried": 2, + "sourceMessageId": "qstash-message-retry", + "sourceHeader": {"X-Knowhere-Event-Id": event_id}, + }, + headers={"upstash-signature": "contract-valid"}, + ) + + assert response.status_code == 200 + assert response.text == "OK" + + event_row = await ContractDatabase.fetch_webhook_event(event_id) + log_rows = await ContractDatabase.fetch_all( + """ + SELECT + job_id, + event_id, + attempt_number, + response_status_code, + response_body, + error_message, + qstash_message_id + FROM webhook_logs + WHERE event_id = :event_id + """, + {"event_id": event_id}, + ) + + assert event_row is not None + assert event_row["status"] == "delivering" + assert event_row["attempts"] == 3 + + assert len(log_rows) == 1 + assert log_rows[0] == { + "job_id": job_id, + "event_id": event_id, + "attempt_number": 3, + "response_status_code": 503, + "response_body": "temporary unavailable", + "error_message": "temporary unavailable", + "qstash_message_id": "qstash-message-retry", + } + + +@pytest.mark.asyncio +async def test_should_not_downgrade_terminal_event_when_retry_callback_arrives_late( + api_client_factory: Callable[[], AbstractAsyncContextManager[AsyncClient]], + monkeypatch: MonkeyPatch, +) -> None: + job_id: str = "" + event_id: str = "" + + async with api_client_factory() as api_client: + job_id, event_id = await _insert_qstash_event( + status="delivered", + attempts=4, + qstash_message_id="qstash-message-late-retry", + ) + qstash_module = importlib.import_module("app.api.v1.routes.qstash_callbacks") + monkeypatch.setattr(qstash_module, "_verify_qstash_signature", lambda *args: True) + response = await api_client.post( + "/api/v1/webhooks/qstash/callback", + json={ + "status": 503, + "body": "late retry callback", + "retried": 1, + "sourceMessageId": "qstash-message-late-retry", + "sourceHeader": {"X-Knowhere-Event-Id": event_id}, + }, + headers={"upstash-signature": "contract-valid"}, + ) + + assert response.status_code == 200 + assert response.text == "OK" + + event_row = await ContractDatabase.fetch_webhook_event(event_id) + log_rows = await ContractDatabase.fetch_all( + """ + SELECT + job_id, + event_id, + attempt_number, + response_status_code, + response_body, + error_message, + qstash_message_id + FROM webhook_logs + WHERE event_id = :event_id + """, + {"event_id": event_id}, + ) + + assert event_row is not None + assert event_row["status"] == "delivered" + assert event_row["attempts"] == 4 + + assert len(log_rows) == 1 + assert log_rows[0] == { + "job_id": job_id, + "event_id": event_id, + "attempt_number": 2, + "response_status_code": 503, + "response_body": "late retry callback", + "error_message": "late retry callback", + "qstash_message_id": "qstash-message-late-retry", + } + + @pytest.mark.asyncio async def test_should_mark_the_matching_event_failed_and_persist_the_error_on_failure_callback( api_client_factory: Callable[[], AbstractAsyncContextManager[AsyncClient]], diff --git a/apps/worker/app/core/tasks/webhook_tasks.py b/apps/worker/app/core/tasks/webhook_tasks.py index 55c5b80c4..0e046eca7 100644 --- a/apps/worker/app/core/tasks/webhook_tasks.py +++ b/apps/worker/app/core/tasks/webhook_tasks.py @@ -5,6 +5,10 @@ published through QStash, which owns retries and callback handling. """ +from datetime import datetime, timezone +from typing import Any +from uuid import NAMESPACE_URL, uuid5 + from loguru import logger from shared.core.celery_app import get_celery_app @@ -15,16 +19,93 @@ # Matches the beat_schedule period in celery_app.py _WEBHOOK_RECOVERY_PERIOD_SECONDS = 1800 +_WEBHOOK_CALLBACK_TEXT_LIMIT = 4096 celery_app = get_celery_app() +def _build_reconciliation_log_idempotency_key( + qstash_message_id: str, + event_id: str, + status: str, +) -> str: + """Build a fixed-width idempotency key for reconstructed QStash logs.""" + return str(uuid5(NAMESPACE_URL, f"{qstash_message_id}:{event_id}:{status}")) + + +def _truncate_callback_text(value: str | None) -> str | None: + """Trim QStash log text to the database column limit used by callbacks.""" + if not value: + return None + + return value[:_WEBHOOK_CALLBACK_TEXT_LIMIT] + + +def _reconcile_stale_delivering_events( + db: Any, + publisher: Any, + cutoff_time: datetime, +) -> int: + """Reconcile stale delivering events whose QStash message is terminal.""" + from sqlalchemy import select as sa_select + + from shared.models.database.webhook import WebhookEvent, WebhookEventStatus + from shared.models.database.webhook_log import WebhookLog + + result = db.execute( + sa_select(WebhookEvent) + .where( + WebhookEvent.status == WebhookEventStatus.DELIVERING, + WebhookEvent.qstash_message_id.is_not(None), + WebhookEvent.updated_at < cutoff_time, + ) + .limit(100) + ) + stale_events = result.scalars().all() + reconciled = 0 + + for event in stale_events: + qstash_message_id = str(event.qstash_message_id) + delivery_status = publisher.get_terminal_delivery_status(qstash_message_id) + if delivery_status is None: + continue + + event.status = delivery_status.status + event.updated_at = datetime.now(timezone.utc).replace(tzinfo=None) + + db.add( + WebhookLog( + job_id=event.job_id, + event_id=event.id, + webhook_url=event.target_url, + attempt_number=max(event.attempts, 1), + request_payload=event.payload, + signature="", + idempotency_key=_build_reconciliation_log_idempotency_key( + qstash_message_id, + event.id, + delivery_status.status, + ), + response_status_code=delivery_status.response_status_code, + response_body=_truncate_callback_text(delivery_status.response_body), + error_message=_truncate_callback_text(delivery_status.error_message), + duration_ms=0, + delivery_provider="qstash", + qstash_message_id=qstash_message_id, + ) + ) + reconciled += 1 + + return reconciled + + @celery_app.task(name="app.core.tasks.webhook_tasks.recover_orphaned_webhooks") def recover_orphaned_webhooks() -> dict: """Periodic task to recover orphaned webhook events. Finds PENDING events with attempts=0 older than 5 minutes and republishes - them via QStash. + them via QStash. Also reconciles stale DELIVERING events when QStash logs + show a terminal result but the callback did not update the database. """ from datetime import datetime, timedelta, timezone @@ -48,6 +129,7 @@ def recover_orphaned_webhooks() -> dict: minutes=age_minutes ) recovered = 0 + reconciled = 0 publisher = get_qstash_webhook_publisher() try: @@ -81,14 +163,27 @@ def recover_orphaned_webhooks() -> dict: except Exception as e: logger.error(f"Error recovering webhook event {event.id}: {e}") + reconciled = _reconcile_stale_delivering_events( + db, + publisher, + cutoff_time, + ) + if recovered > 0: logger.bind(event=LogEvent.WORKER_TASK_COMPLETE.value).info( f"Recovered {recovered} orphaned webhook events via QStash" ) - else: - logger.debug("No orphaned webhook events found") + if reconciled > 0: + logger.bind(event=LogEvent.WORKER_TASK_COMPLETE.value).info( + f"Reconciled {reconciled} stale webhook events from QStash logs" + ) + if recovered == 0 and reconciled == 0: + logger.debug("No orphaned or stale webhook events found") - return {"status": "success", "recovered": recovered, "provider": "qstash"} + result = {"status": "success", "recovered": recovered, "provider": "qstash"} + if reconciled: + result["reconciled"] = reconciled + return result except Exception as e: logger.error(f"Orphaned webhook recovery job failed: {e}", exc_info=True) diff --git a/apps/worker/tests/contract/test_webhook_recovery_contract.py b/apps/worker/tests/contract/test_webhook_recovery_contract.py index dc9266886..b30633b9f 100644 --- a/apps/worker/tests/contract/test_webhook_recovery_contract.py +++ b/apps/worker/tests/contract/test_webhook_recovery_contract.py @@ -34,6 +34,8 @@ def _insert_webhook_event( status: str, attempts: int, created_at: datetime, + updated_at: datetime | None = None, + qstash_message_id: str | None = None, ) -> None: connection.execute( text( @@ -71,9 +73,9 @@ def _insert_webhook_event( "status": status, "attempts": attempts, "next_retry_at": None, - "qstash_message_id": None, + "qstash_message_id": qstash_message_id, "created_at": created_at, - "updated_at": created_at, + "updated_at": updated_at or created_at, }, ) @@ -103,11 +105,13 @@ def test_should_republish_only_orphaned_pending_webhook_events_and_persist_qstas retried_event_id = str(uuid4()) terminal_event_id = str(uuid4()) published_message_ids: list[str] = [] + published_calls: list[dict[str, Any]] = [] publisher = qstash_publisher.QStashWebhookPublisher() class FakeMessageClient: def publish(self, **kwargs: Any) -> SimpleNamespace: + published_calls.append(kwargs) message_id = f"msg_{kwargs['headers']['X-Knowhere-Event-ID']}" published_message_ids.append(message_id) return SimpleNamespace(message_id=message_id) @@ -199,6 +203,8 @@ def publish(self, **kwargs: Any) -> SimpleNamespace: "provider": "qstash", } assert published_message_ids == [f"msg_{orphaned_event_id}"] + assert published_calls[0]["deduplication_id"] == orphaned_event_id + assert published_calls[0]["label"] == "knowhere-webhook" with engine.begin() as connection: event_rows = connection.execute( @@ -265,6 +271,133 @@ def publish(self, **kwargs: Any) -> SimpleNamespace: assert secrets_count_row["secrets_count"] == 1 +def test_should_reconcile_stale_delivering_webhook_events_from_qstash_logs( + worker_contract_environment: None, + monkeypatch: MonkeyPatch, +) -> None: + webhook_tasks, qstash_publisher, engine = _load_worker_modules() + + user_id = f"worker-user-{uuid4().hex[:12]}" + target_url = "https://hooks.contract.test/worker" + job_id = f"job_stale_{uuid4().hex[:12]}" + event_id = str(uuid4()) + qstash_message_id = f"msg_{event_id}" + + class FakePublisher: + def publish_event(self, event_id: str) -> None: + raise AssertionError(f"stale delivering event should not republish: {event_id}") + + def get_terminal_delivery_status( + self, + message_id: str, + ) -> Any: + assert message_id == qstash_message_id + return qstash_publisher.QStashDeliveryStatus( + status="delivered", + response_status_code=204, + response_body="", + error_message=None, + ) + + monkeypatch.setattr( + qstash_publisher, + "get_qstash_webhook_publisher", + lambda: FakePublisher(), + ) + + now = _utc_now() + with engine.begin() as connection: + insert_contract_user(connection, user_id=user_id) + insert_contract_job( + connection, + job_id=job_id, + user_id=user_id, + status="done", + source_type="file", + webhook_url=target_url, + webhook_enabled=True, + job_metadata=_build_file_job_metadata(), + billing_status="charged", + ) + _insert_webhook_event( + connection, + event_id=event_id, + job_id=job_id, + target_url=target_url, + status="delivering", + attempts=2, + created_at=now - timedelta(minutes=20), + updated_at=now - timedelta(minutes=10), + qstash_message_id=qstash_message_id, + ) + + result = webhook_tasks.recover_orphaned_webhooks() + + assert result == { + "status": "success", + "recovered": 0, + "provider": "qstash", + "reconciled": 1, + } + + with engine.begin() as connection: + event_row = ( + connection.execute( + text( + """ + SELECT id, status, attempts, qstash_message_id + FROM webhook_events + WHERE id = :event_id + """ + ), + {"event_id": event_id}, + ) + .mappings() + .one() + ) + log_row = ( + connection.execute( + text( + """ + SELECT + job_id, + event_id, + webhook_url, + attempt_number, + response_status_code, + response_body, + error_message, + delivery_provider, + qstash_message_id + FROM webhook_logs + WHERE event_id = :event_id + """ + ), + {"event_id": event_id}, + ) + .mappings() + .one() + ) + + assert dict(event_row) == { + "id": event_id, + "status": "delivered", + "attempts": 2, + "qstash_message_id": qstash_message_id, + } + assert dict(log_row) == { + "job_id": job_id, + "event_id": event_id, + "webhook_url": target_url, + "attempt_number": 2, + "response_status_code": 204, + "response_body": None, + "error_message": None, + "delivery_provider": "qstash", + "qstash_message_id": qstash_message_id, + } + + def test_should_skip_duplicate_beat_firing_for_webhook_recovery( worker_contract_environment: None, ) -> None: diff --git a/packages/shared-python/shared/core/config/qstash.py b/packages/shared-python/shared/core/config/qstash.py index 33dfb4bb1..04372e80a 100644 --- a/packages/shared-python/shared/core/config/qstash.py +++ b/packages/shared-python/shared/core/config/qstash.py @@ -12,6 +12,10 @@ class QStashConfig(BaseModel): # QStash API credentials (from Upstash console) QSTASH_TOKEN: Optional[str] = Field(default=None, description="QStash API token") + QSTASH_BASE_URL: Optional[str] = Field( + default=None, + description="QStash API base URL override for local development and tests", + ) QSTASH_CURRENT_SIGNING_KEY: Optional[str] = Field( default=None, description="QStash current signing key for callback verification" ) diff --git a/packages/shared-python/shared/services/webhook/qstash_publisher.py b/packages/shared-python/shared/services/webhook/qstash_publisher.py index a3c52659c..8150100b2 100644 --- a/packages/shared-python/shared/services/webhook/qstash_publisher.py +++ b/packages/shared-python/shared/services/webhook/qstash_publisher.py @@ -15,6 +15,7 @@ import hmac import json import time +from dataclasses import dataclass from typing import Any, Dict, Optional from loguru import logger @@ -27,6 +28,16 @@ ) +@dataclass(frozen=True) +class QStashDeliveryStatus: + """Terminal delivery status observed from QStash logs.""" + + status: str + response_status_code: Optional[int] + response_body: Optional[str] + error_message: Optional[str] + + class QStashWebhookPublisher: """Publishes webhook events to customer endpoints via QStash.""" @@ -55,7 +66,7 @@ def _get_client(self) -> Any: operation="initialize_client", ) - self._client = QStash(token) + self._client = QStash(token, base_url=app_config.QSTASH_BASE_URL) return self._client def publish_event(self, event_id: str) -> Optional[str]: @@ -189,6 +200,8 @@ def _publish_to_qstash( "retry_delay": retry_delay_expression, "callback": callback_url, "failure_callback": failure_callback_url, + "deduplication_id": event_id, + "label": "knowhere-webhook", } response = client.message.publish(**publish_kwargs) @@ -199,6 +212,45 @@ def _publish_to_qstash( return message_id + def get_terminal_delivery_status( + self, + qstash_message_id: str, + ) -> Optional[QStashDeliveryStatus]: + """Read QStash logs for a terminal destination delivery state.""" + try: + from qstash.log import LogState + + response = self._get_client().log.list( + filter={"message_id": qstash_message_id}, + count=20, + ) + except Exception as exc: + logger.warning( + f"QStash delivery status lookup failed: " + f"message_id={qstash_message_id}, error={exc}" + ) + return None + + terminal_logs = sorted(response.logs, key=lambda log: log.time, reverse=True) + for log in terminal_logs: + if log.state == LogState.DELIVERED: + return QStashDeliveryStatus( + status=WebhookEventStatus.DELIVERED, + response_status_code=log.response_status, + response_body=log.response_body, + error_message=log.error, + ) + + if log.state == LogState.FAILED: + return QStashDeliveryStatus( + status=WebhookEventStatus.FAILED, + response_status_code=log.response_status, + response_body=log.response_body, + error_message=log.error, + ) + + return None + def _enrich_payload(self, db: Any, event: Any) -> Dict[str, Any]: """Enrich the webhook payload (e.g., generate fresh presigned S3 URL).""" from sqlalchemy import select