From a96759e9c10d04c10e18c382dfc99f89a17fe6b3 Mon Sep 17 00:00:00 2001 From: Andrew Morgan Date: Thu, 14 Nov 2019 14:07:42 +0000 Subject: [PATCH 1/5] upsert on max to_device stream id --- .../storage/data_stores/main/deviceinbox.py | 11 ++++++++-- .../main/schema/delta/56/device_stream_id.sql | 22 +++++++++++++++++++ 2 files changed, 31 insertions(+), 2 deletions(-) create mode 100644 synapse/storage/data_stores/main/schema/delta/56/device_stream_id.sql diff --git a/synapse/storage/data_stores/main/deviceinbox.py b/synapse/storage/data_stores/main/deviceinbox.py index f04aad074339..853224c57092 100644 --- a/synapse/storage/data_stores/main/deviceinbox.py +++ b/synapse/storage/data_stores/main/deviceinbox.py @@ -358,8 +358,15 @@ def add_messages_txn(txn, now_ms, stream_id): def _add_messages_to_local_device_inbox_txn( self, txn, stream_id, messages_by_user_then_device ): - sql = "UPDATE device_max_stream_id" " SET stream_id = ?" " WHERE stream_id < ?" - txn.execute(sql, (stream_id, stream_id)) + # Compatible method of performing an upsert with a dummy table column + sql = """ + INSERT INTO device_max_stream_id + (id, stream_id) VALUES (0, ?) + ON CONFLICT(id) DO UPDATE + SET stream_id = ? + WHERE stream_id < ? + """ + txn.execute(sql, (stream_id, stream_id, stream_id)) local_by_user_then_device = {} for user_id, messages_by_device in messages_by_user_then_device.items(): diff --git a/synapse/storage/data_stores/main/schema/delta/56/device_stream_id.sql b/synapse/storage/data_stores/main/schema/delta/56/device_stream_id.sql new file mode 100644 index 000000000000..d430324663c1 --- /dev/null +++ b/synapse/storage/data_stores/main/schema/delta/56/device_stream_id.sql @@ -0,0 +1,22 @@ +/* Copyright 2019 The Matrix.org Foundation C.I.C. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +-- During migration from separate schema delta files to full schema 54, we ended up with an +-- UPDATE statement that updated a table that never contained a row, thus no data would ever +-- wind up in the table. + +-- Add a new unique, dummy column to the table that we can perform conflict resolution with +ALTER TABLE device_max_stream_id ADD COLUMN id BOOLEAN; +CREATE UNIQUE INDEX device_max_stream_id_id ON device_max_stream_id(id); \ No newline at end of file From 0ef6e52b3d937e4ffa3de322237038511ae15780 Mon Sep 17 00:00:00 2001 From: Andrew Morgan Date: Thu, 14 Nov 2019 14:14:53 +0000 Subject: [PATCH 2/5] Add changelog --- changelog.d/6363.bugfix | 1 + 1 file changed, 1 insertion(+) create mode 100644 changelog.d/6363.bugfix diff --git a/changelog.d/6363.bugfix b/changelog.d/6363.bugfix new file mode 100644 index 000000000000..d023b49181cb --- /dev/null +++ b/changelog.d/6363.bugfix @@ -0,0 +1 @@ +Fix `to_device` stream ID getting reset every time Synapse restarts, which had the potential to cause unable to decrypt errors. \ No newline at end of file From e637b6f27b0617e66d525d07f92ce1347123069c Mon Sep 17 00:00:00 2001 From: Andrew Morgan Date: Thu, 14 Nov 2019 14:50:08 +0000 Subject: [PATCH 3/5] 0 -> false in postgres --- synapse/storage/data_stores/main/deviceinbox.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/synapse/storage/data_stores/main/deviceinbox.py b/synapse/storage/data_stores/main/deviceinbox.py index 853224c57092..2bbc1924c28c 100644 --- a/synapse/storage/data_stores/main/deviceinbox.py +++ b/synapse/storage/data_stores/main/deviceinbox.py @@ -361,7 +361,7 @@ def _add_messages_to_local_device_inbox_txn( # Compatible method of performing an upsert with a dummy table column sql = """ INSERT INTO device_max_stream_id - (id, stream_id) VALUES (0, ?) + (id, stream_id) VALUES (false, ?) ON CONFLICT(id) DO UPDATE SET stream_id = ? WHERE stream_id < ? From 8e6a5f0eedc6861a5b18762b833bdf1bf2a94e97 Mon Sep 17 00:00:00 2001 From: Andrew Morgan Date: Thu, 14 Nov 2019 15:21:31 +0000 Subject: [PATCH 4/5] Add debug logging --- synapse/storage/data_stores/main/deviceinbox.py | 1 + 1 file changed, 1 insertion(+) diff --git a/synapse/storage/data_stores/main/deviceinbox.py b/synapse/storage/data_stores/main/deviceinbox.py index 2bbc1924c28c..3f894aa93098 100644 --- a/synapse/storage/data_stores/main/deviceinbox.py +++ b/synapse/storage/data_stores/main/deviceinbox.py @@ -359,6 +359,7 @@ def _add_messages_to_local_device_inbox_txn( self, txn, stream_id, messages_by_user_then_device ): # Compatible method of performing an upsert with a dummy table column + logger.info("Doing it with stream_id %s", stream_id) sql = """ INSERT INTO device_max_stream_id (id, stream_id) VALUES (false, ?) From 9919c148420b31325f66411537b8d4704dc659a6 Mon Sep 17 00:00:00 2001 From: Andrew Morgan Date: Thu, 14 Nov 2019 16:36:17 +0000 Subject: [PATCH 5/5] UPSERT in python --- .../storage/data_stores/main/deviceinbox.py | 25 +++++++++++-------- .../main/schema/delta/56/device_stream_id.sql | 22 ---------------- 2 files changed, 15 insertions(+), 32 deletions(-) delete mode 100644 synapse/storage/data_stores/main/schema/delta/56/device_stream_id.sql diff --git a/synapse/storage/data_stores/main/deviceinbox.py b/synapse/storage/data_stores/main/deviceinbox.py index 3f894aa93098..96cd0fb77ade 100644 --- a/synapse/storage/data_stores/main/deviceinbox.py +++ b/synapse/storage/data_stores/main/deviceinbox.py @@ -358,16 +358,21 @@ def add_messages_txn(txn, now_ms, stream_id): def _add_messages_to_local_device_inbox_txn( self, txn, stream_id, messages_by_user_then_device ): - # Compatible method of performing an upsert with a dummy table column - logger.info("Doing it with stream_id %s", stream_id) - sql = """ - INSERT INTO device_max_stream_id - (id, stream_id) VALUES (false, ?) - ON CONFLICT(id) DO UPDATE - SET stream_id = ? - WHERE stream_id < ? - """ - txn.execute(sql, (stream_id, stream_id, stream_id)) + # Compatible method of performing an upsert + sql = "SELECT stream_id FROM device_max_stream_id" + + txn.execute(sql) + rows = txn.fetchone() + if rows: + db_stream_id = rows[0] + if db_stream_id < stream_id: + # Insert the new stream_id + sql = "UPDATE device_max_stream_id SET stream_id = ?" + else: + # No rows, perform an insert + sql = "INSERT INTO device_max_stream_id (stream_id) VALUES (?)" + + txn.execute(sql, (stream_id,)) local_by_user_then_device = {} for user_id, messages_by_device in messages_by_user_then_device.items(): diff --git a/synapse/storage/data_stores/main/schema/delta/56/device_stream_id.sql b/synapse/storage/data_stores/main/schema/delta/56/device_stream_id.sql deleted file mode 100644 index d430324663c1..000000000000 --- a/synapse/storage/data_stores/main/schema/delta/56/device_stream_id.sql +++ /dev/null @@ -1,22 +0,0 @@ -/* Copyright 2019 The Matrix.org Foundation C.I.C. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - --- During migration from separate schema delta files to full schema 54, we ended up with an --- UPDATE statement that updated a table that never contained a row, thus no data would ever --- wind up in the table. - --- Add a new unique, dummy column to the table that we can perform conflict resolution with -ALTER TABLE device_max_stream_id ADD COLUMN id BOOLEAN; -CREATE UNIQUE INDEX device_max_stream_id_id ON device_max_stream_id(id); \ No newline at end of file