Faster joins: fix race in calcuating "current state"

[richvdh]

Once we finish un-partial-stating events, we update the current_state_events table to include the complete state of the room. However, that can race against the persistence of an event, which also updates this table - so we may end up with completely bogus "current state" data.

synapse/synapse/storage/controllers/persist_events.py

Lines 384 to 399 in 7c6b220

	async def update_current_state(self, room_id: str) -> None:
	"""Recalculate the current state for a room, and persist it"""
	state = await self._calculate_current_state(room_id)
	delta = await self._calculate_state_delta(room_id, state)
	# TODO(faster_joins): get a real stream ordering, to make this work correctly
	# across workers.
	# https://github.com/matrix-org/synapse/issues/12994
	#
	# TODO(faster_joins): this can race against event persistence, in which case we
	# will end up with incorrect state. Perhaps we should make this a job we
	# farm out to the event persister thread, somehow.
	# https://github.com/matrix-org/synapse/issues/13007
	#
	stream_id = self.main_store.get_room_max_stream_ordering()
	await self.persist_events_store.update_current_state(room_id, delta, stream_id)

Related: #12988

[squahtx]

Having stared at the code for a bit:
update_current_state (above) is inherently racy:

• If we persist an event in between _calculate_current_state and persist_events_store.update_current_state, we'll overwrite the partial current state with a stale non-partial current state.
• If we persist an event during persist_events_store.update_current_state, we can end up with an incoherent current state. I've not yet convinced myself that postgres will raise a SerializationFailure in this case.

In either case, _sync_partial_state_room will then fail to clear the partial state flag for the room and go round again:

synapse/synapse/handlers/federation.py

Lines 1540 to 1575 in a164a46

	while True:
	batch = await self.store.get_partial_state_events_batch(room_id)
	if not batch:
	# all the events are updated, so we can update current state and
	# clear the lazy-loading flag.
	logger.info("Updating current state for %s", room_id)
	# TODO(faster_joins): support workers
	# https://github.com/matrix-org/synapse/issues/12994
	assert (
	self._storage_controllers.persistence is not None
	), "worker-mode deployments not currently supported here"
	await self._storage_controllers.persistence.update_current_state(
	room_id
	)
	logger.info("Clearing partial-state flag for %s", room_id)
	success = await self.store.clear_partial_state_room(room_id)
	if success:
	logger.info("State resync complete for %s", room_id)
	self._storage_controllers.state.notify_room_un_partial_stated(
	room_id
	)
	# TODO(faster_joins) update room stats and user directory?
	# https://github.com/matrix-org/synapse/issues/12814
	# https://github.com/matrix-org/synapse/issues/12815
	return
	# we raced against more events arriving with partial state. Go round
	# the loop again. We've already logged a warning, so no need for more.
	# TODO(faster_joins): there is still a race here, whereby incoming events which raced
	# with us will fail to be persisted after the call to `clear_partial_state_room` due to
	# having partial state.
	# https://github.com/matrix-org/synapse/issues/12988
	#
	continue

The unwanted current state ought to be short lived, if all goes well.

[squahtx]

The easiest fix to reason about would be to move the entirety of update_current_state to the event persister queue on the correct worker. There's supposed to be only one event persist going on per room at a time.

I think this'll require a new replication endpoint plus modification of the queue so that it can hold two types of task.

[squahtx]

Note to self: we determine the correct worker to send replication requests to like so:

synapse/synapse/handlers/message.py

Lines 1345 to 1349 in 92103cb

	# If we're a worker we need to hit out to the master.
	writer_instance = self._events_shard_config.get_instance(event.room_id)
	if writer_instance != self._instance_name:
	result = await self.send_event(
	instance_name=writer_instance,

where

self._events_shard_config = self.config.worker.events_shard_config

and we assert that we are running on the correct worker like so:

assert self._events_shard_config.should_handle(self._instance_name, event.room_id)`