[CVAT] Oracle updates (#3219)

* Rename job_creation_failed to escrow_failed

* Update migration template

* Update cvat_sdk dependency

* Update cvat task creation status checks

* Improve matching for local manifests in debug mode

* Update annotation downloading code

Author: zhiltsov-max

packages/examples/cvat/exchange-oracle/alembic/script.py.mako

@@ -1,4 +1,5 @@
-"""${message}
+"""
+${message}
 
 Revision ID: ${up_revision}
 Revises: ${down_revision | comma,n}

packages/examples/cvat/exchange-oracle/alembic/versions/1742568844_rename_event_c32b36a87539.py

@@ -0,0 +1,49 @@
+"""
+Rename "job_creation_failed" to "escrow_failed"
+
+Revision ID: c32b36a87539
+Revises: 0a91b6a5f7b6
+Create Date: 2025-03-21 16:54:04.057456
+
+"""
+
+from sqlalchemy import Column, String, update
+from sqlalchemy.orm import declarative_base
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "c32b36a87539"
+down_revision = "0a91b6a5f7b6"
+branch_labels = None
+depends_on = None
+
+
+Base = declarative_base()
+
+old_name = "job_creation_failed"
+new_name = "escrow_failed"
+
+
+class Webhook(Base):
+    # Represents the model before the transaction is applied
+
+    __tablename__ = "webhooks"
+    id = Column(String, primary_key=True, index=True)
+    event_type = Column(String, nullable=False)
+
+
+def update_webhook_types():
+    op.execute(update(Webhook).where(Webhook.event_type == old_name).values(event_type=new_name))
+
+
+def revert_webhook_types():
+    op.execute(update(Webhook).where(Webhook.event_type == new_name).values(event_type=old_name))
+
+
+def upgrade() -> None:
+    update_webhook_types()
+
+
+def downgrade() -> None:
+    revert_webhook_types()

packages/examples/cvat/exchange-oracle/debug.py

@@ -3,7 +3,7 @@
 from collections.abc import Generator
 from contextlib import ExitStack, contextmanager
 from logging import Logger
-from pathlib import Path
+from pathlib import Path, PurePosixPath
 from typing import Any
 from unittest import mock
 
@@ -62,7 +62,9 @@ def patched_get_escrow(chain_id: int, escrow_address: str) -> EscrowData:
         minio_manifests = minio_client.list_files(bucket="manifests")
         logger.debug(f"DEV: Local manifests: {format_sequence(minio_manifests)}")
 
-        candidate_files = [fn for fn in minio_manifests if f"{escrow_address}.json" in fn]
+        candidate_files = [
+            fn for fn in minio_manifests if PurePosixPath(fn).name == f"{escrow_address}.json"
+        ]
         if not candidate_files:
             return original_get_escrow(ChainId(chain_id), escrow_address)
         elif len(candidate_files) != 1:

packages/examples/cvat/exchange-oracle/poetry.lock

@@ -16,7 +16,7 @@ sqlalchemy-utils = "^0.41.1"
 alembic = "^1.11.1"
 httpx = "^0.24.1"
 pytest = "^7.2.2"
-cvat-sdk = "2.25.0"
+cvat-sdk = "2.31.0"
 sqlalchemy = "^2.0.16"
 apscheduler = "^3.10.1"
 xmltodict = "^0.13.0"

packages/examples/cvat/exchange-oracle/pyproject.toml

@@ -38,7 +38,7 @@ class RejectedAssignmentInfo(BaseModel):
     assignments: list[RejectedAssignmentInfo]
 
 
-class ExchangeOracleEvent_JobCreationFailed(OracleEvent):
+class ExchangeOracleEvent_EscrowFailed(OracleEvent):
     # no task_id, escrow is enough for now
     reason: str
 
@@ -60,7 +60,7 @@ class ReputationOracleEvent_EscrowCompleted(OracleEvent):
     JobLauncherEventTypes.escrow_canceled: JobLauncherEvent_EscrowCanceled,
     RecordingOracleEventTypes.job_completed: RecordingOracleEvent_JobCompleted,
     RecordingOracleEventTypes.submission_rejected: RecordingOracleEvent_SubmissionRejected,
-    ExchangeOracleEventTypes.job_creation_failed: ExchangeOracleEvent_JobCreationFailed,
+    ExchangeOracleEventTypes.escrow_failed: ExchangeOracleEvent_EscrowFailed,
     ExchangeOracleEventTypes.job_finished: ExchangeOracleEvent_JobFinished,
     ExchangeOracleEventTypes.escrow_cleaned: ExchangeOracleEvent_EscrowCleaned,
     ReputationOracleEventTypes.escrow_completed: ReputationOracleEvent_EscrowCompleted,

packages/examples/cvat/exchange-oracle/src/core/oracle_events.py

@@ -62,7 +62,7 @@ class OracleWebhookTypes(str, Enum, metaclass=BetterEnumMeta):
 
 
 class ExchangeOracleEventTypes(str, Enum, metaclass=BetterEnumMeta):
-    job_creation_failed = "job_creation_failed"
+    escrow_failed = "escrow_failed"
     job_finished = "job_finished"
     escrow_cleaned = "escrow_cleaned"
 

packages/examples/cvat/exchange-oracle/src/core/types.py

@@ -8,7 +8,7 @@
 import src.services.webhook as oracle_db_service
 from src import db
 from src.core.config import CronConfig
-from src.core.oracle_events import ExchangeOracleEvent_JobCreationFailed
+from src.core.oracle_events import ExchangeOracleEvent_EscrowFailed
 from src.core.types import JobStatuses, OracleWebhookTypes, ProjectStatuses
 from src.crons._cron_job import cron_job
 from src.db import SessionLocal
@@ -159,7 +159,7 @@ def track_task_creation(logger: logging.Logger, session: Session) -> None:
     for upload in uploads:
         status, reason = cvat_api.get_task_upload_status(upload.task_id)
         project = upload.task.project
-        if not status or status == cvat_api.UploadStatus.FAILED:
+        if not status or status == cvat_api.RequestStatus.FAILED:
             # TODO: add retries if 5xx
             failed.append(upload)
 
@@ -168,9 +168,9 @@ def track_task_creation(logger: logging.Logger, session: Session) -> None:
                 escrow_address=project.escrow_address,
                 chain_id=project.chain_id,
                 type=OracleWebhookTypes.job_launcher,
-                event=ExchangeOracleEvent_JobCreationFailed(reason=reason),
+                event=ExchangeOracleEvent_EscrowFailed(reason=reason),
             )
-        elif status == cvat_api.UploadStatus.FINISHED:
+        elif status == cvat_api.RequestStatus.FINISHED:
             try:
                 cvat_jobs = cvat_api.fetch_task_jobs(upload.task_id)
 
@@ -200,7 +200,7 @@ def track_task_creation(logger: logging.Logger, session: Session) -> None:
                     escrow_address=project.escrow_address,
                     chain_id=project.chain_id,
                     type=OracleWebhookTypes.job_launcher,
-                    event=ExchangeOracleEvent_JobCreationFailed(reason=str(e)),
+                    event=ExchangeOracleEvent_EscrowFailed(reason=str(e)),
                 )
 
     if completed:

packages/examples/cvat/exchange-oracle/src/crons/cvat/state_trackers.py

@@ -11,7 +11,7 @@
 from src.core.config import Config, CronConfig
 from src.core.oracle_events import (
     ExchangeOracleEvent_EscrowCleaned,
-    ExchangeOracleEvent_JobCreationFailed,
+    ExchangeOracleEvent_EscrowFailed,
 )
 from src.core.types import JobLauncherEventTypes, Networks, OracleWebhookTypes, ProjectStatuses
 from src.crons._cron_job import cron_job
@@ -36,7 +36,7 @@ def handle_failure(session: Session, webhook: Webhook, exc: Exception) -> None:
             escrow_address=webhook.escrow_address,
             chain_id=webhook.chain_id,
             type=OracleWebhookTypes.job_launcher,
-            event=ExchangeOracleEvent_JobCreationFailed(reason=str(exc)),
+            event=ExchangeOracleEvent_EscrowFailed(reason=str(exc)),
         )