Fix dataset downloading

changelog.d/20240508_144202_mzhiltso_fix_dataset_downloading.md

@@ -0,0 +1,5 @@
+### Fixed
+
+- The 500 / "The result file does not exist in export cache" error
+  on dataset export request
+  (<https://github.com/cvat-ai/cvat/pull/7864>)

cvat/apps/dataset_manager/apps.py

@@ -0,0 +1,18 @@
+# Copyright (C) 2024 CVAT.ai Corporation
+#
+# SPDX-License-Identifier: MIT
+
+from django.apps import AppConfig
+
+
+class DatasetManagerConfig(AppConfig):
+    name = "cvat.apps.dataset_manager"
+
+    def ready(self) -> None:
+        from django.conf import settings
+
+        from . import default_settings
+
+        for key in dir(default_settings):
+            if key.isupper() and not hasattr(settings, key):
+                setattr(settings, key, getattr(default_settings, key))

cvat/apps/dataset_manager/default_settings.py

@@ -0,0 +1,11 @@
+# Copyright (C) 2024 CVAT.ai Corporation
+#
+# SPDX-License-Identifier: MIT
+
+import os
+
+DATASET_CACHE_TTL = int(os.getenv("CVAT_DATASET_CACHE_TTL", 60 * 60 * 10))
+"Base lifetime for cached exported datasets, in seconds"
+
+DATASET_CACHE_LOCK_TIMEOUT = int(os.getenv("CVAT_DATASET_CACHE_LOCK_TIMEOUT", 10))
+"Timeout for cache lock acquiring, in seconds"

cvat/apps/dataset_manager/util.py

@@ -1,16 +1,22 @@
-
 # Copyright (C) 2019-2022 Intel Corporation
+# Copyright (C) 2023-2024 CVAT.ai Corporation
 #
 # SPDX-License-Identifier: MIT
 
+from contextlib import contextmanager, suppress
 from copy import deepcopy
-from typing import Sequence
+from datetime import timedelta
+from threading import Lock
+from typing import Any, Generator, Optional, Sequence
 import inspect
-import os, os.path as osp
+import os
+import os.path as osp
 import zipfile
 
+import django_rq
 from django.conf import settings
 from django.db import models
+from pottery import Redlock, ReleaseUnlockedLock
 
 
 def current_function_name(depth=1):
@@ -80,3 +86,56 @@ def deepcopy_simple(v):
         return v
     else:
         return deepcopy(v)
+
+
+
+class LockError(Exception):
+    pass
+
+
+class LockTimeoutError(LockError):
+    pass
+
+
+class LockNotAvailableError(LockError):
+    pass
+
+
+@contextmanager
+def get_dataset_cache_lock(
+    export_path: os.PathLike[str],
+    *,
+    ttl: int | timedelta,
+    block: bool = True,
+    acquire_timeout: Optional[int | timedelta] = None,
+) -> Generator[Lock, Any, Any]:
+    if isinstance(acquire_timeout, timedelta):
+        acquire_timeout = acquire_timeout.total_seconds()
+    elif acquire_timeout and acquire_timeout < 0:
+        raise ValueError("acquire_timeout must be a positive number")
+    elif acquire_timeout is None:
+        acquire_timeout = -1
+
+    if isinstance(ttl, timedelta):
+        ttl = ttl.total_seconds()
+    elif not ttl or ttl < 0:
+        raise ValueError("ttl must be a positive number")
+
+    lock = Redlock(
+        key=export_path,
+        masters={django_rq.get_connection(settings.CVAT_QUEUES.EXPORT_DATA.value)},
+        auto_release_time=ttl,
+    )
+    try:
+        acquired = lock.acquire(blocking=block, timeout=acquire_timeout)
+        if acquired:
+            yield lock
+        else:
+            if 0 < acquire_timeout:
+                raise LockTimeoutError
+            else:
+                raise LockNotAvailableError
+
+    finally:
+        with suppress(ReleaseUnlockedLock):
+            lock.release()

cvat/apps/dataset_manager/views.py

@@ -1,14 +1,18 @@
 # Copyright (C) 2019-2022 Intel Corporation
-# Copyright (C) 2023 CVAT.ai Corporation
+# Copyright (C) 2023-2024 CVAT.ai Corporation
 #
 # SPDX-License-Identifier: MIT
 
+from contextlib import suppress
+import logging
 import os
 import os.path as osp
+import re
 import tempfile
 from datetime import timedelta
 
 import django_rq
+import rq
 from datumaro.util.os_util import make_file_name
 from datumaro.util import to_snake_case
 from django.utils import timezone
@@ -20,7 +24,7 @@
 from cvat.apps.engine.models import Project, Task, Job
 
 from .formats.registry import EXPORT_FORMATS, IMPORT_FORMATS
-from .util import current_function_name
+from .util import current_function_name, get_dataset_cache_lock, LockError
 
 slogger = ServerLogManager(__name__)
 
@@ -32,78 +36,123 @@ def log_exception(logger=None, exc_info=True):
             (_MODULE_NAME, current_function_name(2)),
         exc_info=exc_info)
 
