Handle duplicate OTK uploads racing

[erikjohnston]

Currently this causes one of then to 500.

Stack trace

UniqueViolation: duplicate key value violates unique constraint "e2e_one_time_keys_json_uniqueness"
DETAIL:  Key (user_id, device_id, algorithm, key_id)=(xxx, xxx, xxx, xxx) already exists.

  File "synapse/http/server.py", line 332, in _async_render_wrapper
    callback_return = await self._async_render(request)
  File "synapse/http/server.py", line 544, in _async_render
    callback_return = await raw_callback_return
  File "synapse/replication/http/_base.py", line 433, in _check_auth_and_handle
    code, response = await self._handle_request(request, content, **kwargs)
  File "synapse/replication/http/devices.py", line 161, in _handle_request
    results = await self.e2e_keys_handler.upload_keys_for_user(
  File "synapse/logging/opentracing.py", line 921, in _wrapper
    return await func(*args, **kwargs)
  File "synapse/handlers/e2e_keys.py", line 817, in upload_keys_for_user
    await self._upload_one_time_keys_for_user(
  File "synapse/handlers/e2e_keys.py", line 896, in _upload_one_time_keys_for_user
    await self.store.add_e2e_one_time_keys(user_id, device_id, time_now, new_keys)
  File "synapse/storage/databases/main/end_to_end_keys.py", line 538, in add_e2e_one_time_keys
    await self.db_pool.runInteraction(
  File "synapse/storage/database.py", line 950, in runInteraction
    return await delay_cancellation(_runInteraction())
  File "twisted/internet/defer.py", line 1999, in _inlineCallbacks
    result = context.run(
  File "twisted/python/failure.py", line 519, in throwExceptionIntoGenerator
    return g.throw(self.value.with_traceback(self.tb))
  File "synapse/storage/database.py", line 916, in _runInteraction
    result: R = await self.runWithConnection(
  File "synapse/storage/database.py", line 1045, in runWithConnection
    return await make_deferred_yieldable(
  File "twisted/python/threadpool.py", line 269, in inContext
    result = inContext.theWork()  # type: ignore[attr-defined]
  File "twisted/python/threadpool.py", line 285, in <lambda>
    inContext.theWork = lambda: context.call(  # type: ignore[attr-defined]
  File "twisted/python/context.py", line 117, in callWithContext
    return self.currentContext().callWithContext(ctx, func, *args, **kw)
  File "twisted/python/context.py", line 82, in callWithContext
    return func(*args, **kw)
  File "twisted/enterprise/adbapi.py", line 282, in _runWithConnection
    result = func(conn, *args, **kw)
  File "synapse/storage/database.py", line 1038, in inner_func
    return func(db_conn, *args, **kwargs)
  File "synapse/storage/database.py", line 778, in new_transaction
    r = func(cursor, *args, **kwargs)
  File "synapse/storage/databases/main/end_to_end_keys.py", line 571, in _add_e2e_one_time_keys_txn
    self.db_pool.simple_insert_many_txn(
  File "synapse/storage/database.py", line 1150, in simple_insert_many_txn
    txn.execute_values(sql, values, fetch=False)
  File "synapse/storage/database.py", line 413, in execute_values
    return self._do_execute(
  File "synapse/storage/database.py", line 486, in _do_execute
    return func(sql, *args, **kwargs)
  File "synapse/storage/database.py", line 416, in <lambda>
    lambda the_sql, the_values: execute_values(
  File "psycopg2/extras.py", line 1299, in execute_values
    cur.execute(b''.join(parts))

[reivilibre]

Part of me wonders if we'd rather INSERT OR IGNORE instead, but maybe this is just a plain saner approach and I don't think we should be afraid of linearising these requests for a single device, so fine.

[erikjohnston]

Part of me wonders if we'd rather INSERT OR IGNORE instead, but maybe this is just a plain saner approach and I don't think we should be afraid of linearising these requests for a single device, so fine.

Yeah, though we'd then need to be careful that we are only ignoring where the keys are actually the same, which is a bit harder for JSON