diff --git a/swh/journal/replay.py b/swh/journal/replay.py
index a654d02..873506e 100644
--- a/swh/journal/replay.py
+++ b/swh/journal/replay.py
@@ -1,449 +1,519 @@
 # Copyright (C) 2019-2020 The Software Heritage developers
 # See the AUTHORS file at the top-level directory of this distribution
 # License: GNU General Public License version 3, or any later version
 # See top-level LICENSE file for more information
 
 import copy
 import logging
 from time import time
 from typing import Callable, Dict, List, Optional
 
+from sentry_sdk import capture_exception, push_scope
 try:
     from systemd.daemon import notify
 except ImportError:
     notify = None
 
-from tenacity import retry, stop_after_attempt, wait_random_exponential
+from tenacity import (
+    retry, retry_if_exception_type, stop_after_attempt,
+    wait_random_exponential,
+)
 
 from swh.core.statsd import statsd
 from swh.model.identifiers import normalize_timestamp
 from swh.model.hashutil import hash_to_hex
 from swh.model.model import SHA1_SIZE
 from swh.objstorage.objstorage import (
     ID_HASH_ALGO, ObjNotFoundError, ObjStorage,
 )
 from swh.storage import HashCollision
 
 logger = logging.getLogger(__name__)
 
 GRAPH_OPERATIONS_METRIC = "swh_graph_replayer_operations_total"
 GRAPH_DURATION_METRIC = "swh_graph_replayer_duration_seconds"
 CONTENT_OPERATIONS_METRIC = "swh_content_replayer_operations_total"
