[Dependent] [CVAT] Update last assignment handling

packages/examples/cvat/exchange-oracle/Dockerfile

@@ -4,13 +4,17 @@ WORKDIR /app
 
 RUN apt-get update -y && \
     apt-get install -y jq ffmpeg libsm6 libxext6 && \
-    pip install --no-cache poetry
+    rm -rf /var/lib/apt/lists/*
+
+RUN pip install --no-cache poetry
 
 COPY pyproject.toml poetry.lock ./
 
-RUN poetry config virtualenvs.create false \
-    && poetry install --no-interaction --no-ansi --no-root \
-    && poetry cache clear pypi --all
+RUN --mount=type=cache,target=/root/.cache \
+    poetry config virtualenvs.create false && \
+    poetry install --no-interaction --no-ansi --no-root
+
+RUN python -m pip uninstall -y poetry pip
 
 COPY . .
 

packages/examples/cvat/exchange-oracle/README.md

@@ -18,14 +18,14 @@ For deployment it is required to have PostgreSQL(v14.4)
 
 ### Run the oracle locally:
 
-```
+```sh
 docker compose -f docker-compose.dev.yml up -d
 ./bin/start_dev.sh
 ```
 
 or 
 
-```
+```sh
 docker compose -f docker-compose.dev.yml up -d
 ./bin/start_debug.sh
 ```
@@ -48,17 +48,17 @@ Example: [Alembic env file](https://github.com/humanprotocol/human-protocol/blob
 
 
 Adding new migration:
-```
+```sh
 alembic revision --autogenerate -m "your-migration-name"
 ```
 
 Upgrade:
-```
+```sh
 alembic upgrade head
 ```
 
 Downgrade:
-```
+```sh
 alembic downgrade -{number of migrations}
 ```
 
@@ -72,7 +72,7 @@ Available at `/docs` route
 ### Tests
 
 To run tests
-```
-docker compose -f docker-compose.test.yml up --build test --attach test --exit-code-from test && \
-   docker compose -f docker-compose.test.yml down
+```sh
+docker compose -p "test" -f docker-compose.test.yml up --build test --attach test --exit-code-from test; \
+   docker compose -p "test" -f docker-compose.test.yml down
 ```

packages/examples/cvat/exchange-oracle/dockerfiles/test.Dockerfile

@@ -4,16 +4,20 @@ WORKDIR /app
 
 RUN apt-get update -y && \
     apt-get install -y jq ffmpeg libsm6 libxext6 && \
-    pip install --no-cache poetry
+    rm -rf /var/lib/apt/lists/*
+
+RUN pip install --no-cache poetry
 
 COPY pyproject.toml poetry.lock ./
 
-RUN poetry config virtualenvs.create false \
-    && poetry install --no-interaction --no-ansi --no-root \
-    && poetry cache clear pypi --all
+RUN --mount=type=cache,target=/root/.cache \
+    poetry config virtualenvs.create false && \
+    poetry install --no-interaction --no-ansi --no-root
+
+RUN python -m pip uninstall -y poetry pip
 
 COPY . .
 
 RUN rm -f ./src/.env
 
-CMD ["pytest", "-W", "ignore::DeprecationWarning", "-v"]
+CMD ["pytest"]

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

@@ -134,6 +134,7 @@ ignore = [
     "ANN001",  # |
     "ANN003",  # |
     "ARG001",  # |
+    "FBT001",  # Allow bool-annotated positional args in functions
     "SLF001",  # Allow private attrs access
     "PLR2004", # Allow magic values
     "S",       # security

packages/examples/cvat/exchange-oracle/pytest.ini

@@ -0,0 +1,10 @@
+[pytest]
+addopts = --verbose
+filterwarnings =
+    ignore::DeprecationWarning:cvat_sdk.core
+    ignore::DeprecationWarning:human_protocol_sdk.storage
+    ignore:Field name \"sort\" shadows:UserWarning:pydantic._internal._fields
+
+python_files = test_*.py
+python_classes = *Test
+python_functions = test_*

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

@@ -10,12 +10,6 @@ class Networks(int, Enum, metaclass=BetterEnumMeta):
     localhost = Config.localhost.chain_id
 
 
-class CvatEventTypes(str, Enum, metaclass=BetterEnumMeta):
-    update_job = "update:job"
-    create_job = "create:job"
-    ping = "ping"
-
-
 class ProjectStatuses(str, Enum, metaclass=BetterEnumMeta):
     creation = "creation"
     annotation = "annotation"
@@ -34,7 +28,6 @@ class TaskStatuses(str, Enum, metaclass=BetterEnumMeta):
 class JobStatuses(str, Enum, metaclass=BetterEnumMeta):
     new = "new"
     in_progress = "in progress"
-    rejected = "rejected"
     completed = "completed"
 
 
@@ -47,13 +40,6 @@ class TaskTypes(str, Enum, metaclass=BetterEnumMeta):
     image_polygons = "image_polygons"
 
 
-class CvatLabelTypes(str, Enum, metaclass=BetterEnumMeta):
-    tag = "tag"
-    points = "points"
-    rectangle = "rectangle"
-    polygon = "polygon"
-
-
 class OracleWebhookTypes(str, Enum, metaclass=BetterEnumMeta):
     exchange_oracle = "exchange_oracle"
     job_launcher = "job_launcher"

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

@@ -49,8 +49,23 @@ def track_assignments(logger: logging.Logger) -> None:
     Tracks assignments:
     1. Checks time for each active assignment
     2. If an assignment is timed out, expires it
-    3. If a project or task state is not "annotation", cancels assignments
+    3. If an assignment is canceled, resets it
+    4. If a project or task state is not "annotation", cancels assignments
     """
+
+    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:
+            # Avoid un-assigning if it's not the latest assignment
+
+            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)
+
     with SessionLocal.begin() as session:
         assignments = cvat_service.get_unprocessed_expired_assignments(
             session,
@@ -67,17 +82,27 @@ def track_assignments(logger: logging.Logger) -> None:
                 )
             )
 
-            latest_assignment = cvat_service.get_latest_assignment_by_cvat_job_id(
-                session, assignment.cvat_job_id
-            )
-            if latest_assignment.id == assignment.id:
-                # Avoid un-assigning if it's not the latest assignment
+            cvat_service.expire_assignment(session, assignment.id)
+            _reset_job_after_assignment(session, assignment)
 
-                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.touch(session, cvat_models.Job, [a.job.id for a in assignments])
 
-            cvat_service.expire_assignment(session, assignment.id)
+    with SessionLocal.begin() as session:
+        assignments = cvat_service.get_unprocessed_cancelled_assignments(
+            session,
+            limit=CronConfig.track_assignments_chunk_size,
+            for_update=ForUpdateParams(skip_locked=True),
+        )
+
+        for assignment 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)
 
         cvat_service.touch(session, cvat_models.Job, [a.job.id for a in assignments])
 
@@ -99,17 +124,8 @@ def track_assignments(logger: logging.Logger) -> None:
                     )
                 )
 
-                latest_assignment = cvat_service.get_latest_assignment_by_cvat_job_id(
-                    session, assignment.cvat_job_id
-                )
-                if latest_assignment.id == assignment.id:
-                    # Avoid un-assigning if it's not the latest assignment
-
-                    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.cancel_assignment(session, assignment.id)
+                _reset_job_after_assignment(session, assignment)
 
         cvat_service.touch(session, cvat_models.Job, [a.job.id for a in assignments])
 

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

@@ -38,6 +38,26 @@ class RequestStatus(str, Enum, metaclass=BetterEnumMeta):
     FAILED = "Failed"
 
 
+class JobStatus(str, Enum, metaclass=BetterEnumMeta):
+    new = "new"
+    in_progress = "in progress"
+    rejected = "rejected"
+    completed = "completed"
+
+
+class LabelType(str, Enum, metaclass=BetterEnumMeta):
+    tag = "tag"
+    points = "points"
+    rectangle = "rectangle"
+    polygon = "polygon"
+
+
+class WebhookEventType(str, Enum, metaclass=BetterEnumMeta):
+    update_job = "update:job"
+    create_job = "create:job"
+    ping = "ping"
+
+
 def _request_annotations(endpoint: Endpoint, cvat_id: int, format_name: str) -> str:
     """
     Requests annotations export.