From e4fe239a9dcef39325d52510bf90935d3ec32dd5 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 12 Jul 2023 09:43:40 +0200 Subject: [PATCH] A few fixes in synchronous_mode (#2741) - make sure that physical replication slots are created even before the promote happened (when async executor is busy with promote). - execute `txid_current()` with `synchronous_commit=off` so it doesn't accidentally wait for absent synchronous standbys when `synchronous_mode_strict` is enable and `synchronous_standby_names=*`. These standbys can't connect because replication slots weren't there. - `synchronous_standby_names` wasn't set to `*` after bootstrap with `synchronous_mode` and `synchronous_mode_strict`. - add `-c statement_timeout=0` to `PGOPTIONS` when executing `post_bootstrap` script. Close https://github.com/zalando/patroni/issues/2738 --- patroni/ha.py | 20 ++++++++++++++------ patroni/postgresql/bootstrap.py | 2 +- patroni/postgresql/sync.py | 9 +++++++-- tests/test_ha.py | 1 + 4 files changed, 23 insertions(+), 9 deletions(-) diff --git a/patroni/ha.py b/patroni/ha.py index a85dd690..c55b327b 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -1541,6 +1541,9 @@ class Ha(object): self.dcs.set_config_value(json.dumps(self.patroni.config.dynamic_configuration, separators=(',', ':'))) self.dcs.take_leader() self.set_is_leader(True) + if self.is_synchronous_mode(): + self.state_handler.sync_handler.set_synchronous_standby_names( + CaseInsensitiveSet('*') if self.global_config.is_synchronous_mode_strict else CaseInsensitiveSet()) self.state_handler.call_nowait(CallbackAction.ON_START) self.load_cluster_from_dcs() @@ -1734,16 +1737,21 @@ class Ha(object): msg = self.process_healthy_cluster() ret = self.evaluate_scheduled_restart() or msg - # we might not have a valid PostgreSQL connection here if another thread - # stops PostgreSQL, therefore, we only reload replication slots if no - # asynchronous processes are running (should be always the case for the primary) - if not self._async_executor.busy and not self.state_handler.is_starting(): + # We might not have a valid PostgreSQL connection here if AsyncExecutor is doing + # something with PostgreSQL. Therefore we will sync replication slots only if no + # asynchronous processes are running or we know that this is a standby being promoted. + # But, we don't want to run pg_rewind checks or copy logical slots from itself, + # therefore we have a couple additional `not is_promoting` checks. + is_promoting = self._async_executor.scheduled_action == 'promote' + if (not self._async_executor.busy or is_promoting) and not self.state_handler.is_starting(): create_slots = self._sync_replication_slots(False) + if not self.state_handler.cb_called: - if not self.state_handler.is_leader(): + if not is_promoting and not self.state_handler.is_leader(): self._rewind.trigger_check_diverged_lsn() self.state_handler.call_nowait(CallbackAction.ON_START) - if create_slots and self.cluster.leader: + + if not is_promoting and create_slots and self.cluster.leader: err = self._async_executor.try_run_async('copy_logical_slots', self.state_handler.slots_handler.copy_logical_slots, args=(self.cluster, create_slots)) diff --git a/patroni/postgresql/bootstrap.py b/patroni/postgresql/bootstrap.py index a76f9f5f..6e25012b 100644 --- a/patroni/postgresql/bootstrap.py +++ b/patroni/postgresql/bootstrap.py @@ -185,7 +185,7 @@ class Bootstrap(object): r['host'] = 'localhost' # set it to localhost to write into pgpass env = self._postgresql.config.write_pgpass(r) - env['PGOPTIONS'] = '-c synchronous_commit=local' + env['PGOPTIONS'] = '-c synchronous_commit=local -c statement_timeout=0' try: ret = self._postgresql.cancellable.call(shlex.split(cmd) + [connstring], env=env) diff --git a/patroni/postgresql/sync.py b/patroni/postgresql/sync.py index c56bdbcd..d0f28586 100644 --- a/patroni/postgresql/sync.py +++ b/patroni/postgresql/sync.py @@ -193,7 +193,12 @@ class SyncHandler(object): # Newly connected replicas will be counted as sync only when reached self._primary_flush_lsn self._primary_flush_lsn = self._postgresql.last_operation() - self._postgresql.query('SELECT pg_catalog.txid_current()') # Ensure some WAL traffic to move replication + # Ensure some WAL traffic to move replication + self._postgresql.query("""DO $$ +BEGIN + SET local synchronous_commit = 'off'; + PERFORM * FROM pg_catalog.txid_current(); +END;$$""") self._postgresql.reset_cluster_info_state(None) # Reset internal cache to query fresh values def current_state(self, cluster: Cluster) -> Tuple[CaseInsensitiveSet, CaseInsensitiveSet]: @@ -289,6 +294,6 @@ class SyncHandler(object): # Reset internal cache to query fresh values self._postgresql.reset_cluster_info_state(None) - # timeline == 0 -- indicates that this is the replica, shoudn't ever happen + # timeline == 0 -- indicates that this is the replica if self._postgresql.get_primary_timeline() > 0: self._handle_synchronous_standby_names_change() diff --git a/tests/test_ha.py b/tests/test_ha.py index cfcb51c4..a621b03b 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -583,6 +583,7 @@ class TestHa(PostgresInit): self.p.is_leader = false self.assertEqual(self.ha.run_cycle(), 'waiting for end of recovery after bootstrap') self.p.is_leader = true + self.ha.is_synchronous_mode = true self.assertEqual(self.ha.run_cycle(), 'running post_bootstrap') self.assertEqual(self.ha.run_cycle(), 'initialized a new cluster')