+CONTENT_RETRY_METRIC = "swh_content_replayer_retries_total"
 CONTENT_BYTES_METRIC = "swh_content_replayer_bytes"
 CONTENT_DURATION_METRIC = "swh_content_replayer_duration_seconds"
 
 
 def process_replay_objects(all_objects, *, storage):
     for (object_type, objects) in all_objects.items():
         logger.debug("Inserting %s %s objects", len(objects), object_type)
         with statsd.timed(GRAPH_DURATION_METRIC,
                           tags={'object_type': object_type}):
             _insert_objects(object_type, objects, storage)
         statsd.increment(GRAPH_OPERATIONS_METRIC, len(objects),
                          tags={'object_type': object_type})
     if notify:
         notify('WATCHDOG=1')
 
 
 def _fix_revision_pypi_empty_string(rev):
     """PyPI loader failed to encode empty strings as bytes, see:
     swh:1:rev:8f0095ee0664867055d03de9bcc8f95b91d8a2b9
     or https://forge.softwareheritage.org/D1772
     """
     rev = {
         **rev,
         'author': rev['author'].copy(),
         'committer': rev['committer'].copy(),
     }
     if rev['author'].get('email') == '':
         rev['author']['email'] = b''
     if rev['author'].get('name') == '':
         rev['author']['name'] = b''
     if rev['committer'].get('email') == '':
         rev['committer']['email'] = b''
     if rev['committer'].get('name') == '':
         rev['committer']['name'] = b''
     return rev
 
 
 def _fix_revision_transplant_source(rev):
     if rev.get('metadata') and rev['metadata'].get('extra_headers'):
         rev = copy.deepcopy(rev)
         rev['metadata']['extra_headers'] = [
             [key, value.encode('ascii')]
             if key == 'transplant_source' and isinstance(value, str)
             else [key, value]
             for (key, value) in rev['metadata']['extra_headers']]
     return rev
 
 
 def _check_date(date):
     """Returns whether the date can be represented in backends with sane
     limits on timestamps and timezones (resp. signed 64-bits and
     signed 16 bits), and that microseconds is valid (ie. between 0 and 10^6).
     """
     if date is None:
         return True
     date = normalize_timestamp(date)
     return (-2**63 <= date['timestamp']['seconds'] < 2**63) \
         and (0 <= date['timestamp']['microseconds'] < 10**6) \
         and (-2**15 <= date['offset'] < 2**15)
 
 
 def _check_revision_date(rev):
     """Exclude revisions with invalid dates.
     See https://forge.softwareheritage.org/T1339"""
     return _check_date(rev['date']) and _check_date(rev['committer_date'])
 
 
 def _fix_revisions(revisions):
     good_revisions = []
     for rev in revisions:
         rev = _fix_revision_pypi_empty_string(rev)
         rev = _fix_revision_transplant_source(rev)
         if not _check_revision_date(rev):
             logging.warning('Excluding revision (invalid date): %r', rev)
             continue
         if rev not in good_revisions:
             good_revisions.append(rev)
     return good_revisions
 
 
 def _fix_origin_visits(visits):
     good_visits = []
     for visit in visits:
         visit = visit.copy()
         if 'type' not in visit:
             if isinstance(visit['origin'], dict) and 'type' in visit['origin']:
                 # Very old version of the schema: visits did not have a type,
                 # but their 'origin' field was a dict with a 'type' key.
                 visit['type'] = visit['origin']['type']
             else:
                 # Very very old version of the schema: 'type' is missing,
                 # so there is nothing we can do to fix it.
                 raise ValueError('Got an origin_visit too old to be replayed.')
         if isinstance(visit['origin'], dict):
             # Old version of the schema: visit['origin'] was a dict.
             visit['origin'] = visit['origin']['url']
         good_visits.append(visit)
     return good_visits
 
 
 def fix_objects(object_type, objects):
     """Converts a possibly old object from the journal to its current
     expected format.
 
     List of conversions:
 
     Empty author name/email in PyPI releases:
 
     >>> from pprint import pprint
     >>> date = {
     ...     'timestamp': {
     ...         'seconds': 1565096932,
     ...         'microseconds': 0,
     ...     },
     ...     'offset': 0,
     ... }
     >>> pprint(fix_objects('revision', [{
     ...     'author': {'email': '', 'fullname': b'', 'name': ''},
     ...     'committer': {'email': '', 'fullname': b'', 'name': ''},
     ...     'date': date,
     ...     'committer_date': date,
     ... }]))
     [{'author': {'email': b'', 'fullname': b'', 'name': b''},
       'committer': {'email': b'', 'fullname': b'', 'name': b''},
       'committer_date': {'offset': 0,
                          'timestamp': {'microseconds': 0, 'seconds': 1565096932}},
       'date': {'offset': 0,
                'timestamp': {'microseconds': 0, 'seconds': 1565096932}}}]
 
     Fix type of 'transplant_source' extra headers:
 
     >>> revs = fix_objects('revision', [{
     ...     'author': {'email': '', 'fullname': b'', 'name': ''},
     ...     'committer': {'email': '', 'fullname': b'', 'name': ''},
     ...     'date': date,
     ...     'committer_date': date,
     ...     'metadata': {
     ...         'extra_headers': [
     ...             ['time_offset_seconds', b'-3600'],
     ...             ['transplant_source', '29c154a012a70f49df983625090434587622b39e']
     ...     ]}
     ... }])
     >>> pprint(revs[0]['metadata']['extra_headers'])
     [['time_offset_seconds', b'-3600'],
      ['transplant_source', b'29c154a012a70f49df983625090434587622b39e']]
 
     Filter out revisions with invalid dates:
 
     >>> from copy import deepcopy
     >>> invalid_date1 = deepcopy(date)
     >>> invalid_date1['timestamp']['microseconds'] = 1000000000  # > 10^6
     >>> fix_objects('revision', [{
     ...     'author': {'email': '', 'fullname': b'', 'name': b''},
     ...     'committer': {'email': '', 'fullname': b'', 'name': b''},
     ...     'date': invalid_date1,
     ...     'committer_date': date,
     ... }])
     []
 
     >>> invalid_date2 = deepcopy(date)
     >>> invalid_date2['timestamp']['seconds'] = 2**70  # > 10^63
     >>> fix_objects('revision', [{
     ...     'author': {'email': '', 'fullname': b'', 'name': b''},
     ...     'committer': {'email': '', 'fullname': b'', 'name': b''},
     ...     'date': invalid_date2,
     ...     'committer_date': date,
     ... }])
     []
 
     >>> invalid_date3 = deepcopy(date)
     >>> invalid_date3['offset'] = 2**20  # > 10^15
     >>> fix_objects('revision', [{
     ...     'author': {'email': '', 'fullname': b'', 'name': b''},
     ...     'committer': {'email': '', 'fullname': b'', 'name': b''},
     ...     'date': date,
     ...     'committer_date': invalid_date3,
     ... }])
     []
 
 
     `visit['origin']` is a dict instead of an URL:
 
     >>> pprint(fix_objects('origin_visit', [{
     ...     'origin': {'url': 'http://foo'},
     ...     'type': 'git',
     ... }]))
     [{'origin': 'http://foo', 'type': 'git'}]
 
     `visit['type']` is missing , but `origin['visit']['type']` exists:
 
     >>> pprint(fix_objects('origin_visit', [
     ...     {'origin': {'type': 'hg', 'url': 'http://foo'}
     ... }]))
     [{'origin': 'http://foo', 'type': 'hg'}]
     """  # noqa
 
     if object_type == 'revision':
         objects = _fix_revisions(objects)
     elif object_type == 'origin_visit':
         objects = _fix_origin_visits(objects)
     return objects
 
 
 def _insert_objects(object_type, objects, storage):
     objects = fix_objects(object_type, objects)
     if object_type == 'content':
         try:
             storage.skipped_content_add(
               (obj for obj in objects if obj.get('status') == 'absent'))
         except HashCollision as e:
             logger.error('(SkippedContent) Hash collision: %s', e.args)
 
         try:
             storage.content_add_metadata(
               (obj for obj in objects if obj.get('status') != 'absent'))
         except HashCollision as e:
             logger.error('(Content) Hash collision: %s', e.args)
 
     elif object_type in ('directory', 'revision', 'release',
                          'snapshot', 'origin'):
         # TODO: split batches that are too large for the storage
         # to handle?
         method = getattr(storage, object_type + '_add')
         method(objects)
     elif object_type == 'origin_visit':
         for visit in objects:
             storage.origin_add_one({'url': visit['origin']})
             if 'metadata' not in visit:
                 visit['metadata'] = None
         storage.origin_visit_upsert(objects)
     else:
         logger.warning('Received a series of %s, this should not happen',
                        object_type)
 
 
 def is_hash_in_bytearray(hash_, array, nb_hashes, hash_size=SHA1_SIZE):
     """
     Checks if the given hash is in the provided `array`. The array must be
     a *sorted* list of sha1 hashes, and contain `nb_hashes` hashes
     (so its size must by `nb_hashes*hash_size` bytes).
 
     Args:
         hash_ (bytes): the hash to look for
         array (bytes): a sorted concatenated array of hashes (may be of
             any type supporting slice indexing, eg. :class:`mmap.mmap`)
         nb_hashes (int): number of hashes in the array
         hash_size (int): size of a hash (defaults to 20, for SHA1)
 
     Example:
 
     >>> import os
     >>> hash1 = os.urandom(20)
     >>> hash2 = os.urandom(20)
     >>> hash3 = os.urandom(20)
     >>> array = b''.join(sorted([hash1, hash2]))
     >>> is_hash_in_bytearray(hash1, array, 2)
     True
     >>> is_hash_in_bytearray(hash2, array, 2)
     True
     >>> is_hash_in_bytearray(hash3, array, 2)
     False
     """
     if len(hash_) != hash_size:
         raise ValueError('hash_ does not match the provided hash_size.')
 
     def get_hash(position):
         return array[position*hash_size:(position+1)*hash_size]
 
     # Regular dichotomy:
     left = 0
     right = nb_hashes
     while left < right-1:
         middle = int((right+left)/2)
         pivot = get_hash(middle)
         if pivot == hash_:
             return True
         elif pivot < hash_:
             left = middle
         else:
             right = middle
     return get_hash(left) == hash_
 
 
