erp5_core: ensure Base_reindexAndSenseAlarm never end without calling the alarm

bt5/erp5_core_test/TestTemplateItem/portal_components/test.erp5.testERP5CoreSkins.py

@@ -349,6 +349,41 @@ class TestBase_reindexAndSenseAlarm(ERP5TypeTestCase):
       alarm.workflow_history['edit_workflow'][-1]['comment']
     )
 
+  def test_reindexAndSenseAlarm_manyContextSameTransaction(self):
+    # Check that the script wait for the alarm to be not activate
+    # before triggering it again
+    alarm = self.portal.portal_alarms.invoice_builder_alarm
+    workflow_history_count = len(alarm.workflow_history['edit_workflow'])
+
+    document_list = []
+    for doc in self.portal.contentValues():
+      if doc.getId().endswith('_module'):
+        document_list.append((doc, self.getIndexationDate(doc)))
+        doc.Base_reindexAndSenseAlarm(['invoice_builder_alarm'])
+
+    # Sadly, catalog indexation timestamp has a second precision
+    # It is needed to wait this 1 second to be able to verify new indexation
+    sleep(1)
+    with TemporaryAlarmScript(alarm, 'Alarm_buildInvoice'):
+      self.tic()
+
+    edit_timestamp = alarm.getModificationDate()
+    for doc, previous_indexation_timestamp in document_list:
+      next_indexation_timestamp = self.getIndexationDate(doc)
+      # check that the document has been reindexed
+      self.assertLessThan(previous_indexation_timestamp, next_indexation_timestamp)
+      # check that alarm was called after the object was reindexed
+      self.assertLessThan(next_indexation_timestamp, edit_timestamp)
+
+    self.assertEqual(
+      len(alarm.workflow_history['edit_workflow']),
+      workflow_history_count + 1
+    )
+    self.assertEqual(
+      'Visited by Alarm_buildInvoice',
+      alarm.workflow_history['edit_workflow'][-1]['comment']
+    )
+
   def test_reindexAndSenseAlarm_twoContextDifferentTransaction(self):
     # Check that the script wait for the alarm to be not activate
     # before triggering it again
@@ -434,3 +469,91 @@ class TestBase_reindexAndSenseAlarm(ERP5TypeTestCase):
       'Visited by Alarm_buildPackingList',
       alarm2.workflow_history['edit_workflow'][-1]['comment']
     )
+
+  def test_reindexAndSenseAlarm_concurrentNode(self):
+    # No idea how to reproduce multiple zope nodes handling
+    # many activity, try to fake found problem
+    # Ensure Base_reindexAndSenseAlarm never completely drop itself
+    # from activities
+    alarm_tool = self.portal.portal_alarms
+    alarm = alarm_tool.invoice_builder_alarm
+    workflow_history_count = len(alarm.workflow_history['edit_workflow'])
+
+    deduplication_tag = 'Base_reindexAndSenseAlarm_%s' % alarm.getId()
+    wait_tag = 'wait_tag'
+    alarm_tool.activate(
+      activity='SQLQueue',
+      tag=wait_tag
+    ).getId()
+    alarm_tool.activate(
+      activity='SQLQueue',
+      tag=deduplication_tag,
+      after_tag=wait_tag
+    ).getId()
+    alarm_tool.activate(
+      tag=deduplication_tag
+    ).Base_reindexAndSenseAlarm([alarm.getId()], must_reindex_context=False)
+
+    with TemporaryAlarmScript(alarm, 'Alarm_buildInvoice'):
+      self.tic()
+
+    self.assertEqual(
+      len(alarm.workflow_history['edit_workflow']),
+      workflow_history_count + 1
+    )
+    self.assertEqual(
+      'Visited by Alarm_buildInvoice',
+      alarm.workflow_history['edit_workflow'][-1]['comment']
+    )
+
+
+  def test_reindexAndSenseAlarm_concurrentNodeAnd2Alarms(self):
+    # No idea how to reproduce multiple zope nodes handling
+    # many activity, try to fake found problem
+    # Ensure Base_reindexAndSenseAlarm never completely drop itself
+    # from activities
+    alarm_tool = self.portal.portal_alarms
+    alarm1 = self.portal.portal_alarms.invoice_builder_alarm
+    alarm2 = self.portal.portal_alarms.packing_list_builder_alarm
+
+    workflow_history_count1 = len(alarm1.workflow_history['edit_workflow'])
+    workflow_history_count2 = len(alarm2.workflow_history['edit_workflow'])
+
+    for alarm in [alarm1, alarm2]:
+      deduplication_tag = 'Base_reindexAndSenseAlarm_%s' % alarm.getId()
+      wait_tag = 'wait_tag'
+      alarm_tool.activate(
+        activity='SQLQueue',
+        tag=wait_tag
+      ).getId()
+      alarm_tool.activate(
+        activity='SQLQueue',
+        tag=deduplication_tag,
+        after_tag=wait_tag
+      ).getId()
+      alarm_tool.activate(
+        activity='SQLQueue',
+        tag=deduplication_tag
+      ).Base_reindexAndSenseAlarm([alarm.getId()], must_reindex_context=False)
+
+    with TemporaryAlarmScript(alarm1, 'Alarm_buildInvoice'):
+      with TemporaryAlarmScript(alarm2, 'Alarm_buildPackingList'):
+        self.tic()
+
+    for alarm, workflow_history_count, comment in [
+      (alarm1, workflow_history_count1, 'Visited by Alarm_buildInvoice'),
+      (alarm2, workflow_history_count2, 'Visited by Alarm_buildPackingList')
+    ]:
+      self.assertEqual(
+        len(alarm.workflow_history['edit_workflow']),
+        workflow_history_count + 1,
+        'alarm %s not called %s %s' % (
+          alarm.getId(),
+          len(alarm.workflow_history['edit_workflow']),
+          workflow_history_count + 1
+        )
+      )
+      self.assertEqual(
+        comment,
+        alarm.workflow_history['edit_workflow'][-1]['comment']
+      )