+EXPORT_CACHE_DIR_NAME = 'export_cache'
 
 def get_export_cache_dir(db_instance):
     base_dir = osp.abspath(db_instance.get_dirname())
 
     if osp.isdir(base_dir):
-        return osp.join(base_dir, 'export_cache')
+        return osp.join(base_dir, EXPORT_CACHE_DIR_NAME)
     else:
-        raise FileNotFoundError('{} dir {} does not exist'.format(db_instance.__class__.__name__, base_dir))
+        raise FileNotFoundError(
+            '{} dir {} does not exist'.format(db_instance.__class__.__name__, base_dir)
+        )
 
-DEFAULT_CACHE_TTL = timedelta(hours=10)
-TASK_CACHE_TTL = DEFAULT_CACHE_TTL
+DEFAULT_CACHE_TTL = timedelta(seconds=settings.DATASET_CACHE_TTL)
 PROJECT_CACHE_TTL = DEFAULT_CACHE_TTL / 3
+TASK_CACHE_TTL = DEFAULT_CACHE_TTL
 JOB_CACHE_TTL = DEFAULT_CACHE_TTL
 
+EXPORT_CACHE_LOCK_TIMEOUT = timedelta(seconds=settings.DATASET_CACHE_LOCK_TIMEOUT)
+
+def get_export_cache_ttl(db_instance: str | Project | Task | Job) -> int:
+    TTL_CONSTS = {
+        'project': PROJECT_CACHE_TTL,
+        'task': TASK_CACHE_TTL,
+        'job': JOB_CACHE_TTL,
+    }
+
+    if not isinstance(db_instance, str):
+        db_instance = db_instance.__class__.__name__.lower()
+
+    return TTL_CONSTS[db_instance].total_seconds()
+
+def get_file_instance_timestamp(file_path: str) -> float:
+    match = re.search(r'instance(\d+\.\d+)', osp.basename(file_path))
+    return float(match.group(1))
+
+
 def export(dst_format, project_id=None, task_id=None, job_id=None, server_url=None, save_images=False):
     try:
         if task_id is not None:
             logger = slogger.task[task_id]
-            cache_ttl = TASK_CACHE_TTL
             export_fn = task.export_task
             db_instance = Task.objects.get(pk=task_id)
         elif project_id is not None:
             logger = slogger.project[project_id]
-            cache_ttl = PROJECT_CACHE_TTL
             export_fn = project.export_project
             db_instance = Project.objects.get(pk=project_id)
         else:
             logger = slogger.job[job_id]
-            cache_ttl = JOB_CACHE_TTL
             export_fn = task.export_job
             db_instance = Job.objects.get(pk=job_id)
 
-        cache_dir = get_export_cache_dir(db_instance)
+        cache_ttl = timedelta(seconds=get_export_cache_ttl(db_instance))
 
+        cache_dir = get_export_cache_dir(db_instance)
         exporter = EXPORT_FORMATS[dst_format]
-        output_base = '%s_%s' % ('dataset' if save_images else 'annotations',
-            make_file_name(to_snake_case(dst_format)))
-        output_path = '%s.%s' % (output_base, exporter.EXT)
-        output_path = osp.join(cache_dir, output_path)
 
-        instance_time = timezone.localtime(db_instance.updated_date).timestamp()
+        instance_update_time = timezone.localtime(db_instance.updated_date)
         if isinstance(db_instance, Project):
