[ENG-11860] make archive copy idempotent under retry

osf/management/commands/force_archive.py

@@ -45,7 +45,7 @@
 from api.waffle.utils import flag_is_active
 from scripts import utils as script_utils
 from website.archiver import ARCHIVER_SUCCESS
-from website.settings import ARCHIVE_TIMEOUT_TIMEDELTA, ARCHIVE_PROVIDER, COOKIE_NAME, EXTERNAL_REQUEST_TIMEOUT
+from website.settings import ARCHIVE_TIMEOUT_TIMEDELTA, ARCHIVE_PROVIDER, COOKIE_NAME, ARCHIVE_COPY_REQUEST_TIMEOUT
 from website.files.utils import attach_versions
 
 logger = logging.getLogger(__name__)
@@ -174,7 +174,10 @@ def perform_wb_copy(reg, node_settings, delete_collisions=False, skip_collisions
         'provider': ARCHIVE_PROVIDER,
     }
     url = waterbutler_api_url_for(src._id, node_settings.short_name, _internal=True, base_url=src.osfstorage_region.waterbutler_url, **params)
-    res = requests.post(url, data=json.dumps(data), cookies={COOKIE_NAME: cookie}, timeout=EXTERNAL_REQUEST_TIMEOUT)
+    # WaterButler keeps the connection open until it has copied the whole tree. Big archives take
+    # longer than the general 30s timeout, and those are the ones that need a manual restart, so
+    # use the archiver's longer timeout here.
+    res = requests.post(url, data=json.dumps(data), cookies={COOKIE_NAME: cookie}, timeout=ARCHIVE_COPY_REQUEST_TIMEOUT)
     if res.status_code not in (http_status.HTTP_200_OK, http_status.HTTP_201_CREATED, http_status.HTTP_202_ACCEPTED):
         http_exception = HTTPError(res.status_code)
         sentry.log_exception(http_exception)

osf_tests/test_archiver.py

@@ -612,6 +612,9 @@ def test_archive_addon(self):
             'rename': 'Archive of OSF Storage',
             'resource': self.archive_job.info()[1]._id,
             'provider': 'osfstorage',
+            # 'replace' keeps the copy idempotent so a retried copy request doesn't fail
+            # the archive with WaterButler's "already exists" naming conflict.
+            'conflict': 'replace',
         }
 
     @mock.patch('website.archiver.tasks.archive_callback.delay')
@@ -620,6 +623,30 @@ def test_archive_addon_does_not_trigger_callback_immediately(self, mock_archive_
 
         mock_archive_callback.assert_not_called()
 
+    @mock.patch('website.archiver.tasks.requests.post')
+    def test_copy_request_is_idempotent_and_uses_archive_timeout(self, mock_post):
+        # The copy must be retry-safe: WaterButler defaults to conflict='warn' and raises
+        # "already exists" on a retried copy, which fails the whole archive.
+        payload = make_waterbutler_payload(self.dst._id, 'Archive of OSF Storage')
+        assert payload['conflict'] == 'replace'
+        # WaterButler's copy is synchronous; large trees need more than the general 30s timeout.
+        assert settings.ARCHIVE_COPY_REQUEST_TIMEOUT[1] > settings.EXTERNAL_REQUEST_TIMEOUT[1]
+        mock_post.return_value = mock.Mock(status_code=200)
+        params = archive_addon('osfstorage', self.archive_job._id)
+        make_copy_request(params, self.archive_job._id)
+        assert mock_post.call_args.kwargs['timeout'] == settings.ARCHIVE_COPY_REQUEST_TIMEOUT
+
+    @mock.patch('website.archiver.tasks.requests.post')
+    def test_copy_request_skipped_once_waterbutler_reported_success(self, mock_post):
+        # A SUCCESS target means the callback already reported the copy as done, even if we
+        # never saw the response. Running the task again must not copy a second time: that
+        # would replace the files the registration already uses.
+        params = archive_addon('osfstorage', self.archive_job._id)
+        self.archive_job.update_target('osfstorage', ARCHIVER_SUCCESS)
+        make_copy_request(params, self.archive_job._id)
+        mock_post.assert_not_called()
+        assert self.archive_job.get_target('osfstorage').status == ARCHIVER_SUCCESS
+
     @mock.patch.object(archive_node, 'replace')
     @mock.patch('website.archiver.tasks.archive_callback.si')
     @mock.patch('website.archiver.tasks.make_copy_request.s')

website/archiver/tasks.py

@@ -242,6 +242,11 @@ def stat_addon(self, addon_short_name, job_pk):
     return result
 
 
+def archive_target_succeeded(job, addon_short_name):
+    target = job.get_target(addon_short_name)
+    return target is not None and target.status == ARCHIVER_SUCCESS
+
+
 @celery_app.task(
     bind=True,
     base=ArchiverTask,
@@ -264,27 +269,36 @@ def make_copy_request(self, params, job_pk):
     addon_short_name = params['addon_short_name']
     url = params['url']
     data = params['data']
+
+    if archive_target_succeeded(job, addon_short_name):
+        logger.info(f'Skipping copy request for addon: {addon_short_name} on node: {dst._id}, already archived')
+        return
+
     logger.info(f"Sending copy request for addon: {data['provider']} on node: {dst._id}")
     cookie = furl(url).query.params.get('cookie')
     try:
         res = requests.post(
             url,
             data=json.dumps(data),
             cookies={settings.COOKIE_NAME: cookie},
-            timeout=settings.EXTERNAL_REQUEST_TIMEOUT,
+            timeout=settings.ARCHIVE_COPY_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)
+        if archive_target_succeeded(job, addon_short_name):
+            return
         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))
+        if archive_target_succeeded(job, addon_short_name):
+            return
         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)
 
@@ -295,6 +309,11 @@ def make_waterbutler_payload(dst_id, rename):
         'rename': rename.replace('/', '-'),
         'resource': dst_id,
         'provider': settings.ARCHIVE_PROVIDER,
+        # Archive into a freshly-created registration, so overwriting is always safe. Without this
+        # WaterButler defaults to conflict='warn' and raises "already exists" if the copy is retried
+        # (e.g. after a client-side timeout on a copy WaterButler actually completed), which fails
+        # the whole archive. 'replace' makes the copy idempotent under retry.
+        'conflict': 'replace',
     }
 
 @celery_app.task(

website/settings/defaults.py

@@ -386,6 +386,10 @@ def parent_dir(path):
 SHARE_API_TOKEN = None  # Required to send project updates to SHARE
 
 EXTERNAL_REQUEST_TIMEOUT = (10, 30)  # (connect, read) timeout for outbound requests to external services
+# The archive copy request is synchronous on WaterButler's side: it holds the connection open until
+# the whole osfstorage tree has been copied. Large registrations exceed the 30s general read timeout,
+# so give this specific request a longer read timeout while keeping the connect timeout short.
+ARCHIVE_COPY_REQUEST_TIMEOUT = (10, 600)
 
 SHARE_UPDATE_TASK_SOFT_TIME_LIMIT = 90
 SHARE_UPDATE_TASK_HARD_TIME_LIMIT = 120