[CVAT] Fix moving projects to validation (#2852)

* Fix a possible deadlock in create_escrow_validations

* Improve log message

* Don't expect projects in validation status in completed escrows

* Use limit from config

* USe a separate env var

* Polish some code

* Mark escrow validation completed after RO rejection webhook

* Fix potential deadlock on escrow validation inserts

* Satisfy linter

* Remove backward compatibility for env var

* Update tests, skip ro validation webhooks if no active validations

* Add db type check

Author: zhiltsov-max

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

@@ -49,6 +49,8 @@ TRACK_COMPLETED_TASKS_INT=
 TRACK_COMPLETED_TASKS_CHUNK_SIZE=
 TRACK_COMPLETED_ESCROWS_INT=
 TRACK_COMPLETED_ESCROWS_CHUNK_SIZE=
+TRACK_ESCROW_VALIDATIONS_INT=
+TRACK_ESCROW_VALIDATIONS_CHUNK_SIZE=
 TRACK_CREATING_TASKS_INT=
 TRACK_CREATING_TASKS_CHUNK_SIZE=
 TRACK_ASSIGNMENTS_INT=

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

@@ -142,15 +142,12 @@ class CronConfig:
             "TRACK_COMPLETED_ESCROWS_INT", os.environ.get("RETRIEVE_ANNOTATIONS_INT", 60)
         )
     )
