From d1992513b83148aba2d2866721cdc4326b95c3e3 Mon Sep 17 00:00:00 2001
From: Valentin Rigal <rigal@teklia.com>
Date: Tue, 18 May 2021 16:36:37 +0200
Subject: [PATCH] Handle concurrency while initializing activity
---
arkindex_worker/worker/__init__.py | 14 +++++++++++++-
1 file changed, 13 insertions(+), 1 deletion(-)
diff --git a/arkindex_worker/worker/__init__.py b/arkindex_worker/worker/__init__.py
index cfd043dc..c34a24ae 100644
--- a/arkindex_worker/worker/__init__.py
+++ b/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}"
--
GitLab