Handle concurrency while initializing activity

arkindex_worker/worker/__init__.py

@@ -127,7 +127,18 @@ class ElementsWorker(
                 logger.info(f"Processing {element} ({i}/{count})")
 
                 # Process the element and report its progress if activities are enabled
-                self.update_activity(element.id, ActivityState.Started)
+                response = self.update_activity(element.id, ActivityState.Started)
+                if isinstance(response, ErrorResponse) and response.status_code == 400:
+                    # Bad request error when setting an activity to "started" mean that we cannot
+                    # process this element. We assume that the reason is that the state transition
+                    # was forbidden i.e. that the activity was already in a started or processed state.
+                    # This allow concurrent access to an element activity.
+                    # Element is not counted as failed as it is handled by another process.
+                    logger.info(
+                        f'Cannot start processing element "{element.id}". '
+                        "Another process could have processed it with the same version already."
+                    )
+                    continue
                 self.process_element(element)
                 self.update_activity(element.id, ActivityState.Processed)
             except Exception as e:
@@ -201,6 +212,7 @@ class ElementsWorker(
             logger.warning(
                 f"Failed to update activity of element {element_id} to {state.value} due to an API error: {e.content}"
             )
+            return e
         except Exception as e:
             logger.warning(
                 f"Failed to update activity of element {element_id} to {state.value}: {e}"