-    track_escrow_validations_int = int(
-        os.environ.get(
-            "TRACK_COMPLETED_ESCROWS_INT", os.environ.get("RETRIEVE_ANNOTATIONS_INT", 60)
-        )
+    track_completed_escrows_chunk_size = int(
+        os.environ.get("TRACK_COMPLETED_ESCROWS_CHUNK_SIZE", 100)
     )
-    track_completed_escrows_chunk_size = os.environ.get(
-        # backward compatibility
-        "TRACK_COMPLETED_ESCROWS_CHUNK_SIZE",
-        os.environ.get("RETRIEVE_ANNOTATIONS_CHUNK_SIZE", 5),
+    track_escrow_validations_int = int(os.environ.get("TRACK_COMPLETED_ESCROWS_INT", 60))
+    track_escrow_validations_chunk_size = int(
+        os.environ.get("TRACK_ESCROW_VALIDATIONS_CHUNK_SIZE", 5)
     )
     track_completed_escrows_max_downloading_retries = int(
         os.environ.get("TRACK_COMPLETED_ESCROWS_MAX_DOWNLOADING_RETRIES", 10)

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

@@ -69,7 +69,7 @@ def cron_record():
         scheduler.add_job(
             track_escrow_validations,
             "interval",
-            seconds=Config.cron_config.track_completed_escrows_int,
+            seconds=Config.cron_config.track_escrow_validations_int,
         )
         scheduler.add_job(
             track_task_creation,

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

@@ -116,12 +116,13 @@ def track_assignments(logger: logging.Logger) -> None:
 
 @cron_job
 def track_completed_escrows(logger: logging.Logger, session: Session) -> None:
-    awaiting_validations = cvat_service.create_escrow_validations(session)
-    if awaiting_validations:
-        session.commit()
+    new_validations = cvat_service.create_escrow_validations(
+        session, limit=CronConfig.track_completed_escrows_chunk_size
+    )
+    if new_validations:
         logger.info(
-            f"Got {len(awaiting_validations)} escrows "
-            f"awaiting validation: {format_sequence(awaiting_validations)}"
+            f"Got {len(new_validations)} escrows "
+            f"awaiting validation: {format_sequence([(v[1], v[2]) for v in new_validations])}"
         )
 
 

packages/examples/cvat/exchange-oracle/src/crons/webhooks/recording_oracle.py

@@ -44,13 +44,29 @@ def handle_recording_oracle_event(webhook: Webhook, *, db_session: Session, logg
 
     match webhook.event_type:
         case RecordingOracleEventTypes.job_completed:
+            escrow_validation = cvat_db_service.get_escrow_validation_by_escrow_address(
+                db_session, escrow_address=webhook.escrow_address, chain_id=webhook.chain_id
+            )
+            if (
+                not escrow_validation
+                or escrow_validation.status != EscrowValidationStatuses.in_progress
+            ):
+                logger.error(
+                    "Unexpected event {} received for the escrow. "
+                    "There is no active validation now, ignoring (escrow_address={}) ".format(
+                        webhook.event_type,
+                        webhook.escrow_address,
+                    )
+                )
+                return
+
             chunk_size = CronConfig.accepted_projects_chunk_size
             project_ids = cvat_db_service.get_project_cvat_ids_by_escrow_address(
                 db_session, webhook.escrow_address
             )
             if not project_ids:
                 logger.error(
-                    f"Unexpected event {webhook.event_type} received for an unknown project, "
+                    f"Unexpected event {webhook.event_type} received for an unknown escrow, "
                     f"ignoring (escrow_address={webhook.escrow_address})"
                 )
                 return
@@ -98,6 +114,22 @@ def handle_recording_oracle_event(webhook: Webhook, *, db_session: Session, logg
         case RecordingOracleEventTypes.submission_rejected:
             event = RecordingOracleEvent_SubmissionRejected.model_validate(webhook.event_data)
 
+            escrow_validation = cvat_db_service.get_escrow_validation_by_escrow_address(
+                db_session, escrow_address=webhook.escrow_address, chain_id=webhook.chain_id
+            )
+            if (
+                not escrow_validation
+                or escrow_validation.status != EscrowValidationStatuses.in_progress
+            ):
+                logger.error(
+                    "Unexpected event {} received for the escrow. "
+                    "There is no active validation now, ignoring (escrow_address={}) ".format(
+                        webhook.event_type,
+                        webhook.escrow_address,
+                    )
+                )
+                return
+
             rejected_assignments = cvat_db_service.get_assignments_by_id(
                 db_session, [t.assignment_id for t in event.assignments]
             )
@@ -152,6 +184,26 @@ def handle_recording_oracle_event(webhook: Webhook, *, db_session: Session, logg
                     )
                     cvat_db_service.update_project_status(db_session, project.id, new_status)
 
+            # Mark all the remaining projects completed
+            cvat_db_service.update_project_statuses_by_escrow_address(
+                db_session,
+                escrow_address=webhook.escrow_address,
+                chain_id=webhook.chain_id,
+                status=ProjectStatuses.completed,
+                included_statuses=[ProjectStatuses.validation],
+            )
+
+            # TODO: need to update assignments,
+            # but there is no special db state for validated assignments
+
+            # TODO: maybe delete instead
+            cvat_db_service.update_escrow_validation(
+                db_session,
+                webhook.escrow_address,
+                webhook.chain_id,
+                status=EscrowValidationStatuses.completed,
+            )
+
         case _:
             raise AssertionError(f"Unknown recording oracle event {webhook.event_type}")
 

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

@@ -40,7 +40,6 @@ class BaseUUID(Base):
         # At some point it would make sense to use UUID(as_uuid=True).
         UUID(as_uuid=False),
         primary_key=True,
-        default=lambda: str(uuid4()),
         server_default=func.uuid_generate_v4(),
         sort_order=-1,  # Make sure it's the first column.
         index=True,

packages/examples/cvat/exchange-oracle/src/handlers/completed_escrows.py

@@ -17,7 +17,7 @@
 from src.core.config import CronConfig, StorageConfig
 from src.core.oracle_events import ExchangeOracleEvent_JobFinished
 from src.core.storage import compose_results_bucket_filename
-from src.core.types import EscrowValidationStatuses, OracleWebhookTypes, ProjectStatuses, TaskTypes
+from src.core.types import EscrowValidationStatuses, OracleWebhookTypes, TaskTypes
 from src.db import SessionLocal
 from src.db import errors as db_errors
 from src.db.utils import ForUpdateParams
@@ -138,13 +138,6 @@ def _export_escrow_annotations(
         event=ExchangeOracleEvent_JobFinished(),
     )
 
-    cvat_service.update_project_statuses_by_escrow_address(
-        session,
-        escrow_address=escrow_address,
-        chain_id=chain_id,
-        status=ProjectStatuses.validation,
-    )
-
     logger.info(
         f"The escrow ({escrow_address=}) is completed, "
         f"resulting annotations are processed successfully"
@@ -270,8 +263,9 @@ def handle_escrows_validations(logger: logging.Logger) -> None:
     with SessionLocal.begin() as session:
         escrow_validations = cvat_service.prepare_escrows_for_validation(
             session,
-            limit=CronConfig.track_completed_escrows_chunk_size,
+            limit=CronConfig.track_escrow_validations_chunk_size,
         )
+
     for escrow_address, chain_id in escrow_validations:
         with SessionLocal.begin() as session:
             # Need to work in separate transactions for each escrow, as a failing DB call
@@ -281,6 +275,7 @@ def handle_escrows_validations(logger: logging.Logger) -> None:
             if not handled:
                 # either escrow is invalid, or we couldn't get lock for projects/validations
                 continue
+
             # change status so validation won't be attempted again
             cvat_service.update_escrow_validation(
                 session,

packages/examples/cvat/exchange-oracle/src/models/cvat.py

@@ -143,7 +143,7 @@ class EscrowValidation(BaseUUID):
     chain_id = Column(Integer, Enum(Networks), nullable=False)
 
     created_at = Column(DateTime(timezone=True), server_default=func.now())
-    attempts = Column(Integer, default=0, server_default="0")
+    attempts = Column(Integer, server_default="0")
     status = Column(String, Enum(EscrowValidationStatuses), nullable=False)
     projects: Mapped[list[Project]] = relationship(
         back_populates="escrow_validation",