Added logic to retry database refreshes and file path validation to avoid race conditions when transferring/registering items

Author: tieneupin

src/murfey/server/api/clem.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import re
+import time
 import traceback
 from ast import literal_eval
 from importlib.metadata import EntryPoint  # type hinting only
@@ -94,7 +95,14 @@ def validate_and_sanitise(
         raise ValueError(f"{file} points to a directory that is not permitted")
 
     # Check that it's a file, not a directory
-    if full_path.is_file() is False:
+    # Make a couple of attempts to rule out race condition
+    attempts = 0
+    while attempts < 50:
+        if full_path.is_file() is True:
+            break
+        attempts += 1
+        time.sleep(0.1)
+    else:
         raise ValueError(f"{file} is not a file")
 
     # Check that it is of a permitted file type
@@ -184,7 +192,23 @@ def get_db_entry(
         )
         db.add(db_entry)
         db.commit()
-        db.refresh(db_entry)
+
+        # Make multiple attempts to refresh data in case of race condition
+        attempts = 0
+        while attempts < 50:
+            try:
+                db.refresh(db_entry)
+                break
+            except Exception:
+                pass
+            attempts += 1
+            time.sleep(0.1)
+        else:
+            raise RuntimeError(
+                "Maximum number of attempts reached while trying to refresh database "
+                f"entry for {str(file_path if file_path else series_name)!r}"
+            )
+
     except Exception:
         raise Exception
 
@@ -215,7 +239,10 @@ def register_lif_file(
             file_path=lif_file,
         )
     except Exception:
-        logger.error(traceback.format_exc())
+        logger.error(
+            f"Exception encountered while registering LIF file {str(lif_file)!r}: \n"
+            f"{traceback.format_exc()}"
+        )
         return False
 
     # Add metadata information if provided
@@ -224,7 +251,11 @@ def register_lif_file(
             master_metadata = validate_and_sanitise(master_metadata, session_id, db)
             clem_lif_file.master_metadata = str(master_metadata)
         except Exception:
-            logger.warning(traceback.format_exc())
+            logger.warning(
+                "Unable to add master metadata information to database entry for "
+                f"LIF file {str(lif_file)!r}: \n"
+                f"{traceback.format_exc()}"
+            )
 
     # Register child metadata if provided
     for metadata in child_metadata:
@@ -238,7 +269,11 @@ def register_lif_file(
             # Append to database entry
             clem_lif_file.child_metadata.append(metadata_db_entry)
         except Exception:
-            logger.warning(traceback.format_exc())
+            logger.warning(
+                "Unable to add child metadata information to database entry for "
+                f"LIF file {str(lif_file)!r}: \n"
+                f"{traceback.format_exc()}"
+            )
             continue
 
     # Register child image series if provided
@@ -253,7 +288,11 @@ def register_lif_file(
             # Append to database entry
             clem_lif_file.child_series.append(series_db_entry)
         except Exception:
-            logger.warning(traceback.format_exc())
+            logger.warning(
+                "Unable to add child series information to database entry for "
+                f"LIF file {str(lif_file)!r}: \n"
+                f"{traceback.format_exc()}"
+            )
             continue
 
     # Register child image stacks if provided
@@ -268,7 +307,11 @@ def register_lif_file(
             # Append to database entry
             clem_lif_file.child_stacks.append(stack_db_entry)
         except Exception:
-            logger.warning(traceback.format_exc())
+            logger.warning(
+                "Unable to add child image stack information to database entry for "
+                f"LIF file {str(lif_file)!r}: \n"
+                f"{traceback.format_exc()}"
+            )
             continue
 
     # Commit to database
@@ -296,7 +339,10 @@ def register_tiff_file(
             file_path=tiff_file,
         )
     except Exception:
-        logger.error(traceback.format_exc())
+        logger.error(
+            f"Exception encountered while registering TIFF file {str(tiff_file)!r}: \n"
+            f"{traceback.format_exc()}"
+        )
         return False
 
     # Add metadata if provided
@@ -311,7 +357,10 @@ def register_tiff_file(
             # Link database entries
             clem_tiff_file.associated_metadata = metadata_db_entry
         except Exception:
-            logger.warning(traceback.format_exc())
+            logger.warning(
+                f"Unable to register metadata for TIFF file {str(tiff_file)!r}: \n"
+                f"{traceback.format_exc()}"
+            )
 
     # Add series information if provided
     if associated_series is not None:
@@ -325,7 +374,11 @@ def register_tiff_file(
             # Link database entries
             clem_tiff_file.child_series = series_db_entry
         except Exception:
-            logger.warning(traceback.format_exc())
+            logger.warning(
+                "Unable to register series information for TIFF file "
+                f"{str(tiff_file)!r}: \n"
+                f"{traceback.format_exc()}"
+            )
 
     # Add image stack information if provided
     if associated_stack is not None:
@@ -339,7 +392,11 @@ def register_tiff_file(
             # Link database entries
             clem_tiff_file.child_stack = stack_db_entry
         except Exception:
-            logger.warning(traceback.format_exc())
+            logger.warning(
+                f"Unable to register image stack {str(associated_stack)!r} "
+                f"in database for TIFF file {str(tiff_file)!r}: \n"
+                f"{traceback.format_exc()}"
+            )
 
     # Commit to database
     db.add(clem_tiff_file)

src/murfey/workflows/clem/__init__.py

@@ -2,6 +2,7 @@
 
 import logging
 import re
+import time
 from pathlib import Path
 from typing import Optional, Type, Union
 
@@ -90,7 +91,14 @@ def _validate_and_sanitise(
         raise ValueError(f"{file} points to a directory that is not permitted")
 
     # Check that it's a file, not a directory
-    if full_path.is_file() is False:
+    # Make a couple of attempts to rule out race condition
+    attempts = 0
+    while attempts < 50:
+        if full_path.is_file() is True:
+            break
+        attempts += 1
+        time.sleep(0.1)
+    else:
         raise ValueError(f"{file} is not a file")
 
     # Check that it is of a permitted file type
@@ -180,7 +188,22 @@ def get_db_entry(
         )
         db.add(db_entry)
         db.commit()
-        db.refresh(db_entry)
+
+        # Make multiple attempts data retrieval attempts in case of race condition
+        attempts = 0
+        while attempts < 50:
+            try:
+                db.refresh(db_entry)
+                break
+            except Exception:
+                pass
+            attempts += 1
+            time.sleep(0.1)
+        else:
+            raise RuntimeError(
+                "Maximum number of attempts reached while trying to refresh database "
+                f"entry for {str(file_path if file_path else series_name)!r}"
+            )
     except Exception:
         raise Exception
 

src/murfey/workflows/clem/register_align_and_merge_results.py

@@ -2,6 +2,7 @@
 
 import json
 import logging
+import time
 import traceback
 from ast import literal_eval
 from pathlib import Path
@@ -60,29 +61,22 @@ def register_align_and_merge_result(
     )
 
     # Validate message and try and load results
-    if isinstance(message["result"], str):
-        try:
+    try:
+        if isinstance(message["result"], str):
             json_obj: dict = json.loads(message["result"])
             result = AlignAndMergeResult(**json_obj)
-        except Exception:
-            logger.error(traceback.format_exc())
-            logger.error(
-                "Exception encountered when parsing align-and-merge processing result"
-            )
-            return False
-    elif isinstance(message["result"], dict):
-        try:
+        elif isinstance(message["result"], dict):
             result = AlignAndMergeResult(**message["result"])
-        except Exception:
-            logger.error(traceback.format_exc())
+        else:
             logger.error(
-                "Exception encountered when parsing align-and-merge processing result"
+                "Invalid type for align-and-merge processing result: "
+                f"{type(message['result'])}"
             )
             return False
-    else:
+    except Exception:
         logger.error(
-            "Invalid type for align-and-merge processing result: "
-            f"{type(message['result'])}"
+            "Exception encountered when parsing align-and-merge processing result: \n"
+            f"{traceback.format_exc()}"
         )
         return False
 
@@ -100,18 +94,33 @@ def register_align_and_merge_result(
             clem_img_series.composite_created = True
             murfey_db.add(clem_img_series)
             murfey_db.commit()
-            murfey_db.refresh(clem_img_series)
+
+            # Make multiple attempts to refresh data in case of race condition
+            attempts = 0
+            while attempts < 50:
+                try:
+                    murfey_db.refresh(clem_img_series)
+                    break
+                except Exception:
+                    pass
+                attempts += 1
+                time.sleep(0.1)
+            else:
+                raise RuntimeError(
+                    "Maximum number of attempts reached while trying to refresh database "
+                    f"entry for {result.series_name!r}"
+                )
 
             logger.info(
                 "Align-and-merge processing result registered for "
                 f"{result.series_name!r} series"
             )
 
         except Exception:
-            logger.error(traceback.format_exc())
             logger.error(
                 "Exception encountered when registering LIF preprocessing result for "
-                f"{result.series_name!r} {result.channel!r} image stack"
+                f"{result.series_name!r} {result.channel!r} image stack: \n"
+                f"{traceback.format_exc()}"
             )
             return False