From 3710546e9ce01ab154e303b8b1ef1cd8a056175d Mon Sep 17 00:00:00 2001 From: tcezard Date: Thu, 8 Oct 2026 13:21:56 +0100 Subject: [PATCH 1/6] Handle retryable and non-retryable errors during file upload in StudySubmitter --- eva_sub_cli/exceptions.py | 6 +++++ eva_sub_cli/submit.py | 32 +++++++++++++++++++++--- tests/test_submit.py | 51 +++++++++++++++++++++++++++++++++++++++ 3 files changed, 86 insertions(+), 3 deletions(-) diff --git a/eva_sub_cli/exceptions.py b/eva_sub_cli/exceptions.py index f95c4b7..c677f22 100644 --- a/eva_sub_cli/exceptions.py +++ b/eva_sub_cli/exceptions.py @@ -40,6 +40,12 @@ def __init__(self, message): super().__init__(self.message) +class RetryableUploadException(Exception): + def __init__(self, message): + self.message = message + super().__init__(self.message) + + class DirLockError(Exception): def __init__(self, message): self.message = message diff --git a/eva_sub_cli/submit.py b/eva_sub_cli/submit.py index 825c13d..2911187 100644 --- a/eva_sub_cli/submit.py +++ b/eva_sub_cli/submit.py @@ -1,6 +1,7 @@ #!/usr/bin/env python import json import os +import time import requests from ebi_eva_common_pyutils.config import WritableConfig @@ -8,7 +9,7 @@ from retry import retry from eva_sub_cli import SUB_CLI_CONFIG_FILE, __version__ -from eva_sub_cli.exceptions import InvalidSubmissionException +from eva_sub_cli.exceptions import InvalidSubmissionException, SubmissionUploadException, RetryableUploadException from eva_sub_cli.submission_ws import SubmissionWSClient from eva_sub_cli.validators.validator import READY_FOR_SUBMISSION_TO_EVA @@ -16,6 +17,10 @@ SUB_CLI_CONFIG_KEY_SUBMISSION_UPLOAD_URL = "submission_upload_url" SUB_CLI_CONFIG_KEY_COMPLETE = "submission_complete" +# HTTP status codes that can be caused by transient conditions and are worth retrying +RETRYABLE_HTTP_STATUS_CODES = {408, 425, 429, 500, 502, 503, 504} +MAX_RETRY_AFTER_SECONDS = 300 + class StudySubmitter(AppLogger): def __init__(self, submission_dir, submission_config: WritableConfig = None, username=None, password=None): @@ -63,15 +68,36 @@ def _upload_submission(self): for f in self.vcf_files: self._upload_file(submission_upload_url, f) - @retry(tries=5, delay=10, backoff=5) + @retry(exceptions=(RetryableUploadException, requests.ConnectionError, requests.Timeout), + tries=5, delay=10, backoff=5) def _upload_file(self, submission_upload_url, input_file): base_name = os.path.basename(input_file) self.debug(f'Transfer {base_name} to EVA FTP') with open(input_file, 'rb') as f: r = requests.put(os.path.join(submission_upload_url, base_name), data=f) - r.raise_for_status() + try: + r.raise_for_status() + except requests.HTTPError as e: + self._handle_upload_error(r, input_file, e) self.debug(f'Upload of {base_name} completed') + def _handle_upload_error(self, response, input_file, error): + base_name = os.path.basename(input_file) + if response.status_code in RETRYABLE_HTTP_STATUS_CODES: + retry_after = response.headers.get('Retry-After') + if retry_after and retry_after.isdigit(): + time.sleep(min(int(retry_after), MAX_RETRY_AFTER_SECONDS)) + self.warning(f'Upload of {base_name} failed with a transient error ({error}). It will be retried.') + raise RetryableUploadException(str(error)) from error + if response.status_code == 413: + file_size_gb = os.path.getsize(input_file) / 1024 ** 3 + raise SubmissionUploadException( + f'The file {base_name} ({file_size_gb:.2f} GB) is too large to be uploaded to EVA. ' + f'Please make sure the file is compressed (.vcf.gz) or split it into several smaller files ' + f'(e.g. per chromosome), update the metadata, then re-validate and resubmit.' + ) from error + raise SubmissionUploadException(f'Upload of {base_name} failed: {error}') from error + def _initiate_submission(self): response_json = self.submission_ws_client.initiate_submission() self.debug(f'Submission ID {response_json["submissionId"]} received!!') diff --git a/tests/test_submit.py b/tests/test_submit.py index 808b9a4..1b21275 100644 --- a/tests/test_submit.py +++ b/tests/test_submit.py @@ -4,10 +4,12 @@ import unittest from unittest.mock import MagicMock, patch, Mock +import requests import yaml from ebi_eva_common_pyutils.config import WritableConfig from eva_sub_cli import SUB_CLI_CONFIG_FILE +from eva_sub_cli.exceptions import SubmissionUploadException, RetryableUploadException from eva_sub_cli.file_utils import is_submission_dir_writable from eva_sub_cli.submission_ws import SubmissionWSClient, DEFAULT_SUBMISSION_WS_URL from eva_sub_cli.validators.validator import READY_FOR_SUBMISSION_TO_EVA @@ -163,3 +165,52 @@ def test_upload_file(self): self.submitter._upload_file(submission_upload_url=test_url, input_file=file_to_upload) assert mock_put.mock_calls[0][1][0] == test_url + os.path.basename(file_to_upload) # Cannot test the content of the upload as opening the same file twice give different object + + def _mock_error_response(self, status_code, headers=None): + response = requests.Response() + response.status_code = status_code + response.headers.update(headers or {}) + response.url = 'http://example.com/EVA_Submission_test.json' + return response + + def test_upload_file_too_large_not_retried(self): + test_url = 'http://example.com/' + file_to_upload = os.path.join(self.resource_dir, 'EVA_Submission_test.json') + with patch('eva_sub_cli.submit.requests.put', return_value=self._mock_error_response(413)) as mock_put, \ + patch('retry.api.time.sleep') as mock_sleep: + with self.assertRaises(SubmissionUploadException) as context: + self.submitter._upload_file(submission_upload_url=test_url, input_file=file_to_upload) + assert 'too large' in context.exception.message + assert mock_put.call_count == 1 + mock_sleep.assert_not_called() + + def test_upload_file_client_error_not_retried(self): + test_url = 'http://example.com/' + file_to_upload = os.path.join(self.resource_dir, 'EVA_Submission_test.json') + with patch('eva_sub_cli.submit.requests.put', return_value=self._mock_error_response(403)) as mock_put, \ + patch('retry.api.time.sleep'): + with self.assertRaises(SubmissionUploadException): + self.submitter._upload_file(submission_upload_url=test_url, input_file=file_to_upload) + assert mock_put.call_count == 1 + + def test_upload_file_transient_error_retried(self): + test_url = 'http://example.com/' + file_to_upload = os.path.join(self.resource_dir, 'EVA_Submission_test.json') + success_response = MagicMock() + responses = [self._mock_error_response(503), self._mock_error_response(429, {'Retry-After': '2'}), + success_response] + with patch('eva_sub_cli.submit.requests.put', side_effect=responses) as mock_put, \ + patch('time.sleep') as mock_sleep: + self.submitter._upload_file(submission_upload_url=test_url, input_file=file_to_upload) + assert mock_put.call_count == 3 + # retry backoff after the 503, then Retry-After followed by the retry backoff after the 429 + assert [c.args[0] for c in mock_sleep.call_args_list] == [10, 2, 50] + + def test_upload_file_transient_error_gives_up(self): + test_url = 'http://example.com/' + file_to_upload = os.path.join(self.resource_dir, 'EVA_Submission_test.json') + with patch('eva_sub_cli.submit.requests.put', return_value=self._mock_error_response(502)) as mock_put, \ + patch('retry.api.time.sleep'): + with self.assertRaises(RetryableUploadException): + self.submitter._upload_file(submission_upload_url=test_url, input_file=file_to_upload) + assert mock_put.call_count == 5 From 55e0bc5b1a584b99521625de150329a0e990932b Mon Sep 17 00:00:00 2001 From: tcezard Date: Fri, 9 Oct 2026 15:22:57 +0100 Subject: [PATCH 2/6] Support uploading large files in parts if rejected by the server --- eva_sub_cli/etc/eva_schema.json | 26 ++++++ eva_sub_cli/exceptions.py | 6 ++ eva_sub_cli/submit.py | 159 ++++++++++++++++++++++++++++---- tests/test_submit.py | 145 ++++++++++++++++++++++++++++- 4 files changed, 313 insertions(+), 23 deletions(-) diff --git a/eva_sub_cli/etc/eva_schema.json b/eva_sub_cli/etc/eva_schema.json index 7720290..1ce112d 100644 --- a/eva_sub_cli/etc/eva_schema.json +++ b/eva_sub_cli/etc/eva_schema.json @@ -380,6 +380,32 @@ "md5": { "type": "string", "description": "MD5 hash of the file" + }, + "fileParts": { + "type": "array", + "description": "Parts the file was split into for upload, in order. Concatenated, they form the file.", + "minItems": 1, + "items": { + "type": "object", + "required": [ + "fileName", + "fileSize" + ], + "properties": { + "fileName": { + "type": "string", + "description": "Name of the part" + }, + "fileSize": { + "type": "number", + "description": "Size of the part in bytes" + }, + "md5": { + "type": "string", + "description": "MD5 hash of the part" + } + } + } } } } diff --git a/eva_sub_cli/exceptions.py b/eva_sub_cli/exceptions.py index c677f22..3e0a911 100644 --- a/eva_sub_cli/exceptions.py +++ b/eva_sub_cli/exceptions.py @@ -46,6 +46,12 @@ def __init__(self, message): super().__init__(self.message) +class FileTooLargeException(Exception): + def __init__(self, message): + self.message = message + super().__init__(self.message) + + class DirLockError(Exception): def __init__(self, message): self.message = message diff --git a/eva_sub_cli/submit.py b/eva_sub_cli/submit.py index 2911187..b9df329 100644 --- a/eva_sub_cli/submit.py +++ b/eva_sub_cli/submit.py @@ -1,5 +1,8 @@ #!/usr/bin/env python +import functools +import hashlib import json +import math import os import time @@ -9,7 +12,8 @@ from retry import retry from eva_sub_cli import SUB_CLI_CONFIG_FILE, __version__ -from eva_sub_cli.exceptions import InvalidSubmissionException, SubmissionUploadException, RetryableUploadException +from eva_sub_cli.exceptions import InvalidSubmissionException, SubmissionUploadException, RetryableUploadException, \ + FileTooLargeException from eva_sub_cli.submission_ws import SubmissionWSClient from eva_sub_cli.validators.validator import READY_FOR_SUBMISSION_TO_EVA @@ -21,11 +25,58 @@ RETRYABLE_HTTP_STATUS_CODES = {408, 425, 429, 500, 502, 503, 504} MAX_RETRY_AFTER_SECONDS = 300 +# Files rejected by the upload server as too large are uploaded in parts of this size, halved while still too large +DEFAULT_PART_SIZE = 512 * 1024 ** 2 +MIN_PART_SIZE = 16 * 1024 ** 2 + +# Directory in the submission directory holding the metadata json uploaded to EVA +UPLOAD_DIR_NAME = '.eva_sub_cli_upload' + + +class FilePart: + """Read-only file-like view of a byte range of a file, which computes the md5 of the data read.""" + + def __init__(self, path, offset, size): + self.size = size + self._remaining = size + self._md5 = hashlib.md5() + self._file = open(path, 'rb') + self._file.seek(offset) + + def read(self, amt=-1): + if self._remaining <= 0: + return b'' + if amt is None or amt < 0 or amt > self._remaining: + amt = self._remaining + data = self._file.read(amt) + self._remaining -= len(data) + self._md5.update(data) + return data + + def __len__(self): + # Used by requests to set the Content-Length header + return self.size + + @property + def md5(self): + return self._md5.hexdigest() + + def close(self): + self._file.close() + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + self.close() + class StudySubmitter(AppLogger): def __init__(self, submission_dir, submission_config: WritableConfig = None, username=None, password=None): self.submission_ws_client = SubmissionWSClient(username, password) self.submission_dir = submission_dir + # Parts of the files uploaded in parts during this run, per file path + self.file_parts = {} if submission_config: self.sub_config = submission_config else: @@ -63,26 +114,102 @@ def _upload_submission(self): submission_upload_url = self.sub_config[SUB_CLI_CONFIG_KEY_SUBMISSION_UPLOAD_URL] - self._upload_file(submission_upload_url, self.metadata_json_file) - for f in self.vcf_files: - self._upload_file(submission_upload_url, f) + try: + self._upload_file(submission_upload_url, f) + except FileTooLargeException: + self.warning(f'{os.path.basename(f)} is too large to be uploaded in one go: uploading it in parts') + self._upload_file_in_parts(submission_upload_url, f, DEFAULT_PART_SIZE) + + # The metadata json is uploaded last so that it describes the files uploaded in parts + self._upload_file(submission_upload_url, self._write_metadata_for_submission()) - @retry(exceptions=(RetryableUploadException, requests.ConnectionError, requests.Timeout), - tries=5, delay=10, backoff=5) def _upload_file(self, submission_upload_url, input_file): base_name = os.path.basename(input_file) self.debug(f'Transfer {base_name} to EVA FTP') - with open(input_file, 'rb') as f: - r = requests.put(os.path.join(submission_upload_url, base_name), data=f) + self._put(os.path.join(submission_upload_url, base_name), lambda: open(input_file, 'rb'), base_name) + self.debug(f'Upload of {base_name} completed') + + @retry(exceptions=(RetryableUploadException, requests.ConnectionError, requests.Timeout), + tries=5, delay=10, backoff=5) + def _put(self, url, open_data, name): + """Upload the data returned by open_data, which is called on each attempt. Returns the data object, closed.""" + with open_data() as data: + r = requests.put(url, data=data) try: r.raise_for_status() except requests.HTTPError as e: - self._handle_upload_error(r, input_file, e) - self.debug(f'Upload of {base_name} completed') + self._handle_upload_error(r, name, e) + return data + + def _upload_file_in_parts(self, submission_upload_url, input_file, part_size): + """Upload a file as consecutive parts, halving the part size while parts are still too large. + The parts are recorded so that they can be described in the metadata.""" + base_name = os.path.basename(input_file) + file_size = os.path.getsize(input_file) + while True: + try: + parts = self._upload_parts(submission_upload_url, input_file, file_size, part_size) + break + except FileTooLargeException: + part_size //= 2 + if part_size < MIN_PART_SIZE: + raise SubmissionUploadException( + f'{base_name} could not be uploaded even when split into small parts. ' + f'Please contact EVA Helpdesk.' + ) + self.warning(f'Parts of {base_name} are still too large: ' + f'uploading it in parts of {part_size / 1024 ** 2:.0f} MB') + if sum(part['fileSize'] for part in parts) != file_size: + raise SubmissionUploadException(f'The parts of {base_name} do not add up to its size of {file_size} bytes') + self.file_parts[input_file] = parts + self.info(f'{base_name} uploaded in {len(parts)} parts') - def _handle_upload_error(self, response, input_file, error): + def _upload_parts(self, submission_upload_url, input_file, file_size, part_size): base_name = os.path.basename(input_file) + number_of_parts = max(1, math.ceil(file_size / part_size)) + parts = [] + for part_index in range(number_of_parts): + offset = part_index * part_size + size = min(part_size, file_size - offset) + # {base_name}.part001 - {base_name}.part999 + part_name = f'{base_name}.part{part_index + 1:0{3}d}' + self.debug(f'Transfer {part_name} to EVA FTP') + # Pass a factory rather than a FilePart: each retry of _put needs a new reader starting at the + # beginning of the part, since a failed attempt consumes the reader and its md5 + data = self._put( + os.path.join(submission_upload_url, part_name), + functools.partial(FilePart, input_file, offset, size), + part_name + ) + parts.append({'fileName': part_name, 'fileSize': size, 'md5': data.md5}) + return parts + + def _metadata_for_submission(self): + """Metadata json with the description of the parts for the files uploaded in parts.""" + metadata = self.metadata_json + parts_per_file_name = { + os.path.basename(vcf_file): parts + for vcf_file, parts in self.file_parts.items() if vcf_file in self.vcf_files + } + for file_dict in metadata.get('files', []): + file_name = os.path.basename(file_dict.get('fileName') or '') + if file_name in parts_per_file_name: + file_dict['fileParts'] = parts_per_file_name[file_name] + else: + file_dict.pop('fileParts', None) + return metadata + + def _write_metadata_for_submission(self): + """Write the metadata to upload, without modifying the metadata json provided or generated during validation.""" + upload_dir = os.path.join(self.submission_dir, UPLOAD_DIR_NAME) + os.makedirs(upload_dir, exist_ok=True) + metadata_file = os.path.join(upload_dir, os.path.basename(self.metadata_json_file)) + with open(metadata_file, 'w') as open_file: + json.dump(self._metadata_for_submission(), open_file) + return metadata_file + + def _handle_upload_error(self, response, base_name, error): if response.status_code in RETRYABLE_HTTP_STATUS_CODES: retry_after = response.headers.get('Retry-After') if retry_after and retry_after.isdigit(): @@ -90,12 +217,7 @@ def _handle_upload_error(self, response, input_file, error): self.warning(f'Upload of {base_name} failed with a transient error ({error}). It will be retried.') raise RetryableUploadException(str(error)) from error if response.status_code == 413: - file_size_gb = os.path.getsize(input_file) / 1024 ** 3 - raise SubmissionUploadException( - f'The file {base_name} ({file_size_gb:.2f} GB) is too large to be uploaded to EVA. ' - f'Please make sure the file is compressed (.vcf.gz) or split it into several smaller files ' - f'(e.g. per chromosome), update the metadata, then re-validate and resubmit.' - ) from error + raise FileTooLargeException(f'{base_name} is too large to be uploaded in one request') from error raise SubmissionUploadException(f'Upload of {base_name} failed: {error}') from error def _initiate_submission(self): @@ -105,7 +227,8 @@ def _initiate_submission(self): self.update_config_with_submission_id_and_upload_url(response_json["submissionId"], response_json["uploadUrl"]) def _complete_submission(self): - self.submission_ws_client.mark_submission_uploaded(self.sub_config.get(SUB_CLI_CONFIG_KEY_SUBMISSION_ID), self.metadata_json) + self.submission_ws_client.mark_submission_uploaded(self.sub_config.get(SUB_CLI_CONFIG_KEY_SUBMISSION_ID), + self._metadata_for_submission()) self.debug("Submission ID {} Complete".format(self.sub_config.get(SUB_CLI_CONFIG_KEY_SUBMISSION_ID))) # update config with completion of the submission self.sub_config.set(SUB_CLI_CONFIG_KEY_COMPLETE, value=True) diff --git a/tests/test_submit.py b/tests/test_submit.py index 1b21275..2a1442b 100644 --- a/tests/test_submit.py +++ b/tests/test_submit.py @@ -1,3 +1,4 @@ +import hashlib import json import os import shutil @@ -9,12 +10,12 @@ from ebi_eva_common_pyutils.config import WritableConfig from eva_sub_cli import SUB_CLI_CONFIG_FILE -from eva_sub_cli.exceptions import SubmissionUploadException, RetryableUploadException +from eva_sub_cli.exceptions import SubmissionUploadException, RetryableUploadException, FileTooLargeException from eva_sub_cli.file_utils import is_submission_dir_writable from eva_sub_cli.submission_ws import SubmissionWSClient, DEFAULT_SUBMISSION_WS_URL from eva_sub_cli.validators.validator import READY_FOR_SUBMISSION_TO_EVA from eva_sub_cli.submit import StudySubmitter, SUB_CLI_CONFIG_KEY_SUBMISSION_ID, \ - SUB_CLI_CONFIG_KEY_SUBMISSION_UPLOAD_URL + SUB_CLI_CONFIG_KEY_SUBMISSION_UPLOAD_URL, FilePart class TestSubmit(unittest.TestCase): @@ -156,7 +157,12 @@ def test_upload_submission(self): for vcf_file in self.submitter.vcf_files: mock_upload_file.assert_any_call(test_url, vcf_file) - mock_upload_file.assert_any_call(test_url, self.submitter.metadata_json_file) + # The metadata json is uploaded last, from a copy in the submission directory + uploaded_metadata = os.path.join(self.test_sub_dir, '.eva_sub_cli_upload', + os.path.basename(self.submitter.metadata_json_file)) + assert mock_upload_file.call_args_list[-1].args == (test_url, uploaded_metadata) + with open(uploaded_metadata) as open_file: + assert json.load(open_file) == self.metadata_json def test_upload_file(self): test_url = 'http://example.com/' @@ -178,9 +184,8 @@ def test_upload_file_too_large_not_retried(self): file_to_upload = os.path.join(self.resource_dir, 'EVA_Submission_test.json') with patch('eva_sub_cli.submit.requests.put', return_value=self._mock_error_response(413)) as mock_put, \ patch('retry.api.time.sleep') as mock_sleep: - with self.assertRaises(SubmissionUploadException) as context: + with self.assertRaises(FileTooLargeException): self.submitter._upload_file(submission_upload_url=test_url, input_file=file_to_upload) - assert 'too large' in context.exception.message assert mock_put.call_count == 1 mock_sleep.assert_not_called() @@ -214,3 +219,133 @@ def test_upload_file_transient_error_gives_up(self): with self.assertRaises(RetryableUploadException): self.submitter._upload_file(submission_upload_url=test_url, input_file=file_to_upload) assert mock_put.call_count == 5 + + + def test_file_part(self): + file_to_upload = os.path.join(self.test_sub_dir, 'file.vcf.gz') + os.makedirs(self.test_sub_dir, exist_ok=True) + with open(file_to_upload, 'wb') as open_file: + open_file.write(b'0123456789') + with FilePart(file_to_upload, 3, 4) as part: + prepared = requests.Request('PUT', 'http://example.com/part', data=part).prepare() + assert prepared.headers['Content-Length'] == '4' + assert part.read(3) + part.read() + part.read() == b'3456' + assert part.md5 == hashlib.md5(b'3456').hexdigest() + + +class FakeUploadServer: + """Stores the data uploaded with requests.put and returns 413 for VCF uploads larger than max_size.""" + + def __init__(self, max_size): + self.max_size = max_size + self.uploaded = {} + self.put_urls = [] + + def put(self, url, data): + content = data.read() + self.put_urls.append(url) + response = requests.Response() + response.url = url + # The limit only applies to the VCF files and their parts in these tests + if '.vcf' in url and len(content) > self.max_size: + response.status_code = 413 + else: + response.status_code = 200 + self.uploaded[url] = content + return response + + def uploaded_names(self): + return [os.path.basename(url) for url in self.uploaded] + + +class TestSubmitInParts(unittest.TestCase): + resource_dir = os.path.join(os.path.dirname(__file__), 'resources') + test_sub_dir = os.path.join(resource_dir, 'test_sub_dir_parts') + upload_url = 'http://example.com/upload/' + + def setUp(self): + shutil.rmtree(self.test_sub_dir, ignore_errors=True) + os.makedirs(self.test_sub_dir) + self.large_vcf = os.path.join(self.test_sub_dir, 'large.vcf.gz') + self.large_content = bytes(range(256)) * 4 # 1024 bytes + with open(self.large_vcf, 'wb') as open_file: + open_file.write(self.large_content) + self.small_vcf = os.path.join(self.test_sub_dir, 'small.vcf.gz') + with open(self.small_vcf, 'wb') as open_file: + open_file.write(b'small') + self.metadata_file = os.path.join(self.test_sub_dir, 'metadata.json') + self.metadata = {'files': [{'analysisAlias': 'A1', 'fileName': 'large.vcf.gz', 'fileSize': 1024}, + {'analysisAlias': 'A1', 'fileName': 'small.vcf.gz', 'fileSize': 5}]} + with open(self.metadata_file, 'w') as open_file: + json.dump(self.metadata, open_file) + + self.submitter = StudySubmitter(submission_dir=self.test_sub_dir) + self.submitter.sub_config.set('metadata_json', value=self.metadata_file) + self.submitter.sub_config.set('vcf_files', value=[self.large_vcf, self.small_vcf]) + self.submitter.sub_config.set(READY_FOR_SUBMISSION_TO_EVA, value=True) + self.submitter.sub_config.set(SUB_CLI_CONFIG_KEY_SUBMISSION_UPLOAD_URL, value=self.upload_url) + + def tearDown(self): + shutil.rmtree(self.test_sub_dir, ignore_errors=True) + + def _upload(self, server, default_part_size=300, min_part_size=50): + with patch('eva_sub_cli.submit.requests.put', side_effect=server.put), \ + patch('eva_sub_cli.submit.DEFAULT_PART_SIZE', default_part_size), \ + patch('eva_sub_cli.submit.MIN_PART_SIZE', min_part_size), \ + patch('retry.api.time.sleep'): + self.submitter._upload_submission() + + def _expected_parts(self, part_size): + parts = [] + for index, offset in enumerate(range(0, len(self.large_content), part_size)): + content = self.large_content[offset:offset + part_size] + parts.append({'fileName': f'large.vcf.gz.part{index + 1:03d}', 'fileSize': len(content), + 'md5': hashlib.md5(content).hexdigest()}) + return parts + + def _uploaded_metadata(self, server): + return json.loads(server.uploaded[self.upload_url + 'metadata.json']) + + def test_upload_in_parts_after_413(self): + server = FakeUploadServer(max_size=500) + self._upload(server) + + expected_parts = self._expected_parts(300) + assert [part['fileSize'] for part in expected_parts] == [300, 300, 300, 124] + assert server.uploaded_names() == [part['fileName'] for part in expected_parts] + \ + ['small.vcf.gz', 'metadata.json'] + assert b''.join(server.uploaded[self.upload_url + part['fileName']] for part in expected_parts) == \ + self.large_content + + assert self.submitter.file_parts == {self.large_vcf: expected_parts} + + # Only the file uploaded in parts is described with its parts + uploaded_metadata = self._uploaded_metadata(server) + assert uploaded_metadata['files'][0]['fileParts'] == expected_parts + assert 'fileParts' not in uploaded_metadata['files'][1] + assert uploaded_metadata == self.submitter._metadata_for_submission() + # The metadata json provided is not modified + with open(self.metadata_file) as open_file: + assert json.load(open_file) == self.metadata + + def test_upload_in_parts_halves_part_size(self): + server = FakeUploadServer(max_size=200) + self._upload(server) + + expected_parts = self._expected_parts(150) + assert self.submitter.file_parts == {self.large_vcf: expected_parts} + assert self._uploaded_metadata(server)['files'][0]['fileParts'] == expected_parts + + def test_upload_in_parts_minimum_part_size(self): + server = FakeUploadServer(max_size=40) + with self.assertRaises(SubmissionUploadException): + self._upload(server) + assert self.submitter.file_parts == {} + + def test_complete_submission_sends_metadata_with_parts(self): + self._upload(FakeUploadServer(max_size=500)) + self.submitter.sub_config.set(SUB_CLI_CONFIG_KEY_SUBMISSION_ID, value='submission_id') + with patch.object(self.submitter.submission_ws_client, 'mark_submission_uploaded') as mock_uploaded: + self.submitter._complete_submission() + mock_uploaded.assert_called_once_with('submission_id', self.submitter._metadata_for_submission()) + assert mock_uploaded.call_args.args[1]['files'][0]['fileParts'] == self._expected_parts(300) From cf3b3d67fb7600ee0d898ff08db9a0b357ba7ab8 Mon Sep 17 00:00:00 2001 From: tcezard Date: Tue, 13 Oct 2026 15:17:08 +0100 Subject: [PATCH 3/6] Simplify upload logic and retry strategy --- eva_sub_cli/submit.py | 46 ++++++++++++++++++------------------------- tests/test_submit.py | 21 +++++++------------- 2 files changed, 26 insertions(+), 41 deletions(-) diff --git a/eva_sub_cli/submit.py b/eva_sub_cli/submit.py index b9df329..affef6a 100644 --- a/eva_sub_cli/submit.py +++ b/eva_sub_cli/submit.py @@ -21,16 +21,15 @@ SUB_CLI_CONFIG_KEY_SUBMISSION_UPLOAD_URL = "submission_upload_url" SUB_CLI_CONFIG_KEY_COMPLETE = "submission_complete" -# HTTP status codes that can be caused by transient conditions and are worth retrying -RETRYABLE_HTTP_STATUS_CODES = {408, 425, 429, 500, 502, 503, 504} +# HTTP status codes that can be caused by transient conditions and are worth retrying. +# 404 is included because the Globus HTTPS server can return it when its storage backend fails during an upload, +RETRYABLE_HTTP_STATUS_CODES = {408, 404, 425, 429, 500, 502, 503, 504} MAX_RETRY_AFTER_SECONDS = 300 -# Files rejected by the upload server as too large are uploaded in parts of this size, halved while still too large -DEFAULT_PART_SIZE = 512 * 1024 ** 2 -MIN_PART_SIZE = 16 * 1024 ** 2 +# Files rejected by the upload server as too large are uploaded in parts of this size, +# below the 1 GB limit of the upload server +PART_SIZE = 512 * 1024 ** 2 -# Directory in the submission directory holding the metadata json uploaded to EVA -UPLOAD_DIR_NAME = '.eva_sub_cli_upload' class FilePart: @@ -119,7 +118,7 @@ def _upload_submission(self): self._upload_file(submission_upload_url, f) except FileTooLargeException: self.warning(f'{os.path.basename(f)} is too large to be uploaded in one go: uploading it in parts') - self._upload_file_in_parts(submission_upload_url, f, DEFAULT_PART_SIZE) + self._upload_file_in_parts(submission_upload_url, f) # The metadata json is uploaded last so that it describes the files uploaded in parts self._upload_file(submission_upload_url, self._write_metadata_for_submission()) @@ -131,7 +130,7 @@ def _upload_file(self, submission_upload_url, input_file): self.debug(f'Upload of {base_name} completed') @retry(exceptions=(RetryableUploadException, requests.ConnectionError, requests.Timeout), - tries=5, delay=10, backoff=5) + tries=5, delay=10, backoff=5, max_delay=300) def _put(self, url, open_data, name): """Upload the data returned by open_data, which is called on each attempt. Returns the data object, closed.""" with open_data() as data: @@ -142,24 +141,17 @@ def _put(self, url, open_data, name): self._handle_upload_error(r, name, e) return data - def _upload_file_in_parts(self, submission_upload_url, input_file, part_size): - """Upload a file as consecutive parts, halving the part size while parts are still too large. + def _upload_file_in_parts(self, submission_upload_url, input_file): + """Upload a file as consecutive parts of PART_SIZE. The parts are recorded so that they can be described in the metadata.""" base_name = os.path.basename(input_file) file_size = os.path.getsize(input_file) - while True: - try: - parts = self._upload_parts(submission_upload_url, input_file, file_size, part_size) - break - except FileTooLargeException: - part_size //= 2 - if part_size < MIN_PART_SIZE: - raise SubmissionUploadException( - f'{base_name} could not be uploaded even when split into small parts. ' - f'Please contact EVA Helpdesk.' - ) - self.warning(f'Parts of {base_name} are still too large: ' - f'uploading it in parts of {part_size / 1024 ** 2:.0f} MB') + try: + parts = self._upload_parts(submission_upload_url, input_file, file_size, PART_SIZE) + except FileTooLargeException: + raise SubmissionUploadException( + f'{base_name} could not be uploaded even when split into parts. Please contact EVA Helpdesk.' + ) if sum(part['fileSize'] for part in parts) != file_size: raise SubmissionUploadException(f'The parts of {base_name} do not add up to its size of {file_size} bytes') self.file_parts[input_file] = parts @@ -202,7 +194,7 @@ def _metadata_for_submission(self): def _write_metadata_for_submission(self): """Write the metadata to upload, without modifying the metadata json provided or generated during validation.""" - upload_dir = os.path.join(self.submission_dir, UPLOAD_DIR_NAME) + upload_dir = os.path.join(self.submission_dir, '.metadata_upload_in_parts') os.makedirs(upload_dir, exist_ok=True) metadata_file = os.path.join(upload_dir, os.path.basename(self.metadata_json_file)) with open(metadata_file, 'w') as open_file: @@ -211,10 +203,10 @@ def _write_metadata_for_submission(self): def _handle_upload_error(self, response, base_name, error): if response.status_code in RETRYABLE_HTTP_STATUS_CODES: - retry_after = response.headers.get('Retry-After') + retry_after = response.headers.get('Retry-After') # Wait for whatever the server says on top of the retry delay if retry_after and retry_after.isdigit(): time.sleep(min(int(retry_after), MAX_RETRY_AFTER_SECONDS)) - self.warning(f'Upload of {base_name} failed with a transient error ({error}). It will be retried.') + self.warning(f'Upload of {base_name} failed with error ({error}). It will be retried.') raise RetryableUploadException(str(error)) from error if response.status_code == 413: raise FileTooLargeException(f'{base_name} is too large to be uploaded in one request') from error diff --git a/tests/test_submit.py b/tests/test_submit.py index 2a1442b..7e855d3 100644 --- a/tests/test_submit.py +++ b/tests/test_submit.py @@ -158,7 +158,7 @@ def test_upload_submission(self): mock_upload_file.assert_any_call(test_url, vcf_file) # The metadata json is uploaded last, from a copy in the submission directory - uploaded_metadata = os.path.join(self.test_sub_dir, '.eva_sub_cli_upload', + uploaded_metadata = os.path.join(self.test_sub_dir, '.metadata_upload_in_parts', os.path.basename(self.submitter.metadata_json_file)) assert mock_upload_file.call_args_list[-1].args == (test_url, uploaded_metadata) with open(uploaded_metadata) as open_file: @@ -215,10 +215,12 @@ def test_upload_file_transient_error_gives_up(self): test_url = 'http://example.com/' file_to_upload = os.path.join(self.resource_dir, 'EVA_Submission_test.json') with patch('eva_sub_cli.submit.requests.put', return_value=self._mock_error_response(502)) as mock_put, \ - patch('retry.api.time.sleep'): + patch('retry.api.time.sleep') as mock_sleep: with self.assertRaises(RetryableUploadException): self.submitter._upload_file(submission_upload_url=test_url, input_file=file_to_upload) assert mock_put.call_count == 5 + # The retry backoff is capped + assert [c.args[0] for c in mock_sleep.call_args_list] == [10, 50, 250, 300] def test_file_part(self): @@ -288,10 +290,9 @@ def setUp(self): def tearDown(self): shutil.rmtree(self.test_sub_dir, ignore_errors=True) - def _upload(self, server, default_part_size=300, min_part_size=50): + def _upload(self, server): with patch('eva_sub_cli.submit.requests.put', side_effect=server.put), \ - patch('eva_sub_cli.submit.DEFAULT_PART_SIZE', default_part_size), \ - patch('eva_sub_cli.submit.MIN_PART_SIZE', min_part_size), \ + patch('eva_sub_cli.submit.PART_SIZE', 300), \ patch('retry.api.time.sleep'): self.submitter._upload_submission() @@ -328,16 +329,8 @@ def test_upload_in_parts_after_413(self): with open(self.metadata_file) as open_file: assert json.load(open_file) == self.metadata - def test_upload_in_parts_halves_part_size(self): + def test_upload_in_parts_still_too_large(self): server = FakeUploadServer(max_size=200) - self._upload(server) - - expected_parts = self._expected_parts(150) - assert self.submitter.file_parts == {self.large_vcf: expected_parts} - assert self._uploaded_metadata(server)['files'][0]['fileParts'] == expected_parts - - def test_upload_in_parts_minimum_part_size(self): - server = FakeUploadServer(max_size=40) with self.assertRaises(SubmissionUploadException): self._upload(server) assert self.submitter.file_parts == {} From 0c1a2acf6b908df2270464af2beef1b622da2d10 Mon Sep 17 00:00:00 2001 From: tcezard Date: Tue, 13 Oct 2026 15:44:20 +0100 Subject: [PATCH 4/6] Refactor test helpers and patching in `test_submit.py` --- tests/test_submit.py | 22 +++++++++++----------- 1 file changed, 11 insertions(+), 11 deletions(-) diff --git a/tests/test_submit.py b/tests/test_submit.py index 7e855d3..46c3ff7 100644 --- a/tests/test_submit.py +++ b/tests/test_submit.py @@ -172,7 +172,7 @@ def test_upload_file(self): assert mock_put.mock_calls[0][1][0] == test_url + os.path.basename(file_to_upload) # Cannot test the content of the upload as opening the same file twice give different object - def _mock_error_response(self, status_code, headers=None): + def _mock_put_error_response(self, status_code, headers=None): response = requests.Response() response.status_code = status_code response.headers.update(headers or {}) @@ -182,8 +182,8 @@ def _mock_error_response(self, status_code, headers=None): def test_upload_file_too_large_not_retried(self): test_url = 'http://example.com/' file_to_upload = os.path.join(self.resource_dir, 'EVA_Submission_test.json') - with patch('eva_sub_cli.submit.requests.put', return_value=self._mock_error_response(413)) as mock_put, \ - patch('retry.api.time.sleep') as mock_sleep: + with patch('eva_sub_cli.submit.requests.put', return_value=self._mock_put_error_response(413)) as mock_put, \ + patch('time.sleep') as mock_sleep: with self.assertRaises(FileTooLargeException): self.submitter._upload_file(submission_upload_url=test_url, input_file=file_to_upload) assert mock_put.call_count == 1 @@ -192,8 +192,8 @@ def test_upload_file_too_large_not_retried(self): def test_upload_file_client_error_not_retried(self): test_url = 'http://example.com/' file_to_upload = os.path.join(self.resource_dir, 'EVA_Submission_test.json') - with patch('eva_sub_cli.submit.requests.put', return_value=self._mock_error_response(403)) as mock_put, \ - patch('retry.api.time.sleep'): + with patch('eva_sub_cli.submit.requests.put', return_value=self._mock_put_error_response(403)) as mock_put, \ + patch('time.sleep'): with self.assertRaises(SubmissionUploadException): self.submitter._upload_file(submission_upload_url=test_url, input_file=file_to_upload) assert mock_put.call_count == 1 @@ -202,27 +202,27 @@ def test_upload_file_transient_error_retried(self): test_url = 'http://example.com/' file_to_upload = os.path.join(self.resource_dir, 'EVA_Submission_test.json') success_response = MagicMock() - responses = [self._mock_error_response(503), self._mock_error_response(429, {'Retry-After': '2'}), + responses = [self._mock_put_error_response(503), self._mock_put_error_response(429, {'Retry-After': '2'}), success_response] with patch('eva_sub_cli.submit.requests.put', side_effect=responses) as mock_put, \ patch('time.sleep') as mock_sleep: self.submitter._upload_file(submission_upload_url=test_url, input_file=file_to_upload) assert mock_put.call_count == 3 + # 2 retries but 3 sleeps: # retry backoff after the 503, then Retry-After followed by the retry backoff after the 429 assert [c.args[0] for c in mock_sleep.call_args_list] == [10, 2, 50] def test_upload_file_transient_error_gives_up(self): test_url = 'http://example.com/' file_to_upload = os.path.join(self.resource_dir, 'EVA_Submission_test.json') - with patch('eva_sub_cli.submit.requests.put', return_value=self._mock_error_response(502)) as mock_put, \ - patch('retry.api.time.sleep') as mock_sleep: + with patch('eva_sub_cli.submit.requests.put', return_value=self._mock_put_error_response(502)) as mock_put, \ + patch('time.sleep') as mock_sleep: with self.assertRaises(RetryableUploadException): self.submitter._upload_file(submission_upload_url=test_url, input_file=file_to_upload) assert mock_put.call_count == 5 - # The retry backoff is capped + # The retry backoff is capped at 300 assert [c.args[0] for c in mock_sleep.call_args_list] == [10, 50, 250, 300] - def test_file_part(self): file_to_upload = os.path.join(self.test_sub_dir, 'file.vcf.gz') os.makedirs(self.test_sub_dir, exist_ok=True) @@ -293,7 +293,7 @@ def tearDown(self): def _upload(self, server): with patch('eva_sub_cli.submit.requests.put', side_effect=server.put), \ patch('eva_sub_cli.submit.PART_SIZE', 300), \ - patch('retry.api.time.sleep'): + patch('time.sleep'): self.submitter._upload_submission() def _expected_parts(self, part_size): From 79e0626d31c0bd435b68bc3a04ef28098e01dacf Mon Sep 17 00:00:00 2001 From: tcezard Date: Tue, 13 Oct 2026 15:50:45 +0100 Subject: [PATCH 5/6] Add docstring to `TestSubmitInParts`. --- tests/test_submit.py | 11 +++++------ 1 file changed, 5 insertions(+), 6 deletions(-) diff --git a/tests/test_submit.py b/tests/test_submit.py index 46c3ff7..c83c7e0 100644 --- a/tests/test_submit.py +++ b/tests/test_submit.py @@ -261,6 +261,11 @@ def uploaded_names(self): class TestSubmitInParts(unittest.TestCase): + """ + Upload of a submission with a large VCF rejected by the server (413) and a small VCF accepted as a whole, + against a FakeUploadServer and with a part size of 300 bytes. Checks that the large VCF is uploaded in parts, + that the parts are described in the metadata uploaded and sent to complete the submission. + """ resource_dir = os.path.join(os.path.dirname(__file__), 'resources') test_sub_dir = os.path.join(resource_dir, 'test_sub_dir_parts') upload_url = 'http://example.com/upload/' @@ -329,12 +334,6 @@ def test_upload_in_parts_after_413(self): with open(self.metadata_file) as open_file: assert json.load(open_file) == self.metadata - def test_upload_in_parts_still_too_large(self): - server = FakeUploadServer(max_size=200) - with self.assertRaises(SubmissionUploadException): - self._upload(server) - assert self.submitter.file_parts == {} - def test_complete_submission_sends_metadata_with_parts(self): self._upload(FakeUploadServer(max_size=500)) self.submitter.sub_config.set(SUB_CLI_CONFIG_KEY_SUBMISSION_ID, value='submission_id') From ec036e14537d1937e7183b99751522aa0e722410 Mon Sep 17 00:00:00 2001 From: tcezard Date: Thu, 15 Oct 2026 17:34:16 +0100 Subject: [PATCH 6/6] Make upload part size configurable via the `UPLOAD_PART_SIZE` environment variable. --- eva_sub_cli/__init__.py | 1 + eva_sub_cli/submit.py | 20 +++++++++++++++----- tests/test_submit.py | 4 ++-- 3 files changed, 18 insertions(+), 7 deletions(-) diff --git a/eva_sub_cli/__init__.py b/eva_sub_cli/__init__.py index e85000d..a6e4ea4 100644 --- a/eva_sub_cli/__init__.py +++ b/eva_sub_cli/__init__.py @@ -14,6 +14,7 @@ SUBMISSION_WS_URL_VAR = 'SUBMISSION_WS_URL' ENA_WEBIN_ACCOUNT_VAR = 'ENA_WEBIN_ACCOUNT' ENA_WEBIN_PASSWORD_VAR = 'ENA_WEBIN_PASSWORD' +UPLOAD_PART_SIZE_VAR = 'UPLOAD_PART_SIZE' try: diff --git a/eva_sub_cli/submit.py b/eva_sub_cli/submit.py index affef6a..c92d9ad 100644 --- a/eva_sub_cli/submit.py +++ b/eva_sub_cli/submit.py @@ -11,7 +11,7 @@ from ebi_eva_common_pyutils.logger import AppLogger from retry import retry -from eva_sub_cli import SUB_CLI_CONFIG_FILE, __version__ +from eva_sub_cli import SUB_CLI_CONFIG_FILE, UPLOAD_PART_SIZE_VAR, __version__ from eva_sub_cli.exceptions import InvalidSubmissionException, SubmissionUploadException, RetryableUploadException, \ FileTooLargeException from eva_sub_cli.submission_ws import SubmissionWSClient @@ -27,8 +27,8 @@ MAX_RETRY_AFTER_SECONDS = 300 # Files rejected by the upload server as too large are uploaded in parts of this size, -# below the 1 GB limit of the upload server -PART_SIZE = 512 * 1024 ** 2 +# below the 1 GB limit of the upload server. It can be changed with the UPLOAD_PART_SIZE environment variable (bytes) +DEFAULT_PART_SIZE = 512 * 1024 ** 2 @@ -70,6 +70,16 @@ def __exit__(self, exc_type, exc_val, exc_tb): self.close() +def get_part_size(): + """Size in bytes of the parts when a file is uploaded in parts.""" + part_size = os.environ.get(UPLOAD_PART_SIZE_VAR) + if not part_size: + return DEFAULT_PART_SIZE + if not part_size.isdigit() or int(part_size) == 0: + raise SubmissionUploadException(f'{UPLOAD_PART_SIZE_VAR} must be a positive number of bytes, not {part_size}') + return int(part_size) + + class StudySubmitter(AppLogger): def __init__(self, submission_dir, submission_config: WritableConfig = None, username=None, password=None): self.submission_ws_client = SubmissionWSClient(username, password) @@ -142,12 +152,12 @@ def _put(self, url, open_data, name): return data def _upload_file_in_parts(self, submission_upload_url, input_file): - """Upload a file as consecutive parts of PART_SIZE. + """Upload a file as consecutive parts of the size returned by get_part_size. The parts are recorded so that they can be described in the metadata.""" base_name = os.path.basename(input_file) file_size = os.path.getsize(input_file) try: - parts = self._upload_parts(submission_upload_url, input_file, file_size, PART_SIZE) + parts = self._upload_parts(submission_upload_url, input_file, file_size, get_part_size()) except FileTooLargeException: raise SubmissionUploadException( f'{base_name} could not be uploaded even when split into parts. Please contact EVA Helpdesk.' diff --git a/tests/test_submit.py b/tests/test_submit.py index c83c7e0..ebf4442 100644 --- a/tests/test_submit.py +++ b/tests/test_submit.py @@ -9,7 +9,7 @@ import yaml from ebi_eva_common_pyutils.config import WritableConfig -from eva_sub_cli import SUB_CLI_CONFIG_FILE +from eva_sub_cli import SUB_CLI_CONFIG_FILE, UPLOAD_PART_SIZE_VAR from eva_sub_cli.exceptions import SubmissionUploadException, RetryableUploadException, FileTooLargeException from eva_sub_cli.file_utils import is_submission_dir_writable from eva_sub_cli.submission_ws import SubmissionWSClient, DEFAULT_SUBMISSION_WS_URL @@ -297,7 +297,7 @@ def tearDown(self): def _upload(self, server): with patch('eva_sub_cli.submit.requests.put', side_effect=server.put), \ - patch('eva_sub_cli.submit.PART_SIZE', 300), \ + patch.dict(os.environ, {UPLOAD_PART_SIZE_VAR: '300'}), \ patch('time.sleep'): self.submitter._upload_submission()