Merge pull request #358 from lsst-dm/tickets/DM-53057

DM-53057: Intermittent ConcurrentModification errors on S3 reads

Author: hsinfang

python/activator/middleware_interface.py

@@ -1198,6 +1198,7 @@ def _prep_pipeline_graph(self, pipeline_file) -> lsst.pipe.base.PipelineGraph:
         """
         return self._prep_pipeline(pipeline_file).to_graph()
 
+    @connect.retry(2, botocore.exceptions.ClientError, wait=5)
     def _download(self, remote):
         """Download an image located on a remote store.
 

python/shared/raw.py

@@ -37,19 +37,21 @@
     "get_raw_path",
 ]
 
-import json
 import logging
 import os
 import re
 import time
 import urllib.parse
 
+import botocore.exceptions
+import pydantic_core
 import requests
 
 from lsst.obs.lsst import LsstCam, LsstCamImSim, LsstComCam, LsstComCamSim
 from lsst.obs.lsst.translators.lsst import LsstBaseTranslator
 from lsst.resources import ResourcePath
 
+from .connect_utils import retry
 from .visit import FannedOutVisit
 
 _log = logging.getLogger("lsst." + __name__)
@@ -387,19 +389,38 @@ def get_group_id_from_oid(oid: str) -> str:
         oid.removesuffix(m["extension"])
         + ".json"
     )
-    # Wait a bit but not too long for the file.
-    # It should normally show up before the image.
-    count = 0
-    while not sidecar.exists():
-        count += 1
-        if count > 20:
-            raise RuntimeError(f"Unable to retrieve JSON sidecar: {sidecar}")
-        time.sleep(0.1)
+    group_id = _get_group_id_from_sidecar(sidecar)
+    return group_id
+
+
+@retry(4, botocore.exceptions.ClientError, wait=5)
+def _get_group_id_from_sidecar(sidecar):
+    """Read the group id from a sidecar JSON file.
 
-    with sidecar.open("r") as f:
-        md = json.load(f)
+    The sidecar file normally show up before the image. If not present, wait a
+    bit but not too long for the file. Sometimes, the object store gives other
+    transient ClientErrors, which are retried in a longer timescale.
+
+    Parameters
+    ----------
+    sidecar : `lsst.resources.ResourcePath`
+        URI to a sidecar JSON file.
 
-    return md.get("GROUPID", "")
+    Returns
+    -------
+    group_id : `str`
+        The group identifier as a string.
+    """
+    count = 0
+    while count <= 20:
+        try:
+            md = pydantic_core.from_json(sidecar.read())
+            return md.get("GROUPID", "")
+        # If no such sidecar exists, a FileNotFoundError is raised.
+        except FileNotFoundError:
+            count += 1
+            time.sleep(0.1)
+    raise RuntimeError(f"Unable to retrieve JSON sidecar: {sidecar}")
 
 
 def get_raw_path(instrument, detector, group, snap, exposure_id, physical_filter):