+class ReplayError(Exception):
+    """An error occurred during the replay of an object"""
+    def __init__(self, operation, *, obj_id, exc):
+        self.operation = operation
+        self.obj_id = hash_to_hex(obj_id)
+        self.exc = exc
+
+    def __str__(self):
+        return "ReplayError(doing %s, %s, %s)" % (
+            self.operation, self.obj_id, self.exc
+        )
+
+
+def log_replay_retry(retry_obj, sleep, last_result):
+    """Log a retry of the content replayer"""
+    exc = last_result.exception()
+    logger.debug('Retry operation %(operation)s on %(obj_id)s: %(exc)s',
+                 {'operation': exc.operation, 'obj_id': exc.obj_id,
+                  'exc': str(exc.exc)})
+
+    statsd.increment(CONTENT_RETRY_METRIC, tags={
+        'operation': exc.operation,
+        'attempt': str(retry_obj.statistics['attempt_number']),
+    })
+
+
+def log_replay_error(last_attempt):
+    """Log a replay error to sentry"""
+    exc = last_attempt.exception()
+    with push_scope() as scope:
+        scope.set_tag('operation', exc.operation)
+        scope.set_extra('obj_id', exc.obj_id)
+        capture_exception(exc.exc)
+
+    logger.error(
+        'Failed operation %(operation)s on %(obj_id)s after %(retries)s'
+        ' retries: %(exc)s', {
+            'obj_id': exc.obj_id, 'operation': exc.operation,
+            'exc': str(exc.exc), 'retries': last_attempt.attempt_number,
+        })
+
+    return None
+
+
+CONTENT_REPLAY_RETRIES = 3
+
+content_replay_retry = retry(
+    retry=retry_if_exception_type(ReplayError),
+    stop=stop_after_attempt(CONTENT_REPLAY_RETRIES),
+    wait=wait_random_exponential(multiplier=1, max=60),
+    before_sleep=log_replay_retry,
+    retry_error_callback=log_replay_error,
+)
+
+
+@content_replay_retry
 def copy_object(obj_id, src, dst):
     hex_obj_id = hash_to_hex(obj_id)
     obj = ''
     try:
         with statsd.timed(CONTENT_DURATION_METRIC, tags={'request': 'get'}):
             obj = src.get(obj_id)
             logger.debug('retrieved %(obj_id)s', {'obj_id': hex_obj_id})
 
         with statsd.timed(CONTENT_DURATION_METRIC, tags={'request': 'put'}):
             dst.add(obj, obj_id=obj_id, check_presence=False)
             logger.debug('copied %(obj_id)s', {'obj_id': hex_obj_id})
         statsd.increment(CONTENT_BYTES_METRIC, len(obj))
-    except Exception as exc:
-        logger.error('Failed to copy %(obj_id)s: %(exc)s',
-                     {'obj_id': hex_obj_id, 'exc': str(exc)})
+    except ObjNotFoundError:
+        logger.error('Failed to copy %(obj_id)s: object not found',
+                     {'obj_id': hex_obj_id})
         raise
+    except Exception as exc:
+        raise ReplayError('copy', obj_id=obj_id, exc=exc) from None
     return len(obj)
 
 
-@retry(stop=stop_after_attempt(3),
-       reraise=True,
-       wait=wait_random_exponential(multiplier=1, max=60))
+@content_replay_retry
 def obj_in_objstorage(obj_id, dst):
     """Check if an object is already in an objstorage, tenaciously"""