product/ERP5/bootstrap/erp5_core/SkinTemplateItem/portal_skins/erp5_core/Base_reindexAndSenseAlarm.py

@@ -20,6 +20,7 @@ if alarm_tool.isSubscribed() and len(alarm_id_list):
   for alarm_id in alarm_id_list:
     alarm = alarm_tool.restrictedTraverse(alarm_id)
     deduplication_tag = 'Base_reindexAndSenseAlarm_%s' % alarm_id
+    after_alarm_tag = 'afterAlarm_%s' % deduplication_tag
 
     if alarm.isEnabled():
       # do nothing if the alarm is not enabled
@@ -34,22 +35,39 @@ if alarm_tool.isSubscribed() and len(alarm_id_list):
         alarm_tool.activate(**activate_kw).Base_reindexAndSenseAlarm([alarm_id],
                                                                      must_reindex_context=False)
 
-      elif portal.portal_activities.countMessageWithTag(deduplication_tag) <= 1:
-        if alarm.isActive():
-          # If the alarm is active, wait for it
-          # and try to reduce the number of activities
-          # to reduce the number of alarm execution
-          activate_kw = {}
-          activate_kw['activity'] = 'SQLQueue'
-          activate_kw['priority'] = PRIORITY
-          activate_kw['tag'] = deduplication_tag
-          activate_kw['after_path'] = alarm.getPath()
-          # Wait for the previous alarm run to be finished
-          # call on alarm tool to gather and drop with sqldict
-          alarm_tool.activate(**activate_kw).Base_reindexAndSenseAlarm([alarm_id],
-                                                                       must_reindex_context=False)
-        else:
-          # activeSense create an activity in SQLDict
-          alarm.activeSense()
-          # Prevent 2 nodes to call activateSense concurrently
-          alarm.serialize()
+      elif 1 < (portal.portal_activities.countMessageWithTag(deduplication_tag) +
+                portal.portal_activities.countMessageWithTag(after_alarm_tag)):
+        # If there are multiple scripts trying to run concurrently
+        # try to reduce the number of activities
+        # to reduce the number of alarm execution
+        activate_kw = {}
+        activate_kw['activity'] = 'SQLDict'
+        activate_kw['priority'] = PRIORITY
+        activate_kw['tag'] = deduplication_tag
+        # wait for the after_alarm_tag to be finish
+        # use after_path instead of after_tag
+        # to prevent infinite lock
+        activate_kw['after_path'] = alarm_tool.getPath()
+        # Wait for the previous alarm run to be finished
+        # call on alarm tool to gather and drop with sqldict
+        alarm.activate(**activate_kw).Base_reindexAndSenseAlarm([alarm_id],
+                                                                must_reindex_context=False)
+
+      elif alarm.isActive():
+        # If the alarm is active, wait for it
+        # and try to reduce the number of activities
+        # to reduce the number of alarm execution
+        activate_kw = {}
+        activate_kw['activity'] = 'SQLQueue'
+        activate_kw['priority'] = PRIORITY
+        activate_kw['tag'] = after_alarm_tag
+        activate_kw['after_path'] = alarm.getPath()
+        # Wait for the previous alarm run to be finished
+        # call on alarm tool to gather and drop with sqldict
+        alarm_tool.activate(**activate_kw).Base_reindexAndSenseAlarm([alarm_id],
+                                                                     must_reindex_context=False)
+      else:
+        # activeSense create an activity in SQLDict
+        alarm.activeSense()
+        # Prevent 2 nodes to call activateSense concurrently
+        alarm.serialize()