duplicate key value violates unique constraint "room_retention_pkey"

This issue has been migrated from [#11448](https://github.com/matrix-org/synapse/issues/11448).

---

with v1.48.0rc1 is see this error:

```
2021-11-27 09:58:24,896 - synapse.http.server - 97 - ERROR - POST-2657 - Failed handle request via 'ReplicationFederationSendEventsRestServlet': [26/88052]
Traceback (most recent call last):
File "/opt/venvs/matrix-synapse/lib/python3.9/site-packages/twisted/internet/defer.py", line 1657, in _inlineCallbacks
result = current_context.run(
File "/opt/venvs/matrix-synapse/lib/python3.9/site-packages/twisted/python/failure.py", line 500, in throwExceptionIntoGenerator
return g.throw(self.type, self.value, self.tb)
File "/opt/venvs/matrix-synapse/lib/python3.9/site-packages/synapse/replication/http/federation.py", line 134, in _handle_request
max_stream_id = await self.federation_event_handler.persist_events_and_notify(
File "/opt/venvs/matrix-synapse/lib/python3.9/site-packages/synapse/handlers/federation_event.py", line 1861, in persist_events_and_notify
events, max_stream_token = await self._storage.persistence.persist_events(
File "/opt/venvs/matrix-synapse/lib/python3.9/site-packages/synapse/logging/opentracing.py", line 785, in _trace_inner
return await func(*args, **kwargs)
File "/opt/venvs/matrix-synapse/lib/python3.9/site-packages/synapse/storage/persist_events.py", line 326, in persist_events
ret_vals = await yieldable_gather_results(enqueue, partitioned.items())
File "/opt/venvs/matrix-synapse/lib/python3.9/site-packages/synapse/storage/persist_events.py", line 243, in handle_queue_loop
ret = await self._per_item_callback(
File "/opt/venvs/matrix-synapse/lib/python3.9/site-packages/synapse/storage/persist_events.py", line 581, in _persist_event_batch
await self.persist_events_store._persist_events_and_state_updates(
File "/opt/venvs/matrix-synapse/lib/python3.9/site-packages/synapse/storage/databases/main/events.py", line 175, in _persist_events_and_state_updates
await self.db_pool.runInteraction(
File "/opt/venvs/matrix-synapse/lib/python3.9/site-packages/synapse/storage/database.py", line 699, in runInteraction
result = await self.runWithConnection(
File "/opt/venvs/matrix-synapse/lib/python3.9/site-packages/synapse/storage/database.py", line 804, in runWithConnection
return await make_deferred_yieldable(
File "/opt/venvs/matrix-synapse/lib/python3.9/site-packages/twisted/python/threadpool.py", line 238, in inContext
result = inContext.theWork() # type: ignore[attr-defined]
File "/opt/venvs/matrix-synapse/lib/python3.9/site-packages/twisted/python/threadpool.py", line 254, in
inContext.theWork = lambda: context.call( # type: ignore[attr-defined]
File "/opt/venvs/matrix-synapse/lib/python3.9/site-packages/twisted/python/context.py", line 118, in callWithContext
return self.currentContext().callWithContext(ctx, func, *args, **kw)
File "/opt/venvs/matrix-synapse/lib/python3.9/site-packages/twisted/python/context.py", line 83, in callWithContext
return func(*args, **kw)
File "/opt/venvs/matrix-synapse/lib/python3.9/site-packages/twisted/enterprise/adbapi.py", line 293, in _runWithConnection
compat.reraise(excValue, excTraceback)
File "/opt/venvs/matrix-synapse/lib/python3.9/site-packages/twisted/python/deprecate.py", line 298, in deprecatedFunction
return function(*args, **kwargs)
File "/opt/venvs/matrix-synapse/lib/python3.9/site-packages/twisted/python/compat.py", line 404, in reraise
raise exception.with_traceback(traceback)
File "/opt/venvs/matrix-synapse/lib/python3.9/site-packages/twisted/enterprise/adbapi.py", line 284, in _runWithConnection
result = func(conn, *args, **kw)
File "/opt/venvs/matrix-synapse/lib/python3.9/site-packages/synapse/storage/database.py", line 799, in inner_func
return func(db_conn, *args, **kwargs)
File "/opt/venvs/matrix-synapse/lib/python3.9/site-packages/synapse/storage/database.py", line 567, in new_transaction
r = func(cursor, *args, **kwargs)
File "/opt/venvs/matrix-synapse/lib/python3.9/site-packages/synapse/logging/utils.py", line 73, in wrapped
return f(*args, **kwargs)
File "/opt/venvs/matrix-synapse/lib/python3.9/site-packages/synapse/storage/databases/main/events.py", line 397, in _persist_events_txn
self._update_metadata_tables_txn(
File "/opt/venvs/matrix-synapse/lib/python3.9/site-packages/synapse/storage/databases/main/events.py", line 1488, in _update_metadata_tables_txn
self._store_retention_policy_for_room_txn(txn, event)
File "/opt/venvs/matrix-synapse/lib/python3.9/site-packages/synapse/storage/databases/main/events.py", line 1918, in _store_retention_policy_for_room_txn
self.db_pool.simple_insert_txn(
File "/opt/venvs/matrix-synapse/lib/python3.9/site-packages/synapse/storage/database.py", line 892, in simple_insert_txn
txn.execute(sql, vals)
File "/opt/venvs/matrix-synapse/lib/python3.9/site-packages/synapse/storage/database.py", line 307, in execute
self._do_execute(self.txn.execute, sql, *args)
File "/opt/venvs/matrix-synapse/lib/python3.9/site-packages/synapse/storage/database.py", line 340, in _do_execute
return func(sql, *args)
psycopg2.errors.UniqueViolation: duplicate key value violates unique constraint "room_retention_pkey"
DETAIL: Key (room_id, event_id)=(!SOPFfiEojgdYhVcoaL:matrix.org, $e1bQrYY4-Hh36Vrc-r2kNqv86x3JNhNATb4j959MT3E) already exists.
```

_Originally posted by @cremesk in https://github.com/matrix-org/synapse/issues/11433#issuecomment-980534491_