Fix storage type mismatches where values didn't match their column types

changelog.d/19911.bugfix

@@ -0,0 +1 @@
+Fix storage type mismatches where values were bound with a type that didn't match their database column.

synapse/storage/databases/main/devices.py

@@ -2019,7 +2019,10 @@ def _update_remote_device_list_cache_txn(
             txn,
             table="device_lists_remote_extremeties",
             keyvalues={"user_id": user_id},
-            values={"stream_id": stream_id},
+            # `stream_id` is a TEXT column, so store it as a string (this method
+            # takes an int) rather than relying on the driver to coerce it.
+            # (Ideally we'd fix the schema, but that is non-trivial)
+            values={"stream_id": str(stream_id)},
         )
 
     async def add_device_change_to_streams(

synapse/storage/databases/main/events.py

@@ -3513,7 +3513,10 @@ def _store_rejections_txn(
             values={
                 "event_id": event_id,
                 "reason": reason,
-                "last_check": self._clock.time_msec(),
+                # `last_check` is a TEXT column, so store the timestamp as a
+                # string rather than relying on the driver to coerce an int.
+                # (Ideally we'd fix the schema, but that is non-trivial)
+                "last_check": str(self._clock.time_msec()),
             },
         )
 

synapse/storage/databases/main/events_worker.py

@@ -2702,7 +2702,10 @@ def mark_event_rejected_txn(
                 keyvalues={"event_id": event_id},
                 values={
                     "reason": rejection_reason,
-                    "last_check": self.clock.time_msec(),
+                    # `last_check` is a TEXT column, so store the timestamp as a
+                    # string rather than relying on the driver to coerce an int.
+                    # (Ideally we'd fix the schema, but that is non-trivial)
+                    "last_check": str(self.clock.time_msec()),
                 },
             )
         self.db_pool.simple_update_txn(

synapse/storage/databases/main/filtering.py

@@ -156,7 +156,7 @@ async def get_user_filter(
         # filter_id is BIGINT UNSIGNED, so if it isn't a number, fail
         # with a coherent error message rather than 500 M_UNKNOWN.
         try:
-            int(filter_id)
+            filter_id = int(filter_id)
         except ValueError:
             raise SynapseError(400, "Invalid filter ID", Codes.INVALID_PARAM)
 

synapse/storage/databases/main/user_directory.py

@@ -127,7 +127,7 @@ def _make_staging_area(txn: LoggingTransaction) -> None:
 
             sql = f"""
                 CREATE TABLE IF NOT EXISTS {TEMP_TABLE}_position (
-                    position TEXT NOT NULL
+                    position BIGINT NOT NULL
                 )
             """
             txn.execute(sql)

tests/storage/test_id_generators.py

@@ -78,7 +78,7 @@ def _create_id_generator(
         writers: list[str] | None = None,
     ) -> MultiWriterIdGenerator:
         def _create(conn: LoggingDatabaseConnection) -> MultiWriterIdGenerator:
-            return MultiWriterIdGenerator(
+            id_gen = MultiWriterIdGenerator(
                 db_conn=conn,
                 db=self.db_pool,
                 notifier=self.hs.get_replication_notifier(),
@@ -90,6 +90,15 @@ def _create(conn: LoggingDatabaseConnection) -> MultiWriterIdGenerator:
                 writers=writers or ["master"],
                 positive=self.positive,
             )
+            # Constructing the generator prunes stale `stream_positions` rows
+            # (writers no longer in the config); commit so that persists for the
+            # next generator we create.
+            #
+            # Note we need to commit manually here as the generator is created
+            # in a `runWithConnection` call, which doesn't automatically
+            # commit/rollback.
+            conn.commit()
+            return id_gen
 
         self.instances[instance_name] = self.get_success_or_raise(
             self.db_pool.runWithConnection(_create)

tests/storage/test_state.py

@@ -637,9 +637,7 @@ def test_batched_state_group_storing(self) -> None:
                     self.get_success(
                         self.store.db_pool.simple_select_list(
                             table="state_group_edges",
-                            keyvalues={
-                                "state_group": str(context.state_group_after_event)
-                            },
+                            keyvalues={"state_group": context.state_group_after_event},
                             retcols=("prev_state_group",),
                         )
                     ),