[ENG-11588] fix(osf): execute archive chain via self.replace so registr…

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):

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):