Changeset c987be5 in rattail

Timestamp:

07/19/2022 10:39:41 AM (7 months ago)

Author:

Lance Edgar <lance@…>

Branches:

master

Children:

27d18c42

Parents:

9e8fe47

Message:

Add retry loop for starting transaction in datasync consumer

File:

• rattail/datasync/daemon.py (modified) (1 diff)

rattail/datasync/daemon.py

@@ -570 +570 @@
     log.debug("will process %s changes from %s", len(changes),
               app.localtime(obtained, from_utc=True))
-    consumer.begin_transaction()
-
+
+    # first retry loop is to begin the transaction
+    attempts = 0
+    errtype = None
+    while True:
+        attempts += 1
+
+        try:
+            consumer.begin_transaction()
+
+        except Exception as errobj: # processing failed!
+            exc_type, exc, traceback = sys.exc_info()
+
+            # if we've reached our final attempt, stop retrying
+            if attempts >= consumer.retry_attempts:
+                log.warning("attempt #%s failed calling `consumer.begin_transaction()`; "
+                            "this thread will now *terminate* until datasync restart",
+                            attempts, exc_info=True)
+                app.send_email('datasync_error_consumer_process_changes', {
+                    'watcher': consumer.watcher,
+                    'consumer': consumer,
+                    'error': exc,
+                    'attempts': attempts,
+                    'traceback': ''.join(format_exception(exc_type, exc, traceback)).strip(),
+                    'datasync_url': config.datasync_url(),
+                })
+                return False
+
+            # if this exception is not the first, and is of a different type
+            # than seen previously, do *not* continue to retry
+            if errtype is not None and not isinstance(errobj, errtype):
+                log.exception("new exception differs from previous one(s), "
+                              "giving up on consumer.begin_transaction()")
+                return False
+
+            # record the type of exception seen; maybe pause before next retry
+            errtype = type(errobj)
+            log.warning("attempt #%s failed for '%s' -> '%s' consumer.begin_transaction()",
+                        attempts, consumer.watcher.key, consumer.key)
+            log.debug("pausing for %s seconds before making attempt #%s of %s",
+                      consumer.retry_delay, attempts + 1, consumer.retry_attempts)
+            if consumer.retry_delay:
+                time.sleep(consumer.retry_delay)
+
+        else: # transaction began okay
+
+            # can stop the attempt/retry loop now
+            break
+
+    # second retry loop is to process the changes
     attempts = 0
     errtype = None