@@ -276,9 +276,6 @@ def __init__(
276
276
# no active writes in progress.
277
277
self ._max_position_of_local_instance = self ._max_seen_allocated_stream_id
278
278
279
- # This goes and fills out the above state from the database.
280
- self ._load_current_ids (db_conn , tables )
281
-
282
279
self ._sequence_gen = build_sequence_generator (
283
280
db_conn = db_conn ,
284
281
database_engine = db .engine ,
@@ -303,6 +300,13 @@ def __init__(
303
300
positive = positive ,
304
301
)
305
302
303
+ # This goes and fills out the above state from the database.
304
+ # This may read on the PostgreSQL sequence, and
305
+ # SequenceGenerator.check_consistency might have fixed up the sequence, which
306
+ # means the SequenceGenerator needs to be setup before we read the value from
307
+ # the sequence.
308
+ self ._load_current_ids (db_conn , tables , sequence_name )
309
+
306
310
self ._max_seen_allocated_stream_id = max (
307
311
self ._current_positions .values (), default = 1
308
312
)
@@ -327,6 +331,7 @@ def _load_current_ids(
327
331
self ,
328
332
db_conn : LoggingDatabaseConnection ,
329
333
tables : List [Tuple [str , str , str ]],
334
+ sequence_name : str ,
330
335
) -> None :
331
336
cur = db_conn .cursor (txn_name = "_load_current_ids" )
332
337
@@ -360,6 +365,18 @@ def _load_current_ids(
360
365
if instance in self ._writers
361
366
}
362
367
368
+ # If we're a writer, we can assume we're at the end of the stream
369
+ # Usually, we would get that from the stream_positions, but in some cases,
370
+ # like if we rolled back Synapse, the stream_positions table might not be up to
371
+ # date. If we're using Postgres for the sequences, we can just use the current
372
+ # sequence value as our own position.
373
+ if self ._instance_name in self ._writers :
374
+ if isinstance (self ._db .engine , PostgresEngine ):
375
+ cur .execute (f"SELECT last_value FROM { sequence_name } " )
376
+ row = cur .fetchone ()
377
+ assert row is not None
378
+ self ._current_positions [self ._instance_name ] = row [0 ]
379
+
363
380
# We set the `_persisted_upto_position` to be the minimum of all current
364
381
# positions. If empty we use the max stream ID from the DB table.
365
382
min_stream_id = min (self ._current_positions .values (), default = None )
0 commit comments