enh: wait between retries of failed S3 upload requests

CHANGELOG

@@ -36,6 +36,8 @@
    for database interrogators
  - enh: `get_users` of database interrogators accepts `query` and `limit`
    (only up to `limit` users are returned, by default 20)
+ - enh: wait between retries of failed S3 upload requests (5 s, doubling
+   with each retry) instead of retrying immediately
  - ref: convert `dcoraid.cli` module into a package; the upload task CLI
    now lives in `dcoraid.cli.task_upload_task`
  - ref: use `super()`, PEP 604 optional type hints, and builtin/

dcoraid/api/s3_api.py

@@ -6,6 +6,7 @@
 import os
 import pathlib
 import re
+import time
 import traceback
 from collections.abc import Callable
 from typing import BinaryIO
@@ -280,6 +281,31 @@ def get_etag_from_response(response):
     return etag
 
 
+#: Seconds to wait before the first retry of an S3 request; the waiting
+#: time doubles with each retry (see :func:`wait_before_retry`)
+RETRY_WAIT = 5.0
+
+
+def wait_before_retry(attempt: int, attempts: int, retry_wait: float):
+    """Wait before retrying a failed request
+
+    Parameters
+    ----------
+    attempt: int
+        index of the attempt that just failed (starting at 0)
+    attempts: int
+        total number of attempts
+    retry_wait: float
+        seconds to wait after the first attempt; the waiting time doubles
+        with each attempt. There is no waiting after the last attempt.
+    """
+    if attempt < attempts - 1:
+        wait = retry_wait * 2 ** attempt
+        logger.info(f"Retrying in {wait:.0f}s "
+                    f"(attempt {attempt + 2} of {attempts})")
+        time.sleep(wait)
+
+
 def requests_put_data_and_get_etag(
         file_object: io.FileIO | BinaryIO,
         put_url: str,
@@ -289,13 +315,20 @@ def requests_put_data_and_get_etag(
         callback: Callable,
         timeout=27.3,
         retries=3,
+        retry_wait=RETRY_WAIT,
         ):
     """Upload data via a PUT request and return the ETag
 
     Parameters
     ----------
     part_number: int
         multipart upload part number, indexing starts at 0
+    retries: int
+        number of attempts
+    retry_wait: float
+        seconds to wait before the first retry (doubles with each retry),
+        so that short connection problems (e.g. refused connections)
+        do not immediately lead to a failed upload
 
 
     Since this method makes use of :class:`.FilePart` which
@@ -310,7 +343,7 @@ def requests_put_data_and_get_etag(
                        )
     trcbck = None
     resp = None
-    for _ in range(retries):
+    for attempt in range(retries):
         # Always seek to the zero position.
         fd_part.seek(0)
         try:
@@ -322,13 +355,15 @@ def requests_put_data_and_get_etag(
         except BaseException:
             trcbck = traceback.format_exc()
             logger.warning(f"Encountered Exception for {put_url}:\n{trcbck} ")
+            wait_before_retry(attempt, retries, retry_wait)
             continue
         else:
             trcbck = None
             # Obtain the ETag from the headers
             etag = get_etag_from_response(resp)
             if etag is not None and etag == fd_part.md5():
                 break
+            wait_before_retry(attempt, retries, retry_wait)
     else:
         raise S3UploadError(
             f"Could not upload or server did not return 'ETag' for upload "
@@ -561,6 +596,7 @@ def upload_s3_presigned_multipart(
             trcbck_compl = traceback.format_exc()
             logger.warning(
                 f"Encountered {trcbck_compl} for {complete_multipart_url}")
+            wait_before_retry(jj, retries, RETRY_WAIT)
             continue
         else:
             etag_full = get_etag_from_response(resp_compl)
@@ -572,6 +608,7 @@ def upload_s3_presigned_multipart(
                     logger.warning(f"ETag mismatch, expected {etag_expected}, "
                                    f"got {etag_full}; (retry {jj})")
                     # The server returned the wrong ETag. We will try again.
+                    wait_before_retry(jj, retries, RETRY_WAIT)
                     continue
             else:
                 # Some servers do not properly return the ETag in the response
@@ -585,6 +622,7 @@ def upload_s3_presigned_multipart(
                         f"Server did not return ETag in the response header "
                         f"and returned an error message in the body: {rbody}"
                     )
+                    wait_before_retry(jj, retries, RETRY_WAIT)
                     continue
                 else:
                     # We have to assume everything went well.

tests/test_api_s3_retry.py

@@ -0,0 +1,83 @@
+"""Offline tests for retrying S3 uploads"""
+import hashlib
+import io
+from unittest import mock
+
+import pytest
+import requests
+
+from dcoraid.api import s3_api
+from dcoraid.api.errors import S3UploadError
+
+DATA = b"Hello S3!" * 1000
+MD5 = hashlib.md5(DATA).hexdigest()
+
+
+def make_response(etag):
+    resp = mock.MagicMock()
+    resp.headers = {"ETag": f'"{etag}"'}
+    resp.content = b""
+    return resp
+
+
+def put_part(**kwargs):
+    return s3_api.requests_put_data_and_get_etag(
+        file_object=io.BytesIO(DATA),
+        put_url="https://objectstore.example.com/bucket/object",
+        part_number=0,
+        part_size=len(DATA),
+        file_size=len(DATA),
+        callback=None,
+        **kwargs)
+
+
+def test_wait_before_retry():
+    with mock.patch.object(s3_api.time, "sleep") as sleep:
+        for attempt in range(4):
+            s3_api.wait_before_retry(attempt, attempts=4, retry_wait=5)
+    # no waiting after the last attempt
+    assert [c.args[0] for c in sleep.call_args_list] == [5, 10, 20]
+
+
+def test_put_retry_after_refused_connection():
+    """Connection problems (e.g. refused connections) are retried later"""
+    refused = requests.exceptions.ConnectionError("connection refused")
+    with mock.patch.object(s3_api.requests, "put",
+                           side_effect=[refused, refused,
+                                        make_response(MD5)]) as put, \
+            mock.patch.object(s3_api.time, "sleep") as sleep:
+        etag = put_part(retries=3)
+    assert etag == MD5
+    assert put.call_count == 3
+    assert [c.args[0] for c in sleep.call_args_list] == [5, 10]
+
+
+def test_put_retry_after_etag_mismatch():
+    with mock.patch.object(s3_api.requests, "put",
+                           side_effect=[make_response("0" * 32),
+                                        make_response(MD5)]) as put, \
+            mock.patch.object(s3_api.time, "sleep") as sleep:
+        etag = put_part(retries=3, retry_wait=2)
+    assert etag == MD5
+    assert put.call_count == 2
+    assert [c.args[0] for c in sleep.call_args_list] == [2]
+
+
+def test_put_retries_exhausted():
+    refused = requests.exceptions.ConnectionError("connection refused")
+    with mock.patch.object(s3_api.requests, "put",
+                           side_effect=refused) as put, \
+            mock.patch.object(s3_api.time, "sleep") as sleep, \
+            pytest.raises(S3UploadError, match="connection refused"):
+        put_part(retries=3)
+    assert put.call_count == 3
+    # no waiting after the last attempt
+    assert [c.args[0] for c in sleep.call_args_list] == [5, 10]
+
+
+def test_put_no_wait_on_success():
+    with mock.patch.object(s3_api.requests, "put",
+                           return_value=make_response(MD5)), \
+            mock.patch.object(s3_api.time, "sleep") as sleep:
+        assert put_part() == MD5
+    sleep.assert_not_called()