-    return obj_id in dst
+    try:
+        return obj_id in dst
+    except Exception as exc:
+        raise ReplayError('in_dst', obj_id=obj_id, exc=exc) from None
 
 
 def process_replay_objects_content(
         all_objects: Dict[str, List[dict]],
         *,
         src: ObjStorage,
         dst: ObjStorage,
         exclude_fn: Optional[Callable[[dict], bool]] = None,
         check_dst: bool = True,
 ):
     """
     Takes a list of records from Kafka (see
     :py:func:`swh.journal.client.JournalClient.process`) and copies them
     from the `src` objstorage to the `dst` objstorage, if:
 
     * `obj['status']` is `'visible'`
     * `exclude_fn(obj)` is `False` (if `exclude_fn` is provided)
     * `obj['sha1'] not in dst` (if `check_dst` is True)
 
     Args:
         all_objects: Objects passed by the Kafka client. Most importantly,
             `all_objects['content'][*]['sha1']` is the sha1 hash of each
             content.
         src: An object storage (see :py:func:`swh.objstorage.get_objstorage`)
         dst: An object storage (see :py:func:`swh.objstorage.get_objstorage`)
         exclude_fn: Determines whether an object should be copied.
         check_dst: Determines whether we should check the destination
             objstorage before copying.
 
     Example:
 
     >>> from swh.objstorage import get_objstorage
     >>> src = get_objstorage('memory', {})
     >>> dst = get_objstorage('memory', {})
     >>> id1 = src.add(b'foo bar')
     >>> id2 = src.add(b'baz qux')
     >>> kafka_partitions = {
     ...     'content': [
     ...         {
     ...             'sha1': id1,
     ...             'status': 'visible',
     ...         },
     ...         {
     ...             'sha1': id2,
     ...             'status': 'visible',
     ...         },
     ...     ]
     ... }
     >>> process_replay_objects_content(
     ...     kafka_partitions, src=src, dst=dst,
     ...     exclude_fn=lambda obj: obj['sha1'] == id1)
     >>> id1 in dst
     False
     >>> id2 in dst
     True
     """
     vol = []
     nb_skipped = 0
+    nb_failures = 0
     t0 = time()
 
     for (object_type, objects) in all_objects.items():
         if object_type != 'content':
             logger.warning(
                 'Received a series of %s, this should not happen',
                 object_type)
             continue
         for obj in objects:
             obj_id = obj[ID_HASH_ALGO]
             if obj['status'] != 'visible':
                 nb_skipped += 1
                 logger.debug('skipped %s (status=%s)',
                              hash_to_hex(obj_id), obj['status'])
                 statsd.increment(CONTENT_OPERATIONS_METRIC,
                                  tags={"decision": "skipped",
                                        "status": obj["status"]})
             elif exclude_fn and exclude_fn(obj):
                 nb_skipped += 1
                 logger.debug('skipped %s (manually excluded)',
                              hash_to_hex(obj_id))
                 statsd.increment(CONTENT_OPERATIONS_METRIC,
                                  tags={"decision": "excluded"})
             elif check_dst and obj_in_objstorage(obj_id, dst):
                 nb_skipped += 1
                 logger.debug('skipped %s (in dst)', hash_to_hex(obj_id))
                 statsd.increment(CONTENT_OPERATIONS_METRIC,
                                  tags={"decision": "in_dst"})
             else:
                 try:
                     copied = copy_object(obj_id, src, dst)
                 except ObjNotFoundError:
                     nb_skipped += 1
                     statsd.increment(CONTENT_OPERATIONS_METRIC,
                                      tags={"decision": "not_in_src"})
                 else:
-                    vol.append(copied)
-                    statsd.increment(CONTENT_OPERATIONS_METRIC,
-                                     tags={"decision": "copied"})
+                    if copied is None:
+                        nb_failures += 1
+                        statsd.increment(CONTENT_OPERATIONS_METRIC,
+                                         tags={"decision": "failed"})
+                    else:
+                        vol.append(copied)
+                        statsd.increment(CONTENT_OPERATIONS_METRIC,
+                                         tags={"decision": "copied"})
 
     dt = time() - t0
     logger.info(
         'processed %s content objects in %.1fsec '
         '(%.1f obj/sec, %.1fMB/sec) - %d failures - %d skipped',
         len(vol), dt,
         len(vol)/dt,
         sum(vol)/1024/1024/dt,
-        len([x for x in vol if not x]),
+        nb_failures,
         nb_skipped)
 
     if notify:
         notify('WATCHDOG=1')
