Read new column

synapse/handlers/device.py

@@ -937,7 +937,7 @@ async def incoming_device_list_update(
  # Check if we are partially joining any rooms. If so we need to store
  # all device list updates so that we can handle them correctly once we
  # know who is in the room.
- partial_rooms = await self.store.get_partial_state_rooms_and_servers()
+ partial_rooms = await self.store.get_partial_state_room_resync_info()
  if partial_rooms:
  await self.store.add_remote_device_list_to_pending(
  user_id,

synapse/handlers/federation.py

@@ -1603,13 +1603,13 @@ async def _resume_sync_partial_state_room(self) -> None:
  """Resumes resyncing of all partial-state rooms after a restart."""
  assert not self.config.worker.worker_app
 
- partial_state_rooms = await self.store.get_partial_state_rooms_and_servers()
- for room_id, servers_in_room in partial_state_rooms.items():
+ partial_state_rooms = await self.store.get_partial_state_room_resync_info()
+ for room_id, resync_info in partial_state_rooms.items():
  run_as_background_process(
  desc="sync_partial_state_room",
  func=self._sync_partial_state_room,
  initial_destination=None,
- other_destinations=servers_in_room,
+ other_destinations=resync_info.servers_in_room,
  room_id=room_id,
  )
 

synapse/storage/database.py

@@ -1658,7 +1658,7 @@ async def simple_select_one_onecol(
  table: string giving the table name
  keyvalues: dict of column names and values to select the row with
  retcol: string giving the name of the column to return
- allow_none: If true, return None instead of failing if the SELECT
+ allow_none: If true, return None instead of raising StoreError if the SELECT
  statement returns no rows
  desc: description of the transaction, for logging and metrics
  """

synapse/storage/databases/main/room.py

@@ -97,6 +97,12 @@ class RoomSortOrder(Enum):
  STATE_EVENTS = "state_events"
 
 
+@attr.s(slots=True, frozen=True, auto_attribs=True)
+class PartialStateResyncInfo:
+ joined_via: Optional[str]
+ servers_in_room: List[str] = attr.ib(factory=list)
+
+
 class RoomWorkerStore(CacheInvalidationWorkerStore):
  def __init__(
  self,
@@ -1160,17 +1166,29 @@ async def get_partial_state_servers_at_join(self, room_id: str) -> Sequence[str]
  desc="get_partial_state_servers_at_join",
  )
 
- async def get_partial_state_rooms_and_servers(
+ async def get_partial_state_room_resync_info(
  self,
- ) -> Mapping[str, Collection[str]]:
- """Get all rooms containing events with partial state, and the servers known
- to be in the room.
+ ) -> Mapping[str, PartialStateResyncInfo]:
+ """Get all rooms containing events with partial state, and the information
+ needed to restart a "resync" of those rooms.
 
  Returns:
  A dictionary of rooms with partial state, with room IDs as keys and
  lists of servers in rooms as values.
  """
- room_servers: Dict[str, List[str]] = {}
+ room_servers: Dict[str, PartialStateResyncInfo] = {}
+
+ rows = await self.db_pool.simple_select_list(
+ table="partial_state_rooms",
+ keyvalues={},
+ retcols=("room_id", "joined_via"),
+ desc="get_server_which_served_partial_join",
+ )
+
+ for row in rows:
+ room_id = row["room_id"]
+ joined_via = row["joined_via"]
+ room_servers[room_id] = PartialStateResyncInfo(joined_via=joined_via)
 
  rows = await self.db_pool.simple_select_list(
  "partial_state_rooms_servers",
@@ -1182,7 +1200,12 @@ async def get_partial_state_rooms_and_servers(
  for row in rows:
  room_id = row["room_id"]
  server_name = row["server_name"]
- room_servers.setdefault(room_id, []).append(server_name)
+ # There is a foreign key constraint which enforces that every room_id in
+ # partial_state_rooms_servers appears in partial_state_rooms, so this lookup
+ # should always succeed.
+ # TODO(faster joins): To make this robust we should run both SELECTs in the
+ # same transaction.
+ room_servers[room_id].servers_in_room.append(server_name)
 
  return room_servers