process_repeater_stub() task

corehq/motech/repeaters/const.py

@@ -5,6 +5,13 @@
 MIN_RETRY_WAIT = timedelta(minutes=60)
 CHECK_REPEATERS_INTERVAL = timedelta(minutes=5)
 CHECK_REPEATERS_KEY = 'check-repeaters-key'
+# Number of attempts to an online endpoint before cancelling payload
+MAX_ATTEMPTS = 3
+# Number of exponential backoff attempts to an offline endpoint
+MAX_BACKOFF_ATTEMPTS = 6
+# Limit the number of records to forward at a time so that one repeater
+# can't hold up the rest.
+RECORDS_AT_A_TIME = 1000
 
 RECORD_PENDING_STATE = 'PENDING'
 RECORD_SUCCESS_STATE = 'SUCCESS'

corehq/motech/repeaters/models.py

@@ -63,9 +63,11 @@
 "Data Forwarding Records".
 
 """
+import traceback
 import warnings
 from collections import OrderedDict
 from datetime import datetime, timedelta
+from typing import Any, Optional
 from urllib.parse import parse_qsl, urlencode, urlparse, urlunparse
 
 from django.utils.functional import cached_property
@@ -120,6 +122,8 @@
 from corehq.util.quickcache import quickcache
 
 from .const import (
+    MAX_ATTEMPTS,
+    MAX_BACKOFF_ATTEMPTS,
     MAX_RETRY_WAIT,
     MIN_RETRY_WAIT,
     RECORD_CANCELLED_STATE,
@@ -208,6 +212,38 @@ class Meta:
             models.Index(fields=['repeater_id']),
         ]
 
+    @property
+    @memoized
+    def repeater(self):
+        return Repeater.get(self.repeater_id)
+
+    @property
+    def repeat_records_ready(self):
+        return self.repeat_records.filter(state__in=(RECORD_PENDING_STATE,
+                                                     RECORD_FAILURE_STATE))
+
+    @property
+    def is_ready(self):
+        if self.is_paused:
+            return False
+        if not (self.next_attempt_at is None
+                or self.next_attempt_at < timezone.now()):
+            return False
+        return self.repeat_records_ready.exists()
+
+    def set_next_attempt(self):
+        now = datetime.utcnow()
+        interval = _get_retry_interval(self.last_attempt_at, now)
+        self.last_attempt_at = now
+        self.next_attempt_at = now + interval
+        self.save()
+
+    def reset_next_attempt(self):
+        if self.last_attempt_at or self.next_attempt_at:
+            self.last_attempt_at = None
+            self.next_attempt_at = None
+            self.save()
+
 
 class Repeater(QuickCachedDocumentMixin, Document):
     """
@@ -1021,6 +1057,75 @@ class Meta:
         ]
         ordering = ['registered_at']
 
+    def add_success_attempt(self, response):
+        """
+        ``response`` can be a Requests response instance, or True if the
+        payload did not result in an API call.
+        """
+        self.repeater_stub.reset_next_attempt()
+        self.sqlrepeatrecordattempt_set.create(
+            state=RECORD_SUCCESS_STATE,
+            message=format_response(response),
+        )
+        self.state = RECORD_SUCCESS_STATE
+        self.save()
+
+    def add_client_failure_attempt(self, message):
+        """
+        Retry when ``self.repeater`` is next processed. The remote
+        service is assumed to be in a good state, so do not back off, so
+        that this repeat record does not hold up the rest.
+        """
+        self.repeater_stub.reset_next_attempt()
+        if self.num_attempts < MAX_ATTEMPTS:
+            state = RECORD_FAILURE_STATE
+        else:
+            state = RECORD_CANCELLED_STATE
+        self._add_failure_attempt(message, state)
+
+    def add_server_failure_attempt(self, message):
+        """
+        Server and connection failures are retried later with
+        exponential backoff.
+
+        .. note::
+           ONLY CALL THIS IF RETRYING MUCH LATER STANDS A CHANCE OF
+           SUCCEEDING. Exponential backoff will continue for several
+           days and will hold up all other payloads.
+
+        """
+        self.repeater_stub.set_next_attempt()
+        if self.num_attempts < MAX_BACKOFF_ATTEMPTS:
+            state = RECORD_FAILURE_STATE
+        else:
+            state = RECORD_CANCELLED_STATE
+        self._add_failure_attempt(message, state)
+
+    def _add_failure_attempt(self, message, state):
+        self.sqlrepeatrecordattempt_set.create(
+            state=state,
+            message=message,
+        )
+        self.state = state
+        self.save()
+
+    def add_payload_exception_attempt(self, message, tb_str):
+        self.sqlrepeatrecordattempt_set.create(
+            state=RECORD_CANCELLED_STATE,
+            message=message,
+            traceback=tb_str,
+        )
+        self.state = RECORD_CANCELLED_STATE
+        self.save()
+
+    @property
+    def attempts(self):
+        return self.sqlrepeatrecordattempt_set.all()
+
+    @property
+    def num_attempts(self):
+        return self.sqlrepeatrecordattempt_set.count()
+
 
 class SQLRepeatRecordAttempt(models.Model):
     repeat_record = models.ForeignKey(SQLRepeatRecord,
@@ -1054,6 +1159,100 @@ def _get_retry_interval(last_checked, now):
     return interval
 
 
+def attempt_forward_now(repeater_stub: RepeaterStub):
+    from corehq.motech.repeaters.tasks import process_repeater_stub
+
+    if not domain_can_forward(repeater_stub.domain):
+        return
+    if not repeater_stub.is_ready:
+        return
+    process_repeater_stub.delay(repeater_stub)
+
+
+def get_payload(repeater: Repeater, repeat_record: SQLRepeatRecord) -> Any:
+    try:
+        return repeater.get_payload(repeat_record)
+    except Exception as err:
+        log_repeater_error_in_datadog(
+            repeater.domain,
+            status_code=None,
+            repeater_type=repeater.__class__.__name__
+        )
+        repeat_record.add_payload_exception_attempt(
+            message=str(err),
+            tb_str=traceback.format_exc()
+        )
+        raise
+
+
+def send_request(
+    repeater: Repeater,
+    repeat_record: SQLRepeatRecord,
+    payload: Any,
+) -> bool:
+    """
+    Calls ``repeater.send_request()`` and handles the result.
+
+    Returns True on success or cancelled, so that the caller knows
+    whether to retry later.
+    """
+
+    def is_success(resp):
+        return (
+            is_response(resp)
+            and 200 <= resp.status_code < 300
+            # `response` is `True` if the payload did not need to be
+            # sent. (This can happen, for example, with DHIS2 if the
+            # form that triggered the forwarder doesn't contain data
+            # for a DHIS2 Event.)
+            or resp is True
+        )
+
+    def later_might_be_better(resp):
+        return is_response(resp) and resp.status_code in (
+            502,  # Bad Gateway
+            503,  # Service Unavailable
+            504,  # Gateway Timeout
+        )
+
+    try:
+        response = repeater.send_request(repeat_record, payload)
+    except (Timeout, ConnectionError) as err:
+        log_repeater_timeout_in_datadog(repeat_record.domain)
+        message = str(RequestConnectionError(err))
+        repeat_record.add_server_failure_attempt(message)
+    except Exception as err:
+        repeat_record.add_client_failure_attempt(str(err))
+    else:
+        if is_success(response):
+            if is_response(response):
+                # Don't bother logging success in
+                # Datadog if the payload wasn't sent.
+                log_repeater_success_in_datadog(
+                    repeater.domain,
+                    response.status_code,
+                    repeater_type=repeater.__class__.__name__
+                )
+            repeat_record.add_success_attempt(response)
+        else:
+            message = format_response(response)
+            if later_might_be_better(response):
+                repeat_record.add_server_failure_attempt(message)
+            else:
+                repeat_record.add_client_failure_attempt(message)
+    return repeat_record.state in (RECORD_SUCCESS_STATE,
+                                   RECORD_CANCELLED_STATE)  # Don't retry
+
+
+def format_response(response) -> Optional[str]:
+    if not is_response(response):
+        return None
+    response_text = getattr(response, "text", "")
+    if response_text:
+        return f'{response.status_code}: {response.reason}\n{response_text}'
+    return f'{response.status_code}: {response.reason}'
+
+
 def is_response(duck):
     """
     Returns True if ``duck`` has the attributes of a Requests response

corehq/motech/repeaters/tasks.py

@@ -6,7 +6,7 @@
 from celery.task import periodic_task, task
 from celery.utils.log import get_task_logger
 
-from dimagi.utils.couch import get_redis_lock
+from dimagi.utils.couch import CriticalSection, get_redis_lock
 from dimagi.utils.couch.undo import DELETED_SUFFIX
 
 from corehq.motech.models import RequestLog
@@ -25,12 +25,18 @@
     MAX_RETRY_WAIT,
     RECORD_FAILURE_STATE,
     RECORD_PENDING_STATE,
+    RECORDS_AT_A_TIME,
 )
 from .dbaccessors import (
     get_overdue_repeat_record_count,
     iterate_repeat_records,
 )
-from .models import domain_can_forward
+from .models import (
+    RepeaterStub,
+    domain_can_forward,
+    get_payload,
+    send_request,
+)
 
 _check_repeaters_buckets = make_buckets_from_timedeltas(
     timedelta(seconds=10),
@@ -150,3 +156,28 @@ def process_repeat_record(repeat_record):
     run_every=crontab(),  # every minute
     multiprocess_mode=MPM_MAX
 )
+
+
+@task(serializer='pickle', queue=settings.CELERY_REPEAT_RECORD_QUEUE)
+def process_repeater_stub(repeater_stub: RepeaterStub):
+    """
+    Worker task to send SQLRepeatRecords in chronological order.
+
+    This function assumes that ``repeater_stub`` checks have already
+    been performed. Call via ``models.attempt_forward_now()``.
+    """
+    with CriticalSection(
+        [f'process-repeater-{repeater_stub.repeater_id}'],
+        fail_hard=False, block=False, timeout=5 * 60 * 60,
+    ):
+        for repeat_record in repeater_stub.repeat_records_ready[:RECORDS_AT_A_TIME]:
+            try:
+                payload = get_payload(repeater_stub.repeater, repeat_record)
+            except Exception:
+                # The repeat record is cancelled if there is an error
+                # getting the payload. We can safely move to the next one.
+                continue
+            succeeded_or_cancelled = send_request(repeater_stub.repeater,
+                                                  repeat_record, payload)
+            if not succeeded_or_cancelled:
+                break  # Retry later