diff --git a/swh/journal/tests/test_cli.py b/swh/journal/tests/test_cli.py
index 809f7e0..6ff7650 100644
--- a/swh/journal/tests/test_cli.py
+++ b/swh/journal/tests/test_cli.py
@@ -1,516 +1,617 @@
 # Copyright (C) 2019  The Software Heritage developers
 # See the AUTHORS file at the top-level directory of this distribution
 # License: GNU General Public License version 3, or any later version
 # See top-level LICENSE file for more information
 
 from collections import Counter
 import functools
 import logging
 import re
 import tempfile
 from subprocess import Popen
 from typing import Any, Dict, Tuple
 from unittest.mock import patch
 
 from click.testing import CliRunner
 from confluent_kafka import Producer
 import pytest
 
 from swh.model.hashutil import hash_to_hex
 from swh.objstorage.backends.in_memory import InMemoryObjStorage
 from swh.storage import get_storage
 
 from swh.journal.cli import cli
+from swh.journal.replay import CONTENT_REPLAY_RETRIES
 from swh.journal.serializers import key_to_kafka, value_to_kafka
 
 
 logger = logging.getLogger(__name__)
 
 
 CLI_CONFIG = '''
 storage:
     cls: memory
     args: {}
 objstorage_src:
     cls: mocked
     args:
         name: src
 objstorage_dst:
     cls: mocked
     args:
         name: dst
 '''
 
 
 @pytest.fixture
 def storage():
     """An swh-storage object that gets injected into the CLI functions."""
     storage_config = {
         'cls': 'pipeline',
         'steps': [
             {'cls': 'validate'},
             {'cls': 'memory'},
         ]
     }
     storage = get_storage(**storage_config)
     with patch('swh.journal.cli.get_storage') as get_storage_mock:
         get_storage_mock.return_value = storage
         yield storage
 
 
