diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml
index 20afe311fefb0a5ece7a87345e62c5cace0971be..cdd881fbe19d51b8fa65f84c270339b6b7bb45aa 100644
--- a/.github/workflows/tests.yml
+++ b/.github/workflows/tests.yml
@@ -479,6 +479,9 @@ jobs:
volumes:
- ${{ github.workspace }}:/src
env:
+ # If this is a pull request to a release branch, use that branch as default branch for sytest, else use develop
+ # This works because the release script always create a branch on the sytest repo with the same name as the release branch
+ SYTEST_DEFAULT_BRANCH: ${{ startsWith(github.base_ref, 'release-') && github.base_ref || 'develop' }}
SYTEST_BRANCH: ${{ github.head_ref }}
POSTGRES: ${{ matrix.job.postgres && 1}}
MULTI_POSTGRES: ${{ (matrix.job.postgres == 'multi-postgres') || '' }}
diff --git a/CHANGES.md b/CHANGES.md
index 61c6170c62a1756d300d16f8ea5395d24c209983..94ee661151029f1f2555d7f5622be2e03aa8d2d3 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -1,3 +1,16 @@
+# Synapse 1.109.0rc3 (2024-06-17)
+
+### Bugfixes
+
+- When rolling back to a previous Synapse version and then forwards again to this release, don't require server operators to manually run SQL. ([\#17305](https://github.com/element-hq/synapse/issues/17305), [\#17309](https://github.com/element-hq/synapse/issues/17309))
+
+### Internal Changes
+
+- Use the release branch for sytest in release-branch PRs. ([\#17306](https://github.com/element-hq/synapse/issues/17306))
+
+
+
+
# Synapse 1.109.0rc2 (2024-06-11)
### Bugfixes
diff --git a/debian/changelog b/debian/changelog
index ac2536749da20bcbb4cf57feac54ab82736c2c45..e5b7809bcf9f4c0492dc22c0c812c585f347bdd5 100644
--- a/debian/changelog
+++ b/debian/changelog
@@ -1,3 +1,9 @@
+matrix-synapse-py3 (1.109.0~rc3) stable; urgency=medium
+
+ * New synapse release 1.109.0rc3.
+
+ -- Synapse Packaging team <packages@matrix.org> Mon, 17 Jun 2024 12:05:24 +0000
+
matrix-synapse-py3 (1.109.0~rc2) stable; urgency=medium
* New synapse release 1.109.0rc2.
diff --git a/docs/postgres.md b/docs/postgres.md
index 4b2ba38275be94d6eb5980e57e0318fab70d2954..d06f0cda101094a48539850b8b3860f65afcf724 100644
--- a/docs/postgres.md
+++ b/docs/postgres.md
@@ -255,13 +255,3 @@ however extreme care must be taken to avoid database corruption.
Note that the above may fail with an error about duplicate rows if corruption
has already occurred, and such duplicate rows will need to be manually removed.
-
-### Fixing inconsistent sequences error
-
-Synapse uses Postgres sequences to generate IDs for various tables. A sequence
-and associated table can get out of sync if, for example, Synapse has been
-downgraded and then upgraded again.
-
-To fix the issue shut down Synapse (including any and all workers) and run the
-SQL command included in the error message. Once done Synapse should start
-successfully.
diff --git a/pyproject.toml b/pyproject.toml
index f4f7f7060357fa757856caa7cd9d4e8d95a124b2..7567b1de40bfa44f340849062da62dd648efb90a 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -96,7 +96,7 @@ module-name = "synapse.synapse_rust"
[tool.poetry]
name = "matrix-synapse"
-version = "1.109.0rc2"
+version = "1.109.0rc3"
description = "Homeserver for the Matrix decentralised comms protocol"
authors = ["Matrix.org Team and Contributors <packages@matrix.org>"]
license = "AGPL-3.0-or-later"
diff --git a/synapse/storage/util/id_generators.py b/synapse/storage/util/id_generators.py
index 59c8e05c39497e89f403487ac59c629f20755153..48f88a6f8a68a5140166af996c67b6133aaa0bf6 100644
--- a/synapse/storage/util/id_generators.py
+++ b/synapse/storage/util/id_generators.py
@@ -276,9 +276,6 @@ class MultiWriterIdGenerator(AbstractStreamIdGenerator):
# no active writes in progress.
self._max_position_of_local_instance = self._max_seen_allocated_stream_id
- # This goes and fills out the above state from the database.
- self._load_current_ids(db_conn, tables)
-
self._sequence_gen = build_sequence_generator(
db_conn=db_conn,
database_engine=db.engine,
@@ -303,6 +300,13 @@ class MultiWriterIdGenerator(AbstractStreamIdGenerator):
positive=positive,
)
+ # This goes and fills out the above state from the database.
+ # This may read on the PostgreSQL sequence, and
+ # SequenceGenerator.check_consistency might have fixed up the sequence, which
+ # means the SequenceGenerator needs to be setup before we read the value from
+ # the sequence.
+ self._load_current_ids(db_conn, tables, sequence_name)
+
self._max_seen_allocated_stream_id = max(
self._current_positions.values(), default=1
)
@@ -327,6 +331,7 @@ class MultiWriterIdGenerator(AbstractStreamIdGenerator):
self,
db_conn: LoggingDatabaseConnection,
tables: List[Tuple[str, str, str]],
+ sequence_name: str,
) -> None:
cur = db_conn.cursor(txn_name="_load_current_ids")
@@ -360,6 +365,18 @@ class MultiWriterIdGenerator(AbstractStreamIdGenerator):
if instance in self._writers
}
+ # If we're a writer, we can assume we're at the end of the stream
+ # Usually, we would get that from the stream_positions, but in some cases,
+ # like if we rolled back Synapse, the stream_positions table might not be up to
+ # date. If we're using Postgres for the sequences, we can just use the current
+ # sequence value as our own position.
+ if self._instance_name in self._writers:
+ if isinstance(self._db.engine, PostgresEngine):
+ cur.execute(f"SELECT last_value FROM {sequence_name}")
+ row = cur.fetchone()
+ assert row is not None
+ self._current_positions[self._instance_name] = row[0]
+
# We set the `_persisted_upto_position` to be the minimum of all current
# positions. If empty we use the max stream ID from the DB table.
min_stream_id = min(self._current_positions.values(), default=None)
diff --git a/synapse/storage/util/sequence.py b/synapse/storage/util/sequence.py
index f57e7ec41cd83ecd7b9cbb541fb9d8c31b2b16f9..c4c0602b28e277e21f19c9bd104fe5ccd5ad5d04 100644
--- a/synapse/storage/util/sequence.py
+++ b/synapse/storage/util/sequence.py
@@ -36,21 +36,6 @@ if TYPE_CHECKING:
logger = logging.getLogger(__name__)
-_INCONSISTENT_SEQUENCE_ERROR = """
-Postgres sequence '%(seq)s' is inconsistent with associated
-table '%(table)s'. This can happen if Synapse has been downgraded and
-then upgraded again, or due to a bad migration.
-
-To fix this error, shut down Synapse (including any and all workers)
-and run the following SQL:
-
- SELECT setval('%(seq)s', (
- %(max_id_sql)s
- ));
-
-See docs/postgres.md for more information.
-"""
-
_INCONSISTENT_STREAM_ERROR = """
Postgres sequence '%(seq)s' is inconsistent with associated stream position
of '%(stream_name)s' in the 'stream_positions' table.
@@ -169,25 +154,33 @@ class PostgresSequenceGenerator(SequenceGenerator):
if row:
max_in_stream_positions = row[0]
- txn.close()
-
# If `is_called` is False then `last_value` is actually the value that
# will be generated next, so we decrement to get the true "last value".
if not is_called:
last_value -= 1
if max_stream_id > last_value:
+ # The sequence is lagging behind the tables. This is probably due to
+ # rolling back to a version before the sequence was used and then
+ # forwards again. We resolve this by setting the sequence to the
+ # right value.
logger.warning(
- "Postgres sequence %s is behind table %s: %d < %d",
+ "Postgres sequence %s is behind table %s: %d < %d. Updating sequence.",
self._sequence_name,
table,
last_value,
max_stream_id,
)
- raise IncorrectDatabaseSetup(
- _INCONSISTENT_SEQUENCE_ERROR
- % {"seq": self._sequence_name, "table": table, "max_id_sql": table_sql}
- )
+
+ sql = f"""
+ SELECT setval('{self._sequence_name}', GREATEST(
+ (SELECT last_value FROM {self._sequence_name}),
+ ({table_sql})
+ ));
+ """
+ txn.execute(sql)
+
+ txn.close()
# If we have values in the stream positions table then they have to be
# less than or equal to `last_value`
diff --git a/tests/storage/test_id_generators.py b/tests/storage/test_id_generators.py
index f0307252f3c6c673a0519f455eded6f9aa46c7f3..12b89cecb6e8d107e272d152523dfde7d5716abc 100644
--- a/tests/storage/test_id_generators.py
+++ b/tests/storage/test_id_generators.py
@@ -18,7 +18,7 @@
# [This file includes modifications made by New Vector Limited]
#
#
-from typing import List, Optional
+from typing import Dict, List, Optional
from twisted.test.proto_helpers import MemoryReactor
@@ -28,7 +28,6 @@ from synapse.storage.database import (
LoggingDatabaseConnection,
LoggingTransaction,
)
-from synapse.storage.engines import IncorrectDatabaseSetup
from synapse.storage.types import Cursor
from synapse.storage.util.id_generators import MultiWriterIdGenerator
from synapse.storage.util.sequence import (
@@ -43,9 +42,13 @@ from tests.utils import USE_POSTGRES_FOR_TESTS
class MultiWriterIdGeneratorBase(HomeserverTestCase):
+ positive: bool = True
+ tables: List[str] = ["foobar"]
+
def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
self.store = hs.get_datastores().main
self.db_pool: DatabasePool = self.store.db_pool
+ self.instances: Dict[str, MultiWriterIdGenerator] = {}
self.get_success(self.db_pool.runInteraction("_setup_db", self._setup_db))
@@ -58,18 +61,22 @@ class MultiWriterIdGeneratorBase(HomeserverTestCase):
if USE_POSTGRES_FOR_TESTS:
txn.execute("CREATE SEQUENCE foobar_seq")
- txn.execute(
- """
- CREATE TABLE foobar (
- stream_id BIGINT NOT NULL,
- instance_name TEXT NOT NULL,
- data TEXT
- );
- """
- )
+ for table in self.tables:
+ txn.execute(
+ """
+ CREATE TABLE %s (
+ stream_id BIGINT NOT NULL,
+ instance_name TEXT NOT NULL,
+ data TEXT
+ );
+ """
+ % (table,)
+ )
def _create_id_generator(
- self, instance_name: str = "master", writers: Optional[List[str]] = None
+ self,
+ instance_name: str = "master",
+ writers: Optional[List[str]] = None,
) -> MultiWriterIdGenerator:
def _create(conn: LoggingDatabaseConnection) -> MultiWriterIdGenerator:
return MultiWriterIdGenerator(
@@ -78,36 +85,93 @@ class MultiWriterIdGeneratorBase(HomeserverTestCase):
notifier=self.hs.get_replication_notifier(),
stream_name="test_stream",
instance_name=instance_name,
- tables=[("foobar", "instance_name", "stream_id")],
+ tables=[(table, "instance_name", "stream_id") for table in self.tables],
sequence_name="foobar_seq",
writers=writers or ["master"],
+ positive=self.positive,
+ )
+
+ self.instances[instance_name] = self.get_success_or_raise(
+ self.db_pool.runWithConnection(_create)
+ )
+ return self.instances[instance_name]
+
+ def _replicate(self, instance_name: str) -> None:
+ """Similate a replication event for the given instance."""
+
+ writer = self.instances[instance_name]
+ token = writer.get_current_token_for_writer(instance_name)
+ for generator in self.instances.values():
+ if writer != generator:
+ generator.advance(instance_name, token)
+
+ def _replicate_all(self) -> None:
+ """Similate a replication event for all instances."""
+
+ for instance_name in self.instances:
+ self._replicate(instance_name)
+
+ def _insert_row(
+ self, instance_name: str, stream_id: int, table: Optional[str] = None
+ ) -> None:
+ """Insert one row as the given instance with given stream_id."""
+
+ if table is None:
+ table = self.tables[0]
+
+ factor = 1 if self.positive else -1
+
+ def _insert(txn: LoggingTransaction) -> None:
+ txn.execute(
+ "INSERT INTO %s VALUES (?, ?)" % (table,),
+ (
+ stream_id,
+ instance_name,
+ ),
+ )
+ txn.execute(
+ """
+ INSERT INTO stream_positions VALUES ('test_stream', ?, ?)
+ ON CONFLICT (stream_name, instance_name) DO UPDATE SET stream_id = ?
+ """,
+ (instance_name, stream_id * factor, stream_id * factor),
)
- return self.get_success_or_raise(self.db_pool.runWithConnection(_create))
+ self.get_success(self.db_pool.runInteraction("_insert_row", _insert))
- def _insert_rows(self, instance_name: str, number: int) -> None:
+ def _insert_rows(
+ self,
+ instance_name: str,
+ number: int,
+ table: Optional[str] = None,
+ update_stream_table: bool = True,
+ ) -> None:
"""Insert N rows as the given instance, inserting with stream IDs pulled
from the postgres sequence.
"""
+ if table is None:
+ table = self.tables[0]
+
+ factor = 1 if self.positive else -1
+
def _insert(txn: LoggingTransaction) -> None:
for _ in range(number):
next_val = self.seq_gen.get_next_id_txn(txn)
txn.execute(
- "INSERT INTO foobar (stream_id, instance_name) VALUES (?, ?)",
- (
- next_val,
- instance_name,
- ),
+ "INSERT INTO %s (stream_id, instance_name) VALUES (?, ?)"
+ % (table,),
+ (next_val, instance_name),
)
- txn.execute(
- """
- INSERT INTO stream_positions VALUES ('test_stream', ?, ?)
- ON CONFLICT (stream_name, instance_name) DO UPDATE SET stream_id = ?
- """,
- (instance_name, next_val, next_val),
- )
+ if update_stream_table:
+ txn.execute(
+ """
+ INSERT INTO stream_positions VALUES ('test_stream', ?, ?)
+ ON CONFLICT (stream_name, instance_name) DO UPDATE SET stream_id = ?
+ """,
+ (instance_name, next_val * factor, next_val * factor),
+ )
self.get_success(self.db_pool.runInteraction("_insert_rows", _insert))
@@ -354,7 +418,9 @@ class WorkerMultiWriterIdGeneratorTestCase(MultiWriterIdGeneratorBase):
id_gen = self._create_id_generator("first", writers=["first", "second"])
- self.assertEqual(id_gen.get_positions(), {"first": 3, "second": 5})
+ # When the writer is created, it assumes its own position is the current head of
+ # the sequence
+ self.assertEqual(id_gen.get_positions(), {"first": 5, "second": 5})
self.assertEqual(id_gen.get_persisted_upto_position(), 5)
@@ -376,11 +442,13 @@ class WorkerMultiWriterIdGeneratorTestCase(MultiWriterIdGeneratorBase):
correctly.
"""
self._insert_rows("first", 3)
- self._insert_rows("second", 4)
-
first_id_gen = self._create_id_generator("first", writers=["first", "second"])
+
+ self._insert_rows("second", 4)
second_id_gen = self._create_id_generator("second", writers=["first", "second"])
+ self._replicate_all()
+
self.assertEqual(first_id_gen.get_positions(), {"first": 3, "second": 7})
self.assertEqual(first_id_gen.get_current_token_for_writer("first"), 7)
self.assertEqual(first_id_gen.get_current_token_for_writer("second"), 7)
@@ -399,6 +467,9 @@ class WorkerMultiWriterIdGeneratorTestCase(MultiWriterIdGeneratorBase):
self.assertEqual(
first_id_gen.get_positions(), {"first": 3, "second": 7}
)
+ self.assertEqual(
+ second_id_gen.get_positions(), {"first": 3, "second": 7}
+ )
self.assertEqual(first_id_gen.get_persisted_upto_position(), 7)
self.get_success(_get_next_async())
@@ -433,11 +504,11 @@ class WorkerMultiWriterIdGeneratorTestCase(MultiWriterIdGeneratorBase):
"""
# Insert some rows for two out of three of the ID gens.
self._insert_rows("first", 3)
- self._insert_rows("second", 4)
-
first_id_gen = self._create_id_generator(
"first", writers=["first", "second", "third"]
)
+
+ self._insert_rows("second", 4)
second_id_gen = self._create_id_generator(
"second", writers=["first", "second", "third"]
)
@@ -445,6 +516,8 @@ class WorkerMultiWriterIdGeneratorTestCase(MultiWriterIdGeneratorBase):
"third", writers=["first", "second", "third"]
)
+ self._replicate_all()
+
self.assertEqual(
first_id_gen.get_positions(), {"first": 3, "second": 7, "third": 7}
)
@@ -525,7 +598,7 @@ class WorkerMultiWriterIdGeneratorTestCase(MultiWriterIdGeneratorBase):
self.assertEqual(id_gen_5.get_current_token_for_writer("third"), 6)
def test_sequence_consistency(self) -> None:
- """Test that we error out if the table and sequence diverges."""
+ """Test that we correct the sequence if the table and sequence diverges."""
# Prefill with some rows
self._insert_row_with_id("master", 3)
@@ -536,17 +609,24 @@ class WorkerMultiWriterIdGeneratorTestCase(MultiWriterIdGeneratorBase):
self.get_success(self.db_pool.runInteraction("_insert", _insert))
- # Creating the ID gen should error
- with self.assertRaises(IncorrectDatabaseSetup):
- self._create_id_generator("first")
+ # Creating the ID gen should now fix the inconsistency
+ id_gen = self._create_id_generator()
+
+ async def _get_next_async() -> None:
+ async with id_gen.get_next() as stream_id:
+ self.assertEqual(stream_id, 27)
+
+ self.get_success(_get_next_async())
def test_minimal_local_token(self) -> None:
self._insert_rows("first", 3)
- self._insert_rows("second", 4)
-
first_id_gen = self._create_id_generator("first", writers=["first", "second"])
+
+ self._insert_rows("second", 4)
second_id_gen = self._create_id_generator("second", writers=["first", "second"])
+ self._replicate_all()
+
self.assertEqual(first_id_gen.get_positions(), {"first": 3, "second": 7})
self.assertEqual(first_id_gen.get_minimal_local_current_token(), 3)
@@ -558,15 +638,17 @@ class WorkerMultiWriterIdGeneratorTestCase(MultiWriterIdGeneratorBase):
token when there are no writes.
"""
self._insert_rows("first", 3)
- self._insert_rows("second", 4)
-
first_id_gen = self._create_id_generator(
"first", writers=["first", "second", "third"]
)
+
+ self._insert_rows("second", 4)
second_id_gen = self._create_id_generator(
"second", writers=["first", "second", "third"]
)
+ self._replicate_all()
+
self.assertEqual(second_id_gen.get_current_token_for_writer("first"), 7)
self.assertEqual(second_id_gen.get_current_token_for_writer("second"), 7)
self.assertEqual(second_id_gen.get_current_token(), 7)
@@ -605,68 +687,13 @@ class WorkerMultiWriterIdGeneratorTestCase(MultiWriterIdGeneratorBase):
self.assertEqual(second_id_gen.get_current_token(), 7)
-class BackwardsMultiWriterIdGeneratorTestCase(HomeserverTestCase):
+class BackwardsMultiWriterIdGeneratorTestCase(MultiWriterIdGeneratorBase):
"""Tests MultiWriterIdGenerator that produce *negative* stream IDs."""
if not USE_POSTGRES_FOR_TESTS:
skip = "Requires Postgres"
- def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
- self.store = hs.get_datastores().main
- self.db_pool: DatabasePool = self.store.db_pool
-
- self.get_success(self.db_pool.runInteraction("_setup_db", self._setup_db))
-
- def _setup_db(self, txn: LoggingTransaction) -> None:
- txn.execute("CREATE SEQUENCE foobar_seq")
- txn.execute(
- """
- CREATE TABLE foobar (
- stream_id BIGINT NOT NULL,
- instance_name TEXT NOT NULL,
- data TEXT
- );
- """
- )
-
- def _create_id_generator(
- self, instance_name: str = "master", writers: Optional[List[str]] = None
- ) -> MultiWriterIdGenerator:
- def _create(conn: LoggingDatabaseConnection) -> MultiWriterIdGenerator:
- return MultiWriterIdGenerator(
- conn,
- self.db_pool,
- notifier=self.hs.get_replication_notifier(),
- stream_name="test_stream",
- instance_name=instance_name,
- tables=[("foobar", "instance_name", "stream_id")],
- sequence_name="foobar_seq",
- writers=writers or ["master"],
- positive=False,
- )
-
- return self.get_success(self.db_pool.runWithConnection(_create))
-
- def _insert_row(self, instance_name: str, stream_id: int) -> None:
- """Insert one row as the given instance with given stream_id."""
-
- def _insert(txn: LoggingTransaction) -> None:
- txn.execute(
- "INSERT INTO foobar VALUES (?, ?)",
- (
- stream_id,
- instance_name,
- ),
- )
- txn.execute(
- """
- INSERT INTO stream_positions VALUES ('test_stream', ?, ?)
- ON CONFLICT (stream_name, instance_name) DO UPDATE SET stream_id = ?
- """,
- (instance_name, -stream_id, -stream_id),
- )
-
- self.get_success(self.db_pool.runInteraction("_insert_row", _insert))
+ positive = False
def test_single_instance(self) -> None:
"""Test that reads and writes from a single process are handled
@@ -712,7 +739,7 @@ class BackwardsMultiWriterIdGeneratorTestCase(HomeserverTestCase):
async def _get_next_async() -> None:
async with id_gen_1.get_next() as stream_id:
self._insert_row("first", stream_id)
- id_gen_2.advance("first", stream_id)
+ self._replicate("first")
self.get_success(_get_next_async())
@@ -724,7 +751,7 @@ class BackwardsMultiWriterIdGeneratorTestCase(HomeserverTestCase):
async def _get_next_async2() -> None:
async with id_gen_2.get_next() as stream_id:
self._insert_row("second", stream_id)
- id_gen_1.advance("second", stream_id)
+ self._replicate("second")
self.get_success(_get_next_async2())
@@ -734,98 +761,26 @@ class BackwardsMultiWriterIdGeneratorTestCase(HomeserverTestCase):
self.assertEqual(id_gen_2.get_persisted_upto_position(), -2)
-class MultiTableMultiWriterIdGeneratorTestCase(HomeserverTestCase):
+class MultiTableMultiWriterIdGeneratorTestCase(MultiWriterIdGeneratorBase):
if not USE_POSTGRES_FOR_TESTS:
skip = "Requires Postgres"
- def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
- self.store = hs.get_datastores().main
- self.db_pool: DatabasePool = self.store.db_pool
-
- self.get_success(self.db_pool.runInteraction("_setup_db", self._setup_db))
-
- def _setup_db(self, txn: LoggingTransaction) -> None:
- txn.execute("CREATE SEQUENCE foobar_seq")
- txn.execute(
- """
- CREATE TABLE foobar1 (
- stream_id BIGINT NOT NULL,
- instance_name TEXT NOT NULL,
- data TEXT
- );
- """
- )
-
- txn.execute(
- """
- CREATE TABLE foobar2 (
- stream_id BIGINT NOT NULL,
- instance_name TEXT NOT NULL,
- data TEXT
- );
- """
- )
-
- def _create_id_generator(
- self, instance_name: str = "master", writers: Optional[List[str]] = None
- ) -> MultiWriterIdGenerator:
- def _create(conn: LoggingDatabaseConnection) -> MultiWriterIdGenerator:
- return MultiWriterIdGenerator(
- conn,
- self.db_pool,
- notifier=self.hs.get_replication_notifier(),
- stream_name="test_stream",
- instance_name=instance_name,
- tables=[
- ("foobar1", "instance_name", "stream_id"),
- ("foobar2", "instance_name", "stream_id"),
- ],
- sequence_name="foobar_seq",
- writers=writers or ["master"],
- )
-
- return self.get_success_or_raise(self.db_pool.runWithConnection(_create))
-
- def _insert_rows(
- self,
- table: str,
- instance_name: str,
- number: int,
- update_stream_table: bool = True,
- ) -> None:
- """Insert N rows as the given instance, inserting with stream IDs pulled
- from the postgres sequence.
- """
-
- def _insert(txn: LoggingTransaction) -> None:
- for _ in range(number):
- txn.execute(
- "INSERT INTO %s VALUES (nextval('foobar_seq'), ?)" % (table,),
- (instance_name,),
- )
- if update_stream_table:
- txn.execute(
- """
- INSERT INTO stream_positions VALUES ('test_stream', ?, lastval())
- ON CONFLICT (stream_name, instance_name) DO UPDATE SET stream_id = lastval()
- """,
- (instance_name,),
- )
-
- self.get_success(self.db_pool.runInteraction("_insert_rows", _insert))
+ tables = ["foobar1", "foobar2"]
def test_load_existing_stream(self) -> None:
"""Test creating ID gens with multiple tables that have rows from after
the position in `stream_positions` table.
"""
- self._insert_rows("foobar1", "first", 3)
- self._insert_rows("foobar2", "second", 3)
- self._insert_rows("foobar2", "second", 1, update_stream_table=False)
-
+ self._insert_rows("first", 3, table="foobar1")
first_id_gen = self._create_id_generator("first", writers=["first", "second"])
+
+ self._insert_rows("second", 3, table="foobar2")
+ self._insert_rows("second", 1, table="foobar2", update_stream_table=False)
second_id_gen = self._create_id_generator("second", writers=["first", "second"])
- self.assertEqual(first_id_gen.get_positions(), {"first": 3, "second": 6})
+ self._replicate_all()
+
+ self.assertEqual(first_id_gen.get_positions(), {"first": 3, "second": 7})
self.assertEqual(first_id_gen.get_current_token_for_writer("first"), 7)
self.assertEqual(first_id_gen.get_current_token_for_writer("second"), 7)
self.assertEqual(first_id_gen.get_persisted_upto_position(), 7)