diff --git a/osf_tests/test_archiver.py b/osf_tests/test_archiver.py index 4dc9b1a23e0..0bf357c3215 100644 --- a/osf_tests/test_archiver.py +++ b/osf_tests/test_archiver.py @@ -542,8 +542,9 @@ def test_archiver_task_load_archive_job_final_failure_logs_context(self, mock_lo ) mock_log_exception.assert_called_once() + @mock.patch.object(archive_node, 'replace') @mock.patch('website.archiver.tasks.archive_addon.si') - def test_archive_node_pass(self, mock_archive_addon): + def test_archive_node_pass(self, mock_archive_addon, mock_replace): settings.MAX_ARCHIVE_SIZE = 1024 ** 3 with mock.patch.object(BaseStorageAddon, '_get_file_tree') as mock_file_tree: mock_file_tree.return_value = FILE_TREE @@ -561,9 +562,10 @@ def test_archive_node_fail(self): with pytest.raises(ArchiverSizeExceeded): # Note: Requires task_eager_propagates = True in celery archive_node.apply(args=(results, self.archive_job._id)) + @mock.patch.object(archive_node, 'replace') @mock.patch('website.archiver.tasks.archive_callback.si') @mock.patch('website.archiver.tasks.archive_addon.si') - def test_archive_node_does_not_archive_empty_addons(self, mock_archive_addon, mock_send): + def test_archive_node_does_not_archive_empty_addons(self, mock_archive_addon, mock_send, mock_replace): with mock.patch('osf.models.mixins.AddonModelMixin.get_addon') as mock_get_addon: mock_addon = MockAddon() @@ -582,8 +584,9 @@ def empty_file_tree(user, version): assert mock_send.called @use_fake_addons + @mock.patch.object(archive_node, 'replace') @mock.patch('website.archiver.tasks.archive_addon.si') - def test_archive_node_no_archive_size_limit(self, mock_archive_addon): + def test_archive_node_no_archive_size_limit(self, mock_archive_addon, mock_replace): settings.MAX_ARCHIVE_SIZE = 100 self.archive_job.initiator.add_system_tag(NO_ARCHIVE_LIMIT) self.archive_job.initiator.save() @@ -617,6 +620,7 @@ def test_archive_addon_does_not_trigger_callback_immediately(self, mock_archive_ mock_archive_callback.assert_not_called() + @mock.patch.object(archive_node, 'replace') @mock.patch('website.archiver.tasks.archive_callback.si') @mock.patch('website.archiver.tasks.make_copy_request.s') @mock.patch('website.archiver.tasks.celery.chain') @@ -629,6 +633,7 @@ def test_archive_node_only_enqueues_addon_work_before_callback( mock_chain, mock_make_copy_request_s, mock_archive_callback, + mock_replace, ): settings.MAX_ARCHIVE_SIZE = 1024 ** 3 with mock.patch.object(BaseStorageAddon, '_get_file_tree') as mock_file_tree: @@ -651,6 +656,9 @@ def test_archive_node_only_enqueues_addon_work_before_callback( mock_group.return_value, mock_archive_callback.return_value, ]) + # The built chain must be handed to self.replace() so Celery actually + # executes it; a bare `return chain` would silently never run. + mock_replace.assert_called_once_with(mock_chain.return_value) @pytest.mark.usefixtures('mock_gravy_valet_get_verified_links') def test_archive_success(self): diff --git a/website/archiver/tasks.py b/website/archiver/tasks.py index d390b67dc7e..6b0a304eca3 100644 --- a/website/archiver/tasks.py +++ b/website/archiver/tasks.py @@ -274,10 +274,17 @@ def make_copy_request(self, params, job_pk): timeout=settings.EXTERNAL_REQUEST_TIMEOUT, ) except requests.RequestException as exc: + # A failed copy request marks the target as failed, which fails the whole + # archive and deletes the registration. Retry transient network errors first. + if self.request.retries < self.max_retries: + raise self.retry(exc=exc) job.update_target(addon_short_name, ARCHIVER_FAILURE, errors=[str(exc)]) raise if res.status_code not in (http_status.HTTP_200_OK, http_status.HTTP_201_CREATED, http_status.HTTP_202_ACCEPTED): + # Retry server-side WaterButler errors before failing (and deleting) the archive. + if res.status_code >= 500 and self.request.retries < self.max_retries: + raise self.retry(exc=HTTPError(res.status_code)) job.update_target(addon_short_name, ARCHIVER_FAILURE, errors=[res.text or f'WaterButler request failed with status {res.status_code}']) raise HTTPError(res.status_code) @@ -394,17 +401,21 @@ def archive_node(self, stat_results, job_pk): ) ) + # NOTE: use self.replace() rather than `return celery.chain(...)`. A Celery + # task that merely *returns* a signature does NOT execute it, so the copy + # requests and archive_callback would never run -- the archive would stall + # and the registration would be torn down (deleted) after submission. if not addon_tasks: - return celery.chain([ + return self.replace(celery.chain([ archive_callback.si(dst_id=dst._id) - ]) + ])) - return celery.chain( + return self.replace(celery.chain( [ celery.group(addon_tasks), archive_callback.si(dst_id=dst._id), ] - ) + )) def archive(job_pk):