Add a time_last_handled field to prevent no-op handlers from clogging the queue

ambition_utils/rrule/migrations/0006_rrule_time_last_handled.py

@@ -0,0 +1,18 @@
+# Generated by Django 4.2.20 on 2025-03-24 21:15
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('rrule', '0005_auto_20230802_1548'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='rrule',
+            name='time_last_handled',
+            field=models.DateTimeField(db_index=True, default=None, null=True),
+        ),
+    ]

ambition_utils/rrule/models.py

@@ -65,11 +65,22 @@ def process_occurrence_handler_paths(self, limit=None, **kwargs):
         # Build a list of rrules that get returned from the handler
         rrules = []
         for instance in instances:
-            rrules.extend(instance.handle())
+            handler_rrules = instance.handle()
+            if handler_rrules:
+                rrules.extend(handler_rrules)
 
         # Bulk update the next occurrences
         RRule.objects.update_next_occurrences(rrule_objects=rrules)
 
+        # Update the last handled time
+        handler_paths = [
+            f"{instance.__class__.__module__}.{instance.__class__.__name__}"
+            for instance in instances
+        ]
+        RRule.objects.filter(
+            occurrence_handler_path__in=handler_paths
+        ).update(time_last_handled=datetime.utcnow())
+
     def process_related_model_handlers(self, limit=None, filters=None):
         # Get the rrule objects that are overdue and need to be handled
         rrule_objects = self.get_queryset().filter(
@@ -78,6 +89,7 @@ def process_related_model_handlers(self, limit=None, filters=None):
             related_object_id__isnull=False,
             **(filters or {})
         ).order_by(
+            F('time_last_handled').asc(nulls_first=True),
             'next_occurrence',
             'id'
         ).prefetch_related(
@@ -86,6 +98,8 @@ def process_related_model_handlers(self, limit=None, filters=None):
         if limit:
             rrule_objects = rrule_objects[:limit]
 
+        rrule_objects = list(rrule_objects)
+        rrule_objects_ids = [rrule_object.id for rrule_object in rrule_objects]
         rrules_to_advance = []
         for rrule_object in rrule_objects:
             if hasattr(rrule_object.related_object, rrule_object.related_object_handler_name):
@@ -98,6 +112,10 @@ def process_related_model_handlers(self, limit=None, filters=None):
         # Bulk update the next occurrences
         rrules_to_advance = RRule.objects.update_next_occurrences(rrule_objects=rrules_to_advance)
 
+        # Update the last handled time
+        RRule.objects.filter(id__in=rrule_objects_ids).update(time_last_handled=datetime.utcnow())
+
+        # Return the rules we advanced
         return rrules_to_advance
 
     def overdue_handler_class_instances(self, limit=None, **kwargs):
@@ -118,6 +136,7 @@ def overdue_handler_class_instances(self, limit=None, **kwargs):
         ).filter(
             row_num=1
         ).order_by(
+            F('time_last_handled').asc(nulls_first=True),
             'next_occurrence',
             'occurrence_handler_path'
         )
@@ -171,6 +190,9 @@ class RRule(models.Model):
     # An optional number of days to offset occurrences by
     day_offset = models.SmallIntegerField(blank=True, null=True)
 
+    # The last time we handled this rrule for overdue occurrences
+    time_last_handled = models.DateTimeField(null=True, default=None, db_index=True)
+
     # Custom object manager
     objects = RRuleManager()
 

ambition_utils/rrule/tests/model_tests.py

@@ -50,6 +50,12 @@ def handle(self):
         ).order_by('id')
 
 
+class HandlerNoOp(OccurrenceHandler):
+
+    def handle(self):
+        return None
+
+
 class RRuleManagerTest(TestCase):
     def add_test_rules(self):
         # Make a program with HandlerOne that is not overdue
@@ -242,8 +248,8 @@ def test_run_with_limit(self):
         # except for handler two which should occur three times
         expected_classes = [
             {HandlerTwo},
-            {HandlerOne},
             {HandlerThree},
+            {HandlerOne},
             {HandlerTwo},
             {HandlerOne},
             {HandlerThree},
@@ -266,6 +272,64 @@ def test_run_with_limit(self):
         self.assertEqual(recurrences[2].next_occurrence, datetime.datetime(2017, 1, 4))
         self.assertEqual(recurrences[3].next_occurrence, datetime.datetime(2017, 1, 4))
 
+    def test_run_with_no_op(self):
+        """
+        Test running with a no-op handler
+        """
+        rrule_params = {
+            'freq': rrule.DAILY,
+            'interval': 1,
+            'dtstart': datetime.datetime(2017, 1, 2),
+        }
+        no_op_rrule1 = G(
+            RRule,
+            rrule_params=rrule_params,
+            occurrence_handler_path='ambition_utils.rrule.tests.model_tests.HandlerNoOp'
+        )
+        no_op_rrule2 = G(
+            RRule,
+            rrule_params=rrule_params,
+            occurrence_handler_path='ambition_utils.rrule.tests.model_tests.HandlerNoOp'
+        )
+        rrule1 = G(
+            RRule,
+            rrule_params=rrule_params,
+            occurrence_handler_path='ambition_utils.rrule.tests.model_tests.HandlerOne'
+        )
+
+        # Process with a limit of 1, we should handle the no-op rrules
+        with freeze_time(datetime.datetime(2017, 1, 2)):
+            RRule.objects.handle_overdue(occurrence_handler_limit=1)
+
+        no_op_rrule1.refresh_from_db()
+        no_op_rrule2.refresh_from_db()
+        rrule1.refresh_from_db()
+        self.assertEqual(no_op_rrule1.time_last_handled, datetime.datetime(2017, 1, 2))
+        self.assertEqual(no_op_rrule2.time_last_handled, datetime.datetime(2017, 1, 2))
+        self.assertEqual(rrule1.time_last_handled, None)
+
+        # Process with a limit of 1, we should handle handler one
+        with freeze_time(datetime.datetime(2017, 1, 3)):
+            RRule.objects.handle_overdue(occurrence_handler_limit=1)
+
+        no_op_rrule1.refresh_from_db()
+        no_op_rrule2.refresh_from_db()
+        rrule1.refresh_from_db()
+        self.assertEqual(no_op_rrule1.time_last_handled, datetime.datetime(2017, 1, 2))
+        self.assertEqual(no_op_rrule2.time_last_handled, datetime.datetime(2017, 1, 2))
+        self.assertEqual(rrule1.time_last_handled, datetime.datetime(2017, 1, 3))
+
+        # Process again and we should go back to the no op handler
+        with freeze_time(datetime.datetime(2017, 1, 4)):
+            RRule.objects.handle_overdue(occurrence_handler_limit=1)
+
+        no_op_rrule1.refresh_from_db()
+        no_op_rrule2.refresh_from_db()
+        rrule1.refresh_from_db()
+        self.assertEqual(no_op_rrule1.time_last_handled, datetime.datetime(2017, 1, 4))
+        self.assertEqual(no_op_rrule2.time_last_handled, datetime.datetime(2017, 1, 4))
+        self.assertEqual(rrule1.time_last_handled, datetime.datetime(2017, 1, 3))
+
 
 class RRuleTest(TestCase):
 
@@ -421,6 +485,89 @@ def test_related_object_handlers_with_limit(self):
             self.assertEqual(program2.start_recurrence.next_occurrence, datetime.datetime(2022, 6, 2, 9))
             self.assertEqual(program2.end_recurrence.next_occurrence, datetime.datetime(2022, 6, 2, 17))
 
+    def test_related_object_handlers_no_op(self):
+        """
+        Test a scenario where the handler of the object is no-op and does not progress the rule
+        """
+
+        no_op_program1 = Program.objects.create(name='Program 1')
+        rrule1 = RRule.objects.create(
+            rrule_params={
+                'freq': rrule.DAILY,
+                'interval': 1,
+                'dtstart': datetime.datetime(2022, 6, 1, 9),
+                'byhour': 9,
+            },
+            related_object=no_op_program1,
+            related_object_handler_name='handle_no_op',
+        )
+        no_op_program1.start_recurrence = rrule1
+        no_op_program1.save()
+
+        no_op_program2 = Program.objects.create(name='Program 1')
+        rrule2 = RRule.objects.create(
+            rrule_params={
+                'freq': rrule.DAILY,
+                'interval': 1,
+                'dtstart': datetime.datetime(2022, 6, 1, 9),
+                'byhour': 9,
+            },
+            related_object=no_op_program1,
+            related_object_handler_name='handle_no_op',
+        )
+        no_op_program2.start_recurrence = rrule2
+        no_op_program2.save()
+
+        program = Program.objects.create(name='Program 1')
+        rrule3 = RRule.objects.create(
+            rrule_params={
+                'freq': rrule.DAILY,
+                'interval': 1,
+                'dtstart': datetime.datetime(2022, 6, 1, 9),
+                'byhour': 9,
+            },
+            related_object=program,
+            related_object_handler_name='handle_start_recurrence',
+        )
+        program.start_recurrence = rrule3
+        program.save()
+
+        # Run with a limit of 1, we should handle the no-op program
+        with freeze_time(datetime.datetime(2022, 6, 1, 9)):
+            RRule.objects.handle_overdue(related_model_handler_limit=1)
+
+        # Make sure only the no-op program is handled
+        no_op_program1.refresh_from_db()
+        no_op_program2.refresh_from_db()
+        program.refresh_from_db()
+        self.assertEqual(no_op_program1.start_recurrence.time_last_handled, datetime.datetime(2022, 6, 1, 9))
+        self.assertEqual(no_op_program2.start_recurrence.time_last_handled, None)
+        self.assertEqual(program.start_recurrence.time_last_handled, None)
+
+        # Run with a limit of 1 again , we should handle the no-op program2
+        with freeze_time(datetime.datetime(2022, 6, 1, 9)):
+            RRule.objects.handle_overdue(related_model_handler_limit=1)
+
+        # Make sure only the no-op program is handled
+        no_op_program1.refresh_from_db()
+        no_op_program2.refresh_from_db()
+        program.refresh_from_db()
+        self.assertEqual(no_op_program1.start_recurrence.time_last_handled, datetime.datetime(2022, 6, 1, 9))
+        self.assertEqual(no_op_program2.start_recurrence.time_last_handled, datetime.datetime(2022, 6, 1, 9))
+        self.assertEqual(program.start_recurrence.time_last_handled, None)
+
+        # Run with a limit of 1 again , we should handle the program
+        with freeze_time(datetime.datetime(2022, 6, 1, 9)):
+            RRule.objects.handle_overdue(related_model_handler_limit=1)
+
+        # Make sure only the program is handled
+        no_op_program1.refresh_from_db()
+        no_op_program2.refresh_from_db()
+        program.refresh_from_db()
+        self.assertEqual(no_op_program1.start_recurrence.time_last_handled, datetime.datetime(2022, 6, 1, 9))
+        self.assertEqual(no_op_program2.start_recurrence.time_last_handled, datetime.datetime(2022, 6, 1, 9))
+        self.assertEqual(program.start_recurrence.time_last_handled, datetime.datetime(2022, 6, 1, 9))
+
     def test_related_object_handlers_invalid_handler(self):
         """
         Hits the else block when the handler path is not valid

ambition_utils/rrule/tests/models.py

@@ -24,3 +24,9 @@ def handle_end_recurrence(self, rrule):
         self.end_called += 1
         self.save()
         return rrule
+
+    def handle_no_op(self, rrule):
+        """
+        A no op handler that does nothing
+        """
+        return None

ambition_utils/version.py

@@ -1 +1 @@
-__version__ = '3.2.1'
+__version__ = '3.2.2'