+@pytest.fixture
+def monkeypatch_retry_sleep(monkeypatch):
+    from swh.journal.replay import copy_object, obj_in_objstorage
+    monkeypatch.setattr(copy_object.retry, 'sleep', lambda x: None)
+    monkeypatch.setattr(obj_in_objstorage.retry, 'sleep', lambda x: None)
+
+
 def invoke(catch_exceptions, args, env=None):
     runner = CliRunner()
     with tempfile.NamedTemporaryFile('a', suffix='.yml') as config_fd:
         config_fd.write(CLI_CONFIG)
         config_fd.seek(0)
         args = ['-C' + config_fd.name] + args
         result = runner.invoke(
             cli, args, obj={'log_level': logging.DEBUG}, env=env,
         )
     if not catch_exceptions and result.exception:
         print(result.output)
         raise result.exception
     return result
 
 
 def test_replay(
         storage,
         kafka_prefix: str,
         kafka_consumer_group: str,
         kafka_server: Tuple[Popen, int]):
     (_, port) = kafka_server
     kafka_prefix += '.swh.journal.objects'
 
     producer = Producer({
         'bootstrap.servers': 'localhost:{}'.format(port),
         'client.id': 'test-producer',
         'enable.idempotence': 'true',
     })
 
     snapshot = {'id': b'foo', 'branches': {
         b'HEAD': {
             'target_type': 'revision',
             'target': b'\x01'*20,
         }
     }}  # type: Dict[str, Any]
     producer.produce(
         topic=kafka_prefix+'.snapshot',
         key=key_to_kafka(snapshot['id']),
         value=value_to_kafka(snapshot),
     )
     producer.flush()
 
     logger.debug('Flushed producer')
 
     result = invoke(False, [
         'replay',
         '--broker', '127.0.0.1:%d' % port,
         '--group-id', kafka_consumer_group,
         '--prefix', kafka_prefix,
         '--max-messages', '1',
     ])
     expected = r'Done.\n'
     assert result.exit_code == 0, result.output
     assert re.fullmatch(expected, result.output, re.MULTILINE), result.output
 
     assert storage.snapshot_get(snapshot['id']) == {
         **snapshot, 'next_branch': None}
 
 
 def _patch_objstorages(names):
     objstorages = {name: InMemoryObjStorage() for name in names}
 
     def get_mock_objstorage(cls, args):
         assert cls == 'mocked', cls
         return objstorages[args['name']]
 
     def decorator(f):
         @functools.wraps(f)
         @patch('swh.journal.cli.get_objstorage')
         def newf(get_objstorage_mock, *args, **kwargs):
             get_objstorage_mock.side_effect = get_mock_objstorage
             f(*args, objstorages=objstorages, **kwargs)
 
         return newf
 
     return decorator
 
 
 NUM_CONTENTS = 10
 
 
 def _fill_objstorage_and_kafka(kafka_port, kafka_prefix, objstorages):
     producer = Producer({
         'bootstrap.servers': '127.0.0.1:{}'.format(kafka_port),
         'client.id': 'test-producer',
         'enable.idempotence': 'true',
     })
 
     contents = {}
     for i in range(NUM_CONTENTS):
         content = b'\x00'*19 + bytes([i])
         sha1 = objstorages['src'].add(content)
         contents[sha1] = content
         producer.produce(
             topic=kafka_prefix+'.content',
             key=key_to_kafka(sha1),
             value=key_to_kafka({
                 'sha1': sha1,
                 'status': 'visible',
             }),
         )
 
     producer.flush()
 
     return contents
 
 
 @_patch_objstorages(['src', 'dst'])
 def test_replay_content(
         objstorages,
         storage,
         kafka_prefix: str,
         kafka_consumer_group: str,
         kafka_server: Tuple[Popen, int]):
     (_, kafka_port) = kafka_server
     kafka_prefix += '.swh.journal.objects'
 
     contents = _fill_objstorage_and_kafka(
         kafka_port, kafka_prefix, objstorages)
 
     result = invoke(False, [
         'content-replay',
         '--broker', '127.0.0.1:%d' % kafka_port,
         '--group-id', kafka_consumer_group,
         '--prefix', kafka_prefix,
         '--max-messages', str(NUM_CONTENTS),
     ])
     expected = r'Done.\n'
     assert result.exit_code == 0, result.output
     assert re.fullmatch(expected, result.output, re.MULTILINE), result.output
 
     for (sha1, content) in contents.items():
         assert sha1 in objstorages['dst'], sha1
         assert objstorages['dst'].get(sha1) == content
 
 
 @_patch_objstorages(['src', 'dst'])
 def test_replay_content_structured_log(
         objstorages,
         storage,
         kafka_prefix: str,
         kafka_consumer_group: str,
         kafka_server: Tuple[Popen, int],
         caplog):
     (_, kafka_port) = kafka_server
     kafka_prefix += '.swh.journal.objects'
 
     contents = _fill_objstorage_and_kafka(
         kafka_port, kafka_prefix, objstorages)
 
     caplog.set_level(logging.DEBUG, 'swh.journal.replay')
 
     expected_obj_ids = set(hash_to_hex(sha1) for sha1 in contents)
 
     result = invoke(False, [
         'content-replay',
         '--broker', '127.0.0.1:%d' % kafka_port,
         '--group-id', kafka_consumer_group,
         '--prefix', kafka_prefix,
         '--max-messages', str(NUM_CONTENTS),
     ])
     expected = r'Done.\n'
     assert result.exit_code == 0, result.output
     assert re.fullmatch(expected, result.output, re.MULTILINE), result.output
 
     copied = set()
     for record in caplog.records:
         logtext = record.getMessage()
         if 'copied' in logtext:
             copied.add(record.args['obj_id'])
 
     assert copied == expected_obj_ids, (
         "Mismatched logging; see captured log output for details."
     )
 
 
 @_patch_objstorages(['src', 'dst'])
 def test_replay_content_static_group_id(
         objstorages,
         storage,
         kafka_prefix: str,
         kafka_consumer_group: str,
         kafka_server: Tuple[Popen, int],
         caplog):
     (_, kafka_port) = kafka_server
     kafka_prefix += '.swh.journal.objects'
 
     contents = _fill_objstorage_and_kafka(
         kafka_port, kafka_prefix, objstorages)
 
     # Setup log capture to fish the consumer settings out of the log messages
     caplog.set_level(logging.DEBUG, 'swh.journal.client')
 
     result = invoke(False, [
         'content-replay',
         '--broker', '127.0.0.1:%d' % kafka_port,
         '--group-id', kafka_consumer_group,
         '--prefix', kafka_prefix,
         '--max-messages', str(NUM_CONTENTS),
     ], {'KAFKA_GROUP_INSTANCE_ID': 'static-group-instance-id'})
     expected = r'Done.\n'
     assert result.exit_code == 0, result.output
     assert re.fullmatch(expected, result.output, re.MULTILINE), result.output
 
     consumer_settings = None
     for record in caplog.records:
         if 'Consumer settings' in record.message:
             consumer_settings = record.args
             break
 
     assert consumer_settings is not None, (
         'Failed to get consumer settings out of the consumer log. '
         'See log capture for details.'
     )
     assert consumer_settings['group.instance.id'] == 'static-group-instance-id'
     assert consumer_settings['session.timeout.ms'] == 60 * 10 * 1000
     assert consumer_settings['max.poll.interval.ms'] == 90 * 10 * 1000
 
     for (sha1, content) in contents.items():
         assert sha1 in objstorages['dst'], sha1
         assert objstorages['dst'].get(sha1) == content
 
 
 @_patch_objstorages(['src', 'dst'])
 def test_replay_content_exclude(
         objstorages,
         storage,
         kafka_prefix: str,
         kafka_consumer_group: str,
         kafka_server: Tuple[Popen, int]):
     (_, kafka_port) = kafka_server
     kafka_prefix += '.swh.journal.objects'
 
     contents = _fill_objstorage_and_kafka(
         kafka_port, kafka_prefix, objstorages)
 
     excluded_contents = list(contents)[0::2]  # picking half of them
     with tempfile.NamedTemporaryFile(mode='w+b') as fd:
         fd.write(b''.join(sorted(excluded_contents)))
 
         fd.seek(0)
 
         result = invoke(False, [
             'content-replay',
             '--broker', '127.0.0.1:%d' % kafka_port,
             '--group-id', kafka_consumer_group,
             '--prefix', kafka_prefix,
             '--max-messages', str(NUM_CONTENTS),
             '--exclude-sha1-file', fd.name,
         ])
     expected = r'Done.\n'
     assert result.exit_code == 0, result.output
     assert re.fullmatch(expected, result.output, re.MULTILINE), result.output
 
     for (sha1, content) in contents.items():
         if sha1 in excluded_contents:
             assert sha1 not in objstorages['dst'], sha1
         else:
             assert sha1 in objstorages['dst'], sha1
             assert objstorages['dst'].get(sha1) == content
 
 
 NUM_CONTENTS_DST = 5
 
 
 @_patch_objstorages(['src', 'dst'])
 @pytest.mark.parametrize("check_dst,expected_copied,expected_in_dst", [
     (True, NUM_CONTENTS - NUM_CONTENTS_DST, NUM_CONTENTS_DST),
     (False, NUM_CONTENTS, 0),
 ])
 def test_replay_content_check_dst(
         objstorages,
         storage,
         kafka_prefix: str,
         kafka_consumer_group: str,
         kafka_server: Tuple[Popen, int],
         check_dst: bool,
         expected_copied: int,
         expected_in_dst: int,
         caplog):
     (_, kafka_port) = kafka_server
     kafka_prefix += '.swh.journal.objects'
 
     contents = _fill_objstorage_and_kafka(
         kafka_port, kafka_prefix, objstorages)
 
     for i, (sha1, content) in enumerate(contents.items()):
         if i >= NUM_CONTENTS_DST:
             break
 
         objstorages['dst'].add(content, obj_id=sha1)
 
     caplog.set_level(logging.DEBUG, 'swh.journal.replay')
 
     result = invoke(False, [
         'content-replay',
         '--broker', '127.0.0.1:%d' % kafka_port,
         '--group-id', kafka_consumer_group,
         '--prefix', kafka_prefix,
         '--max-messages', str(NUM_CONTENTS),
         '--check-dst' if check_dst else '--no-check-dst',
     ])
     expected = r'Done.\n'
     assert result.exit_code == 0, result.output
     assert re.fullmatch(expected, result.output, re.MULTILINE), result.output
 
     copied = 0
     in_dst = 0
     for record in caplog.records:
         logtext = record.getMessage()
         if 'copied' in logtext:
             copied += 1
         elif 'in dst' in logtext:
             in_dst += 1
 
     assert (copied == expected_copied and in_dst == expected_in_dst), (
         "Unexpected amount of objects copied, see the captured log for details"
     )
 
     for (sha1, content) in contents.items():
         assert sha1 in objstorages['dst'], sha1
         assert objstorages['dst'].get(sha1) == content
 
 
 class FlakyObjStorage(InMemoryObjStorage):
     def __init__(self, *args, **kwargs):
         state = kwargs.pop('state')
         self.failures_left = Counter(kwargs.pop('failures'))
         super().__init__(*args, **kwargs)
         if state:
             self.state = state
 
     def flaky_operation(self, op, obj_id):
         if self.failures_left[op, obj_id] > 0:
             self.failures_left[op, obj_id] -= 1
             raise RuntimeError(
                 'Failed %s on %s' % (op, hash_to_hex(obj_id))
             )
 
     def get(self, obj_id):
         self.flaky_operation('get', obj_id)
         return super().get(obj_id)
 
     def add(self, data, obj_id=None, check_presence=True):
         self.flaky_operation('add', obj_id)
         return super().add(data, obj_id=obj_id,
                            check_presence=check_presence)
 
     def __contains__(self, obj_id):
         self.flaky_operation('in', obj_id)
         return super().__contains__(obj_id)
 
 
 @_patch_objstorages(['src', 'dst'])
 def test_replay_content_check_dst_retry(
         objstorages,
         storage,
         kafka_prefix: str,
         kafka_consumer_group: str,
-        kafka_server: Tuple[Popen, int]):
+        kafka_server: Tuple[Popen, int],
+        monkeypatch_retry_sleep):
     (_, kafka_port) = kafka_server
     kafka_prefix += '.swh.journal.objects'
 
     contents = _fill_objstorage_and_kafka(
         kafka_port, kafka_prefix, objstorages)
 
     failures = {}
     for i, (sha1, content) in enumerate(contents.items()):
         if i >= NUM_CONTENTS_DST:
             break
 
         objstorages['dst'].add(content, obj_id=sha1)
         failures['in', sha1] = 1
 
     orig_dst = objstorages['dst']
     objstorages['dst'] = FlakyObjStorage(state=orig_dst.state,
                                          failures=failures)
 
     result = invoke(False, [
         'content-replay',
         '--broker', '127.0.0.1:%d' % kafka_port,
         '--group-id', kafka_consumer_group,
         '--prefix', kafka_prefix,
         '--max-messages', str(NUM_CONTENTS),
         '--check-dst',
     ])
     expected = r'Done.\n'
     assert result.exit_code == 0, result.output
     assert re.fullmatch(expected, result.output, re.MULTILINE), result.output
 
     for (sha1, content) in contents.items():
         assert sha1 in objstorages['dst'], sha1
         assert objstorages['dst'].get(sha1) == content
 
 
