fix: remember ETag of uploaded resources (close #80)

Author: paulmueller

CHANGELOG

@@ -1,6 +1,8 @@
 0.18.3
+ - fix: remember ETag of uploaded resources (#80)
  - enh: allow sharing individual datasets instead of collections (#32)
  - enh: add preview option for dataset description in upload dialog (#52)
+ - enh: introduce write lock for eternal job updates
 0.18.2
  - ref: make action buttons in filter view hideable
 0.18.2

dcoraid/upload/job.py

@@ -5,6 +5,7 @@
 import shutil
 import time
 import traceback
+from typing import Callable
 import warnings
 
 import dclab
@@ -54,6 +55,7 @@ def __init__(self,
                  resource_paths: list[str | pathlib.Path],
                  resource_names: list[str] = None,
                  resource_supplements: list[dict] = None,
+                 resource_etags: list[str] = None,
                  collections: list[str] = None,
                  task_id: str = None,
                  cache_dir: str | pathlib.Path = None):
@@ -79,6 +81,8 @@ def __init__(self,
             on DCOR
         resource_supplements: list of dict
             Supplementary resource information
+        resource_etags: list
+            Any previously computed ETags of the resources
         collections: list of strings
             List of unique identifiers of collections that this
             dataset should be appended to
@@ -152,7 +156,7 @@ def __init__(self,
                            for ff in self.paths]
         self.file_bytes_uploaded = [0] * len(self.paths)
         #: ETags for the files uploaded by this UploadJob instance
-        self.etags = [None] * len(self.paths)
+        self.etags = resource_etags or [None] * len(self.paths)
         #: Resource index indicating upload progress
         self.index = 0
         self.start_time = None
@@ -188,6 +192,7 @@ def __getstate__(self):
             "resource_paths": [str(pp) for pp in self.paths],
             "resource_names": self.resource_names,
             "resource_supplements": self.supplements,
+            "resource_etags": self.etags,
             "task_id": self.task_id,
         }
         return uj_state
@@ -550,12 +555,17 @@ def task_compress_resources(self):
                     self.file_sizes[ii] = path_out.stat().st_size
         self.set_state("parcel")
 
-    def task_upload_resources(self):
+    def task_upload_resources(self,
+                              run_after_upload: Callable = None
+                              ):
         """Start the upload
 
         The progress of the upload is monitored and written
         to attributes. The current status can be retrieved
         via :func:`UploadJob.get_status`.
+
+        ``run_after_upload(self)`` is called after each resource upload.
+        This is used to update the eternal/immortalized job on disk.
         """
         if self.state == "parcel":
             # reset everything
@@ -605,6 +615,8 @@ def task_upload_resources(self):
                     self.etags[ii] = etag
                     self.paths_uploaded.append(path)
                     self.wait_time += srv_time
+                    if run_after_upload is not None:
+                        run_after_upload(self)
             self.end_time = time.perf_counter()
             self.set_state("online")
         else:

dcoraid/upload/queue.py

@@ -1,6 +1,8 @@
 import logging
 import pathlib
 import time
+import threading
+import traceback
 import traceback as tb
 import warnings
 
@@ -27,6 +29,7 @@ def __init__(self, path):
         self.path_queued = self.path / "queued"
         self.path_completed.mkdir(parents=True, exist_ok=True)
         self.path_queued.mkdir(parents=True, exist_ok=True)
+        self.write_locks = {}
 
     def __contains__(self, item):
         if isinstance(item, UploadJob):
@@ -49,6 +52,10 @@ def get_queued_dataset_ids(self):
         """Return list of DCOR dataset IDs corresponding to queued jobs"""
         return sorted([pp.stem for pp in self.path_queued.glob("*.json")])
 
+    def get_write_lock(self, dataset_id):
+        """Return a `threading.Lock` object for write operations"""
+        return self.write_locks.setdefault(dataset_id, threading.Lock())
+
     def is_job_done(self, dataset_id):
         jp = self.path_completed / (dataset_id + ".json")
         return jp.exists()
@@ -68,21 +75,27 @@ def immortalize_job(self, upload_job):
             # is done after checking for queued (above case).
             raise FileExistsError(f"The job '{upload_job.dataset_id}' is "
                                   f"already done!")
-        save_task(upload_job=upload_job, path=pout)
+        with self.get_write_lock(upload_job.dataset_id):
+            save_task(upload_job=upload_job, path=pout)
 
     def job_exists(self, dataset_id):
-        return self.is_job_queued(dataset_id) or self.is_job_done(dataset_id)
+        # Use a write lock here to avoid race conditions
+        with self.get_write_lock(dataset_id):
+            return (self.is_job_queued(dataset_id)
+                    or self.is_job_done(dataset_id))
 
     def obliterate_job(self, dataset_id):
         """Remove a job from the persistent queue list"""
         pdel = self.path_queued / (dataset_id + ".json")
-        pdel.unlink()
+        with self.get_write_lock(dataset_id):
+            pdel.unlink()
 
     def set_job_done(self, dataset_id):
         """Move the job from the queue to the complete list"""
         pin = self.path_queued / (dataset_id + ".json")
         pout = self.path_completed / (dataset_id + ".json")
-        pin.rename(pout)
+        with self.get_write_lock(dataset_id):
+            pin.rename(pout)
 
     def summon_job(self, dataset_id, api, cache_dir=None):
         """Instantiate job from the persistent queue list"""
@@ -91,6 +104,20 @@ def summon_job(self, dataset_id, api, cache_dir=None):
         assert upload_job.dataset_id == dataset_id
         return upload_job
 
+    def update_job(self, upload_job):
+        """Update an immortalized job dictionary
+
+        The job must exist. Updating a job makes sense when e.g.
+        the ETag of a resource becomes known.
+        """
+        if not self.job_exists(upload_job.dataset_id):
+            raise ValueError(f"Cannot update non-existent job {upload_job}")
+        with self.get_write_lock(upload_job.dataset_id):
+            pq = self.path_queued / (upload_job.dataset_id + ".json")
+            pc = self.path_completed / (upload_job.dataset_id + ".json")
+            pp = pq if pq.exists() else pc
+            save_task(upload_job=upload_job, path=pp)
+
 
 class UploadQueue:
     def __init__(self, api, path_persistent_job_list=None, cache_dir=None):
@@ -142,13 +169,19 @@ def __init__(self, api, path_persistent_job_list=None, cache_dir=None):
         else:
             self.jobs_eternal = None
         self.daemon_compress = CompressDaemon(self.jobs)
-        self.daemon_upload = UploadDaemon(self.jobs)
+        self.daemon_upload = UploadDaemon(self.jobs, self.jobs_eternal)
         self.daemon_verify = VerifyDaemon(self.jobs)
 
     def __contains__(self, upload_job):
         return upload_job in self.jobs
 
     def __del__(self):
+        # Attempt to update the eternal jobs (important for ETags)
+        for job in self.jobs:
+            try:
+                self.jobs_eternal.update_job(job)
+            except BaseException:
+                self.logger.error(traceback.format_exc())
         self.daemon_upload.shutdown_flag.set()
         self.daemon_verify.shutdown_flag.set()
         self.daemon_compress.shutdown_flag.set()
@@ -199,7 +232,7 @@ def abort_job(self, dataset_id):
             # the daemon to set all the other jobs to "abort".
             self.daemon_upload.shutdown_flag.set()
             self.daemon_upload.terminate()
-            self.daemon_upload = UploadDaemon(self.jobs)
+            self.daemon_upload = UploadDaemon(self.jobs, self.jobs_eternal)
         elif prev_state == "compress":
             self.daemon_compress.shutdown_flag.set()
             self.daemon_compress.terminate()
@@ -327,12 +360,20 @@ def __init__(self, jobs):
 
 
 class UploadDaemon(Daemon):
-    def __init__(self, jobs):
+    def __init__(self, jobs, jobs_eternal):
         """Upload daemon"""
+        if jobs_eternal is None:
+            run_after_upload = None
+        else:
+            run_after_upload = jobs_eternal.update_job
         super(UploadDaemon, self).__init__(
             jobs,
             job_trigger_state="parcel",
-            job_function_name="task_upload_resources")
+            job_function_name="task_upload_resources",
+            job_function_kwargs={
+                "run_after_upload": run_after_upload
+            }
+        )
 
 
 class VerifyDaemon(Daemon):

dcoraid/worker/daemon.py

@@ -11,11 +11,17 @@
 
 
 class Daemon(KThread):
-    def __init__(self, queue, job_trigger_state, job_function_name):
+    def __init__(self,
+                 jobs,
+                 job_trigger_state: str,
+                 job_function_name: str,
+                 job_function_kwargs: dict = None,
+                 ):
         """Daemon base class for running uploads/downloads in the background"""
-        self.queue = queue
+        self.jobs = jobs
         self.job_trigger_state = job_trigger_state
         self.job_function_name = job_function_name
+        self.job_function_kwargs = job_function_kwargs
         super(Daemon, self).__init__()
         self.daemon = True  # We don't have to worry about ending this thread
 
@@ -31,7 +37,7 @@ def run(self):
         try:
             while not self.shutdown_flag.is_set():
                 # Get the first job that is in the trigger state
-                for job in self.queue:
+                for job in self.jobs:
                     if job.state == self.job_trigger_state:
                         break
                 else:
@@ -42,7 +48,10 @@ def run(self):
                 task = getattr(job, self.job_function_name)
                 logger = logging.getLogger(__name__)
                 try:
-                    task()
+                    if self.job_function_kwargs is None:
+                        task()
+                    else:
+                        task(**self.job_function_kwargs)
                 except ConnectionTimeoutErrors:
                     # Set the job to the error state for 10s (so the user
                     # sees it in the UI) and then go back to the initial

tests/test_upload_queue.py

@@ -1,3 +1,4 @@
+import json
 import pathlib
 import tempfile
 import time
@@ -87,6 +88,24 @@ def test_queue_create_dataset_with_resource():
     common.wait_for_job(joblist, data["id"])
 
 
+def test_queue_create_dataset_with_resource_write_etag(tmp_path):
+    api = common.get_api()
+    # create some metadata
+    dataset_dict = common.make_dataset_dict(hint="create-with-resource")
+    # post dataset creation request
+    data = dataset_create(dataset_dict=dataset_dict, api=api)
+    joblist = UploadQueue(api=api,
+                          path_persistent_job_list=tmp_path)
+    joblist.new_job(dataset_id=data["id"],
+                    paths=[dpath])
+    pujl = joblist.jobs_eternal
+    common.wait_for_job(joblist, data["id"])
+
+    # Make sure the resource etag is stored in the uploaded job.
+    uj_d = json.loads((pujl.path_completed / f"{data["id"]}.json").read_text())
+    assert len(uj_d["upload_job"]["resource_etags"][0]) == 32
+
+
 def test_queue_find_zombie_caches():
     api = common.get_api()
     # create some metadata
@@ -245,6 +264,20 @@ def test_persistent_upload_joblist_error_exists():
         pujl.immortalize_job(uj)
 
 
+def test_persistent_upload_joblist_error_exists_done():
+    """test things when a job is done"""
+    api = common.get_api()
+    td = pathlib.Path(tempfile.mkdtemp(prefix="persistent_uj_list_"))
+    pujl_path = td / "joblistdir"
+    task_path = common.make_upload_task()
+    pujl = PersistentUploadJobList(pujl_path)
+    uj = load_task(task_path, api=api)
+    pujl.immortalize_job(uj)
+    pujl.set_job_done(uj.dataset_id)
+    with pytest.raises(FileExistsError, match="is already done"):
+        pujl.immortalize_job(uj)
+
+
 def test_persistent_upload_joblist_skip_finished_resources():
     api = common.get_api()
     td = pathlib.Path(tempfile.mkdtemp(prefix="persistent_uj_list_"))
@@ -340,7 +373,7 @@ def test_persistent_upload_joblist_skip_queued_resources():
     assert uq.jobs_eternal.num_queued == 1
 
 
-def test_persistent_upload_joblist_error_exists_done():
+def test_persistent_upload_joblist_update():
     """test things when a job is done"""
     api = common.get_api()
     td = pathlib.Path(tempfile.mkdtemp(prefix="persistent_uj_list_"))
@@ -349,9 +382,20 @@ def test_persistent_upload_joblist_error_exists_done():
     pujl = PersistentUploadJobList(pujl_path)
     uj = load_task(task_path, api=api)
     pujl.immortalize_job(uj)
-    pujl.set_job_done(uj.dataset_id)
-    with pytest.raises(FileExistsError, match="is already done"):
-        pujl.immortalize_job(uj)
+
+    uj.etags[0] = etag = str(uuid.uuid4())
+    pujl.update_job(uj)
+
+    assert pujl.job_exists(uj.dataset_id)
+    assert pujl.is_job_queued(uj.dataset_id)
+    assert not pujl.is_job_done(uj.dataset_id)
+
+    ids = pujl.get_queued_dataset_ids()
+    assert uj.dataset_id in ids
+
+    # make sure the etag was stored
+    uj_d = json.loads((pujl.path_queued / f"{uj.dataset_id}.json").read_text())
+    assert uj_d["upload_job"]["resource_etags"][0] == etag
 
 
 def test_persistent_upload_joblist_warning_dataset_deleted_on_server():