[CVAT] CVAT webhook queue (#3558)

Author: zhiltsov-max

packages/examples/cvat/exchange-oracle/alembic/versions/1757437703_add_cvat_webhooks_queue_216af22b5590.py

@@ -0,0 +1,70 @@
+"""
+Add CVAT webhook queue
+
+Revision ID: 216af22b5590
+Revises: c32b36a87539
+Create Date: 2025-09-09 20:08:23.474046
+
+"""
+
+import sqlalchemy as sa
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "216af22b5590"
+down_revision = "c32b36a87539"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic ###
+    op.create_table(
+        "cvat_webhooks",
+        sa.Column(
+            "id",
+            sa.UUID(as_uuid=False),
+            server_default=sa.text("uuid_generate_v4()"),
+            nullable=False,
+        ),
+        sa.Column("cvat_project_id", sa.Integer(), nullable=False),
+        sa.Column("cvat_task_id", sa.Integer(), nullable=False),
+        sa.Column("cvat_job_id", sa.Integer(), nullable=False),
+        sa.Column("status", sa.String(), server_default="pending", nullable=True),
+        sa.Column("attempts", sa.Integer(), server_default="0", nullable=True),
+        sa.Column(
+            "created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=True
+        ),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column(
+            "wait_until", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=True
+        ),
+        sa.Column("event_type", sa.String(), nullable=False),
+        sa.Column("event_data", sa.JSON(), nullable=True),
+        sa.ForeignKeyConstraint(["cvat_job_id"], ["jobs.cvat_id"], ondelete="CASCADE"),
+        sa.ForeignKeyConstraint(["cvat_project_id"], ["projects.cvat_id"], ondelete="CASCADE"),
+        sa.ForeignKeyConstraint(["cvat_task_id"], ["tasks.cvat_id"], ondelete="CASCADE"),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_index(
+        op.f("ix_cvat_webhooks_cvat_job_id"), "cvat_webhooks", ["cvat_job_id"], unique=False
+    )
+    op.create_index(
+        op.f("ix_cvat_webhooks_cvat_project_id"), "cvat_webhooks", ["cvat_project_id"], unique=False
+    )
+    op.create_index(
+        op.f("ix_cvat_webhooks_cvat_task_id"), "cvat_webhooks", ["cvat_task_id"], unique=False
+    )
+    op.create_index(op.f("ix_cvat_webhooks_id"), "cvat_webhooks", ["id"], unique=False)
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic ###
+    op.drop_index(op.f("ix_cvat_webhooks_id"), table_name="cvat_webhooks")
+    op.drop_index(op.f("ix_cvat_webhooks_cvat_task_id"), table_name="cvat_webhooks")
+    op.drop_index(op.f("ix_cvat_webhooks_cvat_project_id"), table_name="cvat_webhooks")
+    op.drop_index(op.f("ix_cvat_webhooks_cvat_job_id"), table_name="cvat_webhooks")
+    op.drop_table("cvat_webhooks")
+    # ### end Alembic commands ###

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

@@ -98,7 +98,11 @@ def patched_get_escrow(chain_id: int, escrow_address: str) -> EscrowData:
         logger.info(f"DEV: Using local manifest '{manifest_file}' for escrow '{escrow_address}'")
         return escrow
 
-    with mock.patch.object(EscrowUtils, "get_escrow", patched_get_escrow):
+    with (
+        mock.patch.object(EscrowUtils, "get_escrow", patched_get_escrow),
+        mock.patch("src.chain.escrow.get_token_symbol", return_value="HMT"),
+        mock.patch("src.chain.web3.get_token_symbol", return_value="HMT"),
+    ):
         yield
 
 

packages/examples/cvat/exchange-oracle/src/.env.template

@@ -49,6 +49,9 @@ PROCESS_RECORDING_ORACLE_WEBHOOKS_INT=
 PROCESS_RECORDING_ORACLE_WEBHOOKS_CHUNK_SIZE=
 PROCESS_REPUTATION_ORACLE_WEBHOOKS_CHUNK_SIZE=
 PROCESS_REPUTATION_ORACLE_WEBHOOKS_INT=
+PROCESS_CVAT_WEBHOOKS_WORKERS=
+PROCESS_CVAT_WEBHOOKS_INT=
+PROCESS_CVAT_WEBHOOKS_CHUNK_SIZE=
 TRACK_COMPLETED_PROJECTS_INT=
 TRACK_COMPLETED_TASKS_INT=
 TRACK_COMPLETED_ESCROWS_INT=

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

@@ -126,27 +126,42 @@ class CronConfig:
     process_job_launcher_webhooks_chunk_size = int(
         getenv("PROCESS_JOB_LAUNCHER_WEBHOOKS_CHUNK_SIZE", 5)
     )
+
     process_recording_oracle_webhooks_int = int(getenv("PROCESS_RECORDING_ORACLE_WEBHOOKS_INT", 30))
     process_recording_oracle_webhooks_chunk_size = int(
         getenv("PROCESS_RECORDING_ORACLE_WEBHOOKS_CHUNK_SIZE", 5)
     )
+
     process_reputation_oracle_webhooks_chunk_size = int(
         getenv("PROCESS_REPUTATION_ORACLE_WEBHOOKS_CHUNK_SIZE", 5)
     )
     process_reputation_oracle_webhooks_int = int(
         getenv("PROCESS_REPUTATION_ORACLE_WEBHOOKS_INT", 5)
     )
+
+    process_cvat_webhooks_workers = int(getenv("PROCESS_CVAT_WEBHOOKS_WORKERS", 10))
+    """
+    The maximum number of parallel workers. Workers are added lazily, if the existing workers
+    can't finish in time.
+    """
+    process_cvat_webhooks_int = int(getenv("PROCESS_CVAT_WEBHOOKS_INT", 5))
+    process_cvat_webhooks_chunk_size = int(getenv("PROCESS_CVAT_WEBHOOKS_CHUNK_SIZE", 10))
+
     track_completed_projects_int = int(getenv("TRACK_COMPLETED_PROJECTS_INT", 30))
     track_completed_tasks_int = int(getenv("TRACK_COMPLETED_TASKS_INT", 30))
+
     track_creating_tasks_int = int(getenv("TRACK_CREATING_TASKS_INT", 300))
     track_creating_tasks_chunk_size = getenv("TRACK_CREATING_TASKS_CHUNK_SIZE", 5)
+
     track_assignments_int = int(getenv("TRACK_ASSIGNMENTS_INT", 5))
     track_assignments_chunk_size = int(getenv("TRACK_ASSIGNMENTS_CHUNK_SIZE", 10))
 
     track_completed_escrows_int = int(getenv("TRACK_COMPLETED_ESCROWS_INT", 60))
     track_completed_escrows_chunk_size = int(getenv("TRACK_COMPLETED_ESCROWS_CHUNK_SIZE", 100))
+
     track_escrow_validations_int = int(getenv("TRACK_ESCROW_VALIDATIONS_INT", 60))
     track_escrow_validations_chunk_size = int(getenv("TRACK_ESCROW_VALIDATIONS_CHUNK_SIZE", 1))
+
     track_completed_escrows_max_downloading_retries = int(
         getenv("TRACK_COMPLETED_ESCROWS_MAX_DOWNLOADING_RETRIES", 10)
     )

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

@@ -75,6 +75,12 @@ class OracleWebhookStatuses(str, Enum, metaclass=BetterEnumMeta):
     failed = "failed"
 
 
+class CvatWebhookStatuses(str, Enum, metaclass=BetterEnumMeta):
+    pending = "pending"
+    completed = "completed"
+    failed = "failed"
+
+
 class AssignmentStatuses(str, Enum, metaclass=BetterEnumMeta):
     """
     State changes:

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

@@ -3,6 +3,7 @@
 
 from src.core.config import Config
 from src.crons.cvat.state_trackers import (
+    process_incoming_cvat_webhooks,
     track_assignments,
     track_completed_escrows,
     track_completed_projects,
@@ -57,6 +58,12 @@ def cron_record():
             "interval",
             seconds=Config.cron_config.process_reputation_oracle_webhooks_int,
         )
+        scheduler.add_job(
+            process_incoming_cvat_webhooks,
+            trigger="interval",
+            seconds=Config.cron_config.process_cvat_webhooks_int,
+            max_instances=Config.cron_config.process_cvat_webhooks_workers,
+        )
         scheduler.add_job(
             track_completed_projects,
             "interval",

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

@@ -1,4 +1,5 @@
 import logging
+from contextlib import suppress
 
 from sqlalchemy.orm import Session
 
@@ -15,7 +16,30 @@
 from src.db import errors as db_errors
 from src.db.utils import ForUpdateParams
 from src.handlers.completed_escrows import handle_escrows_validations
+from src.handlers.cvat_events import cvat_webhook_handler
 from src.utils.logging import format_sequence
+from src.utils.time import utcnow
+
+
+@cron_job
+def process_incoming_cvat_webhooks(logger: logging.Logger, session: Session) -> None:
+    webhooks = cvat_service.incoming_webhooks.get_pending_webhooks(
+        session=session,
+        limit=CronConfig.process_cvat_webhooks_chunk_size,
+        for_update=ForUpdateParams(skip_locked=True),
+    )
+
+    for webhook in webhooks:
+        try:
+            with session.begin_nested():
+                cvat_webhook_handler(webhook, session)
+                cvat_service.incoming_webhooks.handle_webhook_success(
+                    session, webhook_id=webhook.id
+                )
+        except Exception as e:
+            with session.begin_nested():
+                logger.exception(e)
+                cvat_service.incoming_webhooks.handle_webhook_fail(session, webhook_id=webhook.id)
 
 
 @cron_job
@@ -53,18 +77,51 @@ def track_assignments(logger: logging.Logger) -> None:
     4. If a project or task state is not "annotation", cancels assignments
     """
 
+    def _try_complete_assignment(
+        session: Session,
+        assignment: cvat_models.Assignment,
+    ) -> bool:
+        """
+        Checks if we haven't received a notification, but the job might have been completed.
+
+        Returns: assignment completed
+        """
+
+        latest_assignment = cvat_service.get_latest_assignment_by_cvat_job_id(
+            session, assignment.cvat_job_id
+        )
+        if not latest_assignment or latest_assignment.id != assignment.id:
+            return False
+
+        try:
+            cvat_job = cvat_api.get_job(assignment.cvat_job_id)
+        except cvat_api.exceptions.NotFoundException:
+            return False
+
+        if cvat_job.state != cvat_api.JobStatus.completed:
+            return False
+
+        if not cvat_job.assignee or cvat_job.assignee.id != latest_assignment.user.cvat_id:
+            return False
+
+        logger.info(f"Found completed job #{assignment.cvat_job_id}. Completing the assignment")
+        cvat_service.complete_assignment(session, assignment.id, completed_at=utcnow())
+        cvat_api.update_job_assignee(assignment.cvat_job_id, assignee_id=None)
+        cvat_service.update_job_status(session, assignment.job.id, status=JobStatuses.completed)
+        cvat_service.touch(session, cvat_models.Job, [assignment.job.id])
+        return True
+
     def _reset_job_after_assignment(session: Session, assignment: cvat_models.Assignment):
         latest_assignment = cvat_service.get_latest_assignment_by_cvat_job_id(
             session, assignment.cvat_job_id
         )
-        if latest_assignment.id == assignment.id:
+        if latest_assignment.id != assignment.id:
             # Avoid un-assigning if it's not the latest assignment
+            return
 
-            cvat_api.update_job_assignee(
-                assignment.cvat_job_id, assignee_id=None
-            )  # note that calling it in a loop can take too much time
-
-            cvat_service.update_job_status(session, assignment.job.id, status=JobStatuses.new)
+        cvat_api.update_job_assignee(assignment.cvat_job_id, assignee_id=None)
+        cvat_service.update_job_status(session, assignment.job.id, status=JobStatuses.new)
+        cvat_service.touch(session, cvat_models.Job, [assignment.job.id])
 
     with SessionLocal.begin() as session:
         assignments = cvat_service.get_unprocessed_expired_assignments(
@@ -74,18 +131,27 @@ def _reset_job_after_assignment(session: Session, assignment: cvat_models.Assign
         )
 
         for assignment in assignments:
-            logger.info(
-                "Expiring the unfinished assignment {} (user {}, job id {})".format(
-                    assignment.id,
-                    assignment.user_wallet_address,
-                    assignment.cvat_job_id,
+            with (
+                session.begin_nested(),
+                suppress(db_errors.LockNotAvailable),
+            ):
+                cvat_service.get_jobs_by_cvat_id(
+                    session,
+                    cvat_ids=[assignment.cvat_job_id],
+                    for_update=ForUpdateParams(nowait=True),
                 )
-            )
 
-            cvat_service.expire_assignment(session, assignment.id)
-            _reset_job_after_assignment(session, assignment)
+                if not _try_complete_assignment(session, assignment):
+                    logger.info(
+                        "Expiring the unfinished assignment {} (user {}, job id {})".format(
+                            assignment.id,
+                            assignment.user_wallet_address,
+                            assignment.cvat_job_id,
+                        )
+                    )
 
-        cvat_service.touch(session, cvat_models.Job, [a.job.id for a in assignments])
+                    cvat_service.expire_assignment(session, assignment.id)
+                    _reset_job_after_assignment(session, assignment)
 
     with SessionLocal.begin() as session:
         assignments = cvat_service.get_unprocessed_cancelled_assignments(
@@ -95,16 +161,24 @@ def _reset_job_after_assignment(session: Session, assignment: cvat_models.Assign
         )
 
         for assignment in assignments:
-            logger.info(
-                "Finalizing the canceled assignment {} (user {}, job id {})".format(
-                    assignment.id,
-                    assignment.user_wallet_address,
-                    assignment.cvat_job_id,
+            with (
+                session.begin_nested(),
+                suppress(db_errors.LockNotAvailable),
+            ):
+                cvat_service.get_jobs_by_cvat_id(
+                    session,
+                    cvat_ids=[assignment.cvat_job_id],
+                    for_update=ForUpdateParams(nowait=True),
                 )
-            )
-            _reset_job_after_assignment(session, assignment)
 
-        cvat_service.touch(session, cvat_models.Job, [a.job.id for a in assignments])
+                logger.info(
+                    "Finalizing the canceled assignment {} (user {}, job id {})".format(
+                        assignment.id,
+                        assignment.user_wallet_address,
+                        assignment.cvat_job_id,
+                    )
+                )
+                _reset_job_after_assignment(session, assignment)
 
     with SessionLocal.begin() as session:
         assignments = cvat_service.get_active_assignments(
@@ -115,19 +189,27 @@ def _reset_job_after_assignment(session: Session, assignment: cvat_models.Assign
 
         for assignment in assignments:
             if assignment.job.project.status != ProjectStatuses.annotation:
-                logger.warning(
-                    "Canceling the unfinished assignment {} (user {}, job id {}) - "
-                    "the project state is not annotation".format(
-                        assignment.id,
-                        assignment.user_wallet_address,
-                        assignment.cvat_job_id,
+                with (
+                    session.begin_nested(),
+                    suppress(db_errors.LockNotAvailable),
+                ):
+                    cvat_service.get_jobs_by_cvat_id(
+                        session,
+                        cvat_ids=[assignment.cvat_job_id],
+                        for_update=ForUpdateParams(nowait=True),
                     )
-                )
 
-                cvat_service.cancel_assignment(session, assignment.id)
-                _reset_job_after_assignment(session, assignment)
+                    logger.warning(
+                        "Canceling the unfinished assignment {} (user {}, job id {}) - "
+                        "the project state is not annotation".format(
+                            assignment.id,
+                            assignment.user_wallet_address,
+                            assignment.cvat_job_id,
+                        )
+                    )
 
-        cvat_service.touch(session, cvat_models.Job, [a.job.id for a in assignments])
+                    cvat_service.cancel_assignment(session, assignment.id)
+                    _reset_job_after_assignment(session, assignment)
 
 
 @cron_job