Fix storage type mismatches where values didn't match their column types (#19911)

A handful of places in the storage layer bound a value whose Python type
didn't match the declared column type — an `int` into a `TEXT` column,
or a `str` into a `BIGINT` column — and relied on psycopg2's loose
coercion to paper over the mismatch. These are latent correctness bugs:
they only work because the driver silently converts, and a stricter
driver that binds typed parameters rejects them outright.

Found during a Rust port of the database pool where the driver does not
coerce automatically.

Each fix binds the value with the type the column actually declares,
rather than depending on driver-specific coercion. All changes are
behaviour-preserving on psycopg2 and sqlite.

There is also a fix to the multi-writer id-gen tests where we forgot to
commit. This is tangential, but was found during the same effort.

  ### Changes

- **`device_lists_remote_extremeties.stream_id (TEXT)`** —
`_update_remote_device_list_cache_txn` (typed `stream_id: int`) bound an
int; store it as a string, matching the column and the sibling
`_update_remote_device_list_cache_entry_txn` (typed `stream_id: str`).
The old mismatch, when rejected, was swallowed inside the device-list
resync and hung `query_devices`.
- **`user_filters.filter_id (BIGINT)`** — `get_user_filter` bound the
raw `int | str` (a string, from sync requests). It already validates via
`int(filter_id)`; bind that int so it matches the column.
- **`rejections.last_check (TEXT)`** — both writers stored
`clock.time_msec()` (an int); store the timestamp as a string.
- **user-directory temp position** — the `populate_user_directory`
background update's staging column
`_temp_populate_user_directory_position.position` was `TEXT` but held an
int (read back into a `BIGINT` column). Declare it `BIGINT`. The temp
table is created and dropped within the background update, so there's no
migration.

- **`test_batched_state_group_storing`** — selected from
`state_group_edges` with a stringified `state_group`; bind the int
directly (the column is an integer).
- **Multi-writer id-generator tests** — constructing a
`MultiWriterIdGenerator` prunes stale `stream_positions` rows, but the
harness never committed, so the cleanup only survived because adbapi
keeps one connection per thread with its transaction open. Commit after
construction so it persists regardless of pool semantics (the delete is
legitimate work that should be committed anyway).

changelog.d/19911.misc

@@ -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 @@ class DeviceWorkerStore(RoomMemberWorkerStore, EndToEndKeyWorkerStore):
             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 @@ class PersistEventsStore:
             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 @@ class EventsWorkerStore(SQLBaseStore):
                 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 @@ class FilteringWorkerStore(SQLBaseStore):
         # 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 @@ class UserDirectoryBackgroundUpdateStore(StateDeltasStore):
 
             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 @@ class MultiWriterIdGeneratorBase(HomeserverTestCase):
         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 @@ class MultiWriterIdGeneratorBase(HomeserverTestCase):
                 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 @@ class StateStoreTestCase(HomeserverTestCase):
                     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",),
                         )
                     ),