rust/neon
1
0
Fork 0
mirror of https://github.com/neondatabase/neon.git synced 2026-01-17 02:12:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fix LR tests waiting for synced data.

Browse Source 
Even if pg_stat_subscription.latest_end_lsn is caughtup, some tables might not
be synced because until sync worker finishes main apply worker continues to
advance.
This commit is contained in:
Arseny Sher
2024-03-13 13:42:49 +03:00
parent 83855a907c
commit 5598f712fa
1 changed files with 18 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
test_runner/fixtures/neon_fixtures.py
View File

@@ -3983,18 +3983,27 @@ def check_restored_datadir_content(test_output_dir: Path, env: NeonEnv, endpoint
def logical_replication_sync(subscriber: VanillaPostgres, publisher: Endpoint) -> Lsn:
"""Wait logical replication subscriber to sync with publisher."""
publisher_lsn = Lsn(publisher.safe_psql("SELECT pg_current_wal_flush_lsn()")[0][0])
while True:
def is_synced(publisher_lsn):
# Even if pg_stat_subscription.latest_end_lsn is caughtup, some tables
# might not be synced because until sync worker finishes main apply
# continues to advance.
rels_synced = subscriber.safe_psql(
"select count(*) = 0 from pg_subscription_rel where srsubstate != 'r'"
)[0][0]
log.info(f"number of not synced rels: {rels_synced}")
assert rels_synced
res = subscriber.safe_psql("select latest_end_lsn from pg_catalog.pg_stat_subscription")[0][
0
]
if res:
log.info(f"subscriber_lsn={res}")
subscriber_lsn = Lsn(res)
log.info(f"Subscriber LSN={subscriber_lsn}, publisher LSN={ publisher_lsn}")
if subscriber_lsn >= publisher_lsn:
return subscriber_lsn
time.sleep(0.5)
log.info(f"subscriber_lsn={res}")
subscriber_lsn = Lsn(res)
log.info(f"Subscriber LSN={subscriber_lsn}, publisher LSN={ publisher_lsn}")
assert subscriber_lsn >= publisher_lsn
publisher_lsn = Lsn(publisher.safe_psql("SELECT pg_current_wal_flush_lsn()")[0][0])
wait_until(30, 0.5, partial(is_synced, publisher_lsn))
return publisher_lsn
def tenant_get_shards(