From 7dec9288c8322d6e5b72facc6fe489d1871b1dfc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Paul=20M=C3=BCller?= Date: Fri, 23 Oct 2026 12:25:46 +0200 Subject: [PATCH] enh: wait between retries of failed S3 upload requests Failed part uploads and multipart completion requests were retried immediately, so short connection problems (e.g. "connection refused" by the object store, as seen on the Windows CI runners) led to failed uploads. Now, DCOR-Aid waits 5 s before the first retry, and the waiting time doubles with each retry. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG | 2 + dcoraid/api/s3_api.py | 40 +++++++++++++++++- tests/test_api_s3_retry.py | 83 ++++++++++++++++++++++++++++++++++++++ 3 files changed, 124 insertions(+), 1 deletion(-) create mode 100644 tests/test_api_s3_retry.py 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()