diff --git a/CHANGELOG b/CHANGELOG index 55ec669..ce9f046 100644 --- a/CHANGELOG +++ b/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/ diff --git a/dcoraid/api/s3_api.py b/dcoraid/api/s3_api.py index af3d4d5..c460cc4 100644 --- a/dcoraid/api/s3_api.py +++ b/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,6 +315,7 @@ 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 @@ -296,6 +323,12 @@ def requests_put_data_and_get_etag( ---------- 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,6 +355,7 @@ 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 @@ -329,6 +363,7 @@ def requests_put_data_and_get_etag( 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. diff --git a/tests/test_api_s3_retry.py b/tests/test_api_s3_retry.py new file mode 100644 index 0000000..ee1ab40 --- /dev/null +++ b/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()