+@_patch_objstorages(['src', 'dst'])
+def test_replay_content_failed_copy_retry(
+        objstorages,
+        storage,
+        kafka_prefix: str,
+        kafka_consumer_group: str,
+        kafka_server: Tuple[Popen, int],
+        caplog,
+        monkeypatch_retry_sleep):
+    (_, kafka_port) = kafka_server
+    kafka_prefix += '.swh.journal.objects'
+
+    contents = _fill_objstorage_and_kafka(
+        kafka_port, kafka_prefix, objstorages)
+
+    add_failures = {}
+    get_failures = {}
+    definitely_failed = set()
+
+    # We want to generate operations failing 1 to CONTENT_REPLAY_RETRIES times.
+    # We generate failures for 2 different operations, get and add.
+    num_retry_contents = 2 * CONTENT_REPLAY_RETRIES
+
+    assert num_retry_contents < NUM_CONTENTS, (
+        "Need to generate more test contents to properly test retry behavior"
+    )
+
+    for i, sha1 in enumerate(contents):
+        if i >= num_retry_contents:
+            break
+
+        # This generates a number of failures, up to CONTENT_REPLAY_RETRIES
+        num_failures = (i % CONTENT_REPLAY_RETRIES) + 1
+
+        # This generates failures of add for the first CONTENT_REPLAY_RETRIES
+        # objects, then failures of get.
+        if i < CONTENT_REPLAY_RETRIES:
+            add_failures['add', sha1] = num_failures
+        else:
+            get_failures['get', sha1] = num_failures
+
+        # Only contents that have CONTENT_REPLAY_RETRIES or more are
+        # definitely failing
+        if num_failures >= CONTENT_REPLAY_RETRIES:
+            definitely_failed.add(hash_to_hex(sha1))
+
+    objstorages['dst'] = FlakyObjStorage(
+        state=objstorages['dst'].state,
+        failures=add_failures,
+    )
+    objstorages['src'] = FlakyObjStorage(
+        state=objstorages['src'].state,
+        failures=get_failures,
+    )
+
+    caplog.set_level(logging.DEBUG, 'swh.journal.replay')
+
+    result = invoke(False, [
+        'content-replay',
+        '--broker', '127.0.0.1:%d' % kafka_port,
+        '--group-id', kafka_consumer_group,
+        '--prefix', kafka_prefix,
+        '--max-messages', str(NUM_CONTENTS),
+    ])
+    expected = r'Done.\n'
+    assert result.exit_code == 0, result.output
+    assert re.fullmatch(expected, result.output, re.MULTILINE), result.output
+
+    copied = 0
+    actually_failed = set()
+    for record in caplog.records:
+        logtext = record.getMessage()
+        if 'copied' in logtext:
+            copied += 1
+        elif 'Failed operation' in logtext:
+            assert record.levelno == logging.ERROR
+            assert record.args['retries'] == CONTENT_REPLAY_RETRIES
+            actually_failed.add(record.args['obj_id'])
+
+    assert actually_failed == definitely_failed, (
+        'Unexpected object copy failures; see captured log for details'
+    )
+
+    for (sha1, content) in contents.items():
+        if hash_to_hex(sha1) in definitely_failed:
+            assert sha1 not in objstorages['dst']
+            continue
+
+        assert sha1 in objstorages['dst'], sha1
+        assert objstorages['dst'].get(sha1) == content
+
+
 @_patch_objstorages(['src', 'dst'])
 def test_replay_content_objnotfound(
         objstorages,
         storage,
         kafka_prefix: str,
         kafka_consumer_group: str,
         kafka_server: Tuple[Popen, int],
         caplog):
     (_, kafka_port) = kafka_server
     kafka_prefix += '.swh.journal.objects'
 
     contents = _fill_objstorage_and_kafka(
         kafka_port, kafka_prefix, objstorages)
 
     num_contents_deleted = 5
     contents_deleted = set()
 
     for i, sha1 in enumerate(contents):
         if i >= num_contents_deleted:
             break
 
         del objstorages['src'].state[sha1]
         contents_deleted.add(hash_to_hex(sha1))
 
     caplog.set_level(logging.DEBUG, 'swh.journal.replay')
 
     result = invoke(False, [
         'content-replay',
         '--broker', '127.0.0.1:%d' % kafka_port,
         '--group-id', kafka_consumer_group,
         '--prefix', kafka_prefix,
         '--max-messages', str(NUM_CONTENTS),
     ])
     expected = r'Done.\n'
     assert result.exit_code == 0, result.output
     assert re.fullmatch(expected, result.output, re.MULTILINE), result.output
 
     copied = 0
     not_in_src = set()
     for record in caplog.records:
         logtext = record.getMessage()
         if 'copied' in logtext:
             copied += 1
         elif 'object not found' in logtext:
             # Check that the object id can be recovered from logs
             assert record.levelno == logging.ERROR
             not_in_src.add(record.args['obj_id'])
 
     assert copied == NUM_CONTENTS - num_contents_deleted, (
         "Unexpected number of contents copied"
     )
 
     assert not_in_src == contents_deleted, (
         "Mismatch between deleted contents and not_in_src logs"
     )
 
     for (sha1, content) in contents.items():
         if sha1 not in objstorages['src']:
             continue
         assert sha1 in objstorages['dst'], sha1
         assert objstorages['dst'].get(sha1) == content