-            tasks_update = list(map(lambda db_task: timezone.localtime(
-                db_task.updated_date).timestamp(), db_instance.tasks.all()))
-            instance_time = max(tasks_update + [instance_time])
-        if not (osp.exists(output_path) and \
-                instance_time <= osp.getmtime(output_path)):
-            os.makedirs(cache_dir, exist_ok=True)
-            with tempfile.TemporaryDirectory(dir=cache_dir) as temp_dir:
-                temp_file = osp.join(temp_dir, 'result')
-                export_fn(db_instance.id, temp_file, dst_format,
-                    server_url=server_url, save_images=save_images)
-                os.replace(temp_file, output_path)
-
-            archive_ctime = osp.getctime(output_path)
-            scheduler = django_rq.get_scheduler(settings.CVAT_QUEUES.EXPORT_DATA.value)
-            cleaning_job = scheduler.enqueue_in(time_delta=cache_ttl,
-                func=clear_export_cache,
-                file_path=output_path,
-                file_ctime=archive_ctime,
-                logger=logger)
-            logger.info(
-                "The {} '{}' is exported as '{}' at '{}' "
-                "and available for downloading for the next {}. "
-                "Export cache cleaning job is enqueued, id '{}'".format(
-                    db_instance.__class__.__name__.lower(),
-                    db_instance.name if isinstance(db_instance, (Project, Task)) else db_instance.id,
-                    dst_format, output_path, cache_ttl,
-                    cleaning_job.id
-                ))
+            tasks_update = list(map(
+                lambda db_task: timezone.localtime(db_task.updated_date),
+                db_instance.tasks.all()
+            ))
+            instance_update_time = max(tasks_update + [instance_update_time])
+
+        output_path = '%s-instance%s_%s.%s' % (
+            'dataset' if save_images else 'annotations',
+            # store the instance timestamp in the file name to reliably get this information
+            # ctime / mtime do not return file creation time on linux
+            # mtime is used for file usage checks
+            instance_update_time.timestamp(),
+            make_file_name(to_snake_case(dst_format)),
+            exporter.EXT,
+        )
+        output_path = osp.join(cache_dir, output_path)
+
+        os.makedirs(cache_dir, exist_ok=True)
+
+        with get_dataset_cache_lock(
+            output_path,
+            block=True,
+            acquire_timeout=EXPORT_CACHE_LOCK_TIMEOUT,
+            ttl=rq.get_current_job().timeout,
+        ):
+            if not (
+                osp.exists(output_path)
+                and instance_update_time.timestamp() <= get_file_instance_timestamp(output_path)
+            ):
+                with tempfile.TemporaryDirectory(dir=cache_dir) as temp_dir:
+                    temp_file = osp.join(temp_dir, 'result')
+                    export_fn(db_instance.id, temp_file, dst_format,
+                        server_url=server_url, save_images=save_images)
+                    os.replace(temp_file, output_path)
+
+                scheduler = django_rq.get_scheduler(settings.CVAT_QUEUES.EXPORT_DATA.value)
+                cleaning_job = scheduler.enqueue_in(
+                    time_delta=cache_ttl,
+                    func=clear_export_cache,
+                    file_path=output_path,
+                    file_ctime=instance_update_time.timestamp(),
+                    logger=logger
+                )
+                logger.info(
+                    "The {} '{}' is exported as '{}' at '{}' "
+                    "and available for downloading for the next {}. "
+                    "Export cache cleaning job is enqueued, id '{}'".format(
+                        db_instance.__class__.__name__.lower(),
+                        db_instance.name if isinstance(db_instance, (Project, Task)) else db_instance.id,
+                        dst_format, output_path, cache_ttl,
+                        cleaning_job.id
+                    )
+                )
 
         return output_path
+    except LockError:
+        rq_job = rq.get_current_job()
+        rq_job.retry(django_rq.get_queue(settings.CVAT_QUEUES.EXPORT_DATA.value))
+        return
     except Exception:
         log_exception(logger)
         raise
@@ -127,14 +176,55 @@ def export_project_annotations(project_id, dst_format=None, server_url=None):
     return export(dst_format, project_id=project_id, server_url=server_url, save_images=False)
 
 
-def clear_export_cache(file_path, file_ctime, logger):
+def clear_export_cache(file_path: str, file_ctime: float, logger: logging.Logger) -> None:
     try:
-        if osp.exists(file_path) and osp.getctime(file_path) == file_ctime:
+        with get_dataset_cache_lock(
+            file_path,
+            block=True,
+            acquire_timeout=EXPORT_CACHE_LOCK_TIMEOUT,
+            ttl=rq.get_current_job().timeout,
+        ):
+            if 'job' in file_path:
+                instance_type = 'job'
+            elif 'task' in file_path:
+                instance_type = 'task'
+            elif 'project' in file_path:
+                instance_type = 'project'
+            else:
+                assert False
+
+            cache_ttl = get_export_cache_ttl(instance_type)
+
+            instance_timestamp = None
+            with suppress(AttributeError): # backward compatibility
+                instance_timestamp = get_file_instance_timestamp(file_path)
+            if instance_timestamp and instance_timestamp != file_ctime:
+                logger.info("Export cache file '{}' has changed, skipping".format(file_path))
+                return
+
+            if timezone.now().timestamp() <= osp.getmtime(file_path) + cache_ttl:
+                # Need to retry later, the export is in use
+                rq_job = rq.get_current_job()
+                rq_job.retries_left = 1
+                rq_job.retry_intervals = [cache_ttl]
+                rq_job.retry(
+                    django_rq.get_queue(settings.CVAT_QUEUES.EXPORT_DATA.value), pipeline=None
+                )
+                logger.info(
+                    "Export cache file '{}' is recently accessed, will retry in {}".format(
+                        file_path, timedelta(seconds=cache_ttl)
+                    )
+                )
+                return
+
+            # TODO: maybe remove all outdated exports
             os.remove(file_path)
-
-            logger.info(
-                "Export cache file '{}' successfully removed" \
-                .format(file_path))
+            logger.info("Export cache file '{}' successfully removed".format(file_path))
+    except LockError:
+        # Need to retry later if the lock was not available
+        rq_job = rq.get_current_job()
+        rq_job.retry(django_rq.get_queue(settings.CVAT_QUEUES.EXPORT_DATA.value), pipeline=None)
+        return
     except Exception:
         log_exception(logger)
         raise