Process Repeaters, Part 1

corehq/ex-submodules/dimagi/utils/couch/tests/test_redis_lock.py

@@ -0,0 +1,31 @@
+import uuid
+
+from redis.lock import Lock as RedisLock
+
+from dimagi.utils.couch import get_redis_lock
+
+from corehq.tests.noseplugins.redislocks import TestLock
+from corehq.util.metrics.lockmeter import MeteredLock
+
+
+def test_get_redis_lock_with_token():
+    lock_name = 'test-1'
+    metered_lock = get_redis_lock(key=lock_name, name=lock_name, timeout=1)
+    assert isinstance(metered_lock, MeteredLock)
+    # metered_lock.lock is a TestLock instance because of
+    # corehq.tests.noseplugins.redislocks.RedisLockTimeoutPlugin
+    test_lock = metered_lock.lock
+    assert isinstance(test_lock, TestLock)
+    redis_lock = test_lock.lock
+    assert isinstance(redis_lock, RedisLock)
+
+    token = uuid.uuid1().hex
+    acquired = redis_lock.acquire(blocking=False, token=token)
+    assert acquired
+
+    # What we want to be able to do in a separate process:
+    metered_lock_2 = get_redis_lock(key=lock_name, name=lock_name, timeout=1)
+    redis_lock_2 = metered_lock_2.lock.lock
+    redis_lock_2.local.token = token
+    # Does not raise LockNotOwnedError:
+    redis_lock_2.release()

corehq/motech/repeaters/const.py

@@ -13,6 +13,8 @@
 CHECK_REPEATERS_INTERVAL = timedelta(minutes=5)
 CHECK_REPEATERS_PARTITION_COUNT = settings.CHECK_REPEATERS_PARTITION_COUNT
 CHECK_REPEATERS_KEY = 'check-repeaters-key'
+PROCESS_REPEATERS_INTERVAL = timedelta(minutes=1)
+PROCESS_REPEATERS_KEY = 'process-repeaters-key'
 ENDPOINT_TIMER = 'endpoint_timer'
 # Number of attempts to an online endpoint before cancelling payload
 MAX_ATTEMPTS = 3

corehq/motech/repeaters/migrations/0014_repeater_max_workers.py

@@ -0,0 +1,16 @@
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("repeaters", "0013_alter_repeatrecord_state_and_more"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="repeater",
+            name="max_workers",
+            field=models.IntegerField(default=0),
+        ),
+    ]

corehq/motech/repeaters/migrations/0015_alter_repeatrecord_state_and_more.py

@@ -0,0 +1,35 @@
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("repeaters", "0014_repeater_max_workers"),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name="repeatrecord",
+            name="state",
+            field=models.PositiveSmallIntegerField(
+                choices=[
+                    (1, "Pending"),
+                    (2, "Failed"),
+                    (4, "Succeeded"),
+                    (8, "Cancelled"),
+                    (16, "Empty"),
+                    (32, "Invalid Payload"),
+                ],
+                db_index=True,
+                default=1,
+            ),
+        ),
+        migrations.AddIndex(
+            model_name="repeater",
+            index=models.Index(
+                condition=models.Q(("is_paused", False)),
+                fields=["next_attempt_at"],
+                name="next_attempt_at_not_paused_idx",
+            ),
+        ),
+    ]

corehq/motech/repeaters/models.py

@@ -73,6 +73,7 @@
 from http import HTTPStatus
 from urllib.parse import parse_qsl, urlencode, urlparse, urlunparse
 
+from django.conf import settings
 from django.db import models, router
 from django.db.models.base import Deferred
 from django.dispatch import receiver
@@ -230,10 +231,19 @@ def all_ready(self):
         repeat_records_ready_to_send = models.Q(
             repeat_records__state__in=(State.Pending, State.Fail)
         )
-        return (self.get_queryset()
-                .filter(not_paused)
-                .filter(next_attempt_not_in_the_future)
-                .filter(repeat_records_ready_to_send))
+        return (
+            self.get_queryset()
+            .filter(not_paused)
+            .filter(next_attempt_not_in_the_future)
+            .filter(repeat_records_ready_to_send)
+        )
+
+    def get_all_ready_ids_by_domain(self):
+        results = defaultdict(list)
+        query = self.all_ready().values_list('domain', 'id')
+        for (domain, id_uuid) in query.all():
+            results[domain].append(id_uuid.hex)
+        return results
 
     def get_queryset(self):
         repeater_obj = self.model()
@@ -260,6 +270,7 @@ class Repeater(RepeaterSuperProxy):
     is_paused = models.BooleanField(default=False)
     next_attempt_at = models.DateTimeField(null=True, blank=True)
     last_attempt_at = models.DateTimeField(null=True, blank=True)
+    max_workers = models.IntegerField(default=0)
     options = JSONField(default=dict)
     connection_settings_id = models.IntegerField(db_index=True)
     is_deleted = models.BooleanField(default=False, db_index=True)
@@ -271,6 +282,13 @@ class Repeater(RepeaterSuperProxy):
 
     class Meta:
         db_table = 'repeaters_repeater'
+        indexes = [
+            models.Index(
+                fields=['next_attempt_at'],
+                condition=models.Q(is_paused=False),
+                name='next_attempt_at_not_paused_idx',
+            ),
+        ]
 
     payload_generator_classes = ()
 
@@ -350,9 +368,24 @@ def _repeater_type(cls):
 
     @property
     def repeat_records_ready(self):
-        return self.repeat_records.filter(state__in=(State.Pending, State.Fail))
+        """
+        A QuerySet of repeat records in the Pending or Fail state in the
+        order in which they were registered
+        """
+        return (
+            self.repeat_records
+            .filter(state__in=(State.Pending, State.Fail))
+            .order_by('registered_at')
+        )
 
-    def set_next_attempt(self):
+    @property
+    def num_workers(self):
+        # If num_workers is 1, repeat records are sent in the order in
+        # which they were registered.
+        num_workers = self.max_workers or settings.DEFAULT_REPEATER_WORKERS
+        return min(num_workers, settings.MAX_REPEATER_WORKERS)
+
+    def set_backoff(self):
         now = datetime.utcnow()
         interval = _get_retry_interval(self.last_attempt_at, now)
         self.last_attempt_at = now
@@ -365,7 +398,7 @@ def set_next_attempt(self):
             next_attempt_at=now + interval,
         )
 
-    def reset_next_attempt(self):
+    def reset_backoff(self):
         if self.last_attempt_at or self.next_attempt_at:
             self.last_attempt_at = None
             self.next_attempt_at = None
@@ -977,11 +1010,17 @@ def get_domains_with_records(self):
 class RepeatRecord(models.Model):
     domain = models.CharField(max_length=126)
     payload_id = models.CharField(max_length=255)
-    repeater = models.ForeignKey(Repeater,
-                                 on_delete=DB_CASCADE,
-                                 db_column="repeater_id_",
-                                 related_name='repeat_records')
-    state = models.PositiveSmallIntegerField(choices=State.choices, default=State.Pending)
+    repeater = models.ForeignKey(
+        Repeater,
+        on_delete=DB_CASCADE,
+        db_column="repeater_id_",
+        related_name='repeat_records',
+    )
+    state = models.PositiveSmallIntegerField(
+        choices=State.choices,
+        default=State.Pending,
+        db_index=True,
+    )
     registered_at = models.DateTimeField()
     next_check = models.DateTimeField(null=True, default=None)
     max_possible_tries = models.IntegerField(default=MAX_BACKOFF_ATTEMPTS)
@@ -1161,7 +1200,9 @@ def fire(self, force_send=False, timing_context=None):
                 self.repeater.fire_for_record(self, timing_context=timing_context)
             except Exception as e:
                 self.handle_payload_error(str(e), traceback_str=traceback.format_exc())
-                raise
+            finally:
+                return self.state
+        return None
 
     def attempt_forward_now(self, *, is_retry=False, fire_synchronously=False):
         from corehq.motech.repeaters.tasks import (
@@ -1171,6 +1212,19 @@ def attempt_forward_now(self, *, is_retry=False, fire_synchronously=False):
             retry_process_datasource_repeat_record,
         )
 
+        def is_new_synchronous_case_repeater_record():
+            """
+            Repeat record is a new record for a synchronous case repeater
+            See corehq.motech.repeaters.signals.fire_synchronous_case_repeaters
+            """
+            return fire_synchronously and self.state == State.Pending
+
+        if (
+            toggles.PROCESS_REPEATERS.enabled(self.domain, toggles.NAMESPACE_DOMAIN)
+            and not is_new_synchronous_case_repeater_record()
+        ):
+            return
+
         if self.next_check is None or self.next_check > datetime.utcnow():
             return
 
@@ -1323,7 +1377,27 @@ def is_response(duck):
 
 
 def domain_can_forward(domain):
+    """
+    Returns whether ``domain`` has data forwarding or Zapier integration
+    privileges.
+
+    Used for determining whether to register a repeat record.
+    """
     return domain and (
         domain_has_privilege(domain, ZAPIER_INTEGRATION)
         or domain_has_privilege(domain, DATA_FORWARDING)
     )
+
+
+@quickcache(['domain'], timeout=60)
+def domain_can_forward_now(domain):
+    """
+    Returns ``True`` if ``domain`` has the requisite privileges and data
+    forwarding is not paused.
+
+    Used for determining whether to send a repeat record now.
+    """
+    return (
+        domain_can_forward(domain)
+        and not toggles.PAUSE_DATA_FORWARDING.enabled(domain)
+    )