Release v3.2.2 (#3007 )

- update release notes - bump Patroni version - bump pyright version and fix reported issues - improve compatibility with legacy psycopg2 Co-authored-by: Polina Bungina <[email protected]>
Fixes around pending_restart flag (#3003 )
2026-08-26 07:30:14 +00:00 · 2024-01-17 08:35:35 +01:00 · 2024-01-16 10:44:30 +01:00 · 2024-01-15 12:24:42 +01:00 · 2024-01-08 09:17:17 +01:00 · 2024-01-05 10:17:07 +01:00
46 changed files with 977 additions and 248 deletions
@@ -174,7 +174,7 @@ jobs:

    - uses: jakebailey/pyright-action@v1
      with:
-        version: 1.1.333
+        version: 1.1.347

  docs:
    runs-on: ubuntu-latest
@@ -3,11 +3,15 @@
 Contributing guidelines
 =======================

+.. _chatting:
+
 Chatting
 --------

 If you have a question, looking for an interactive troubleshooting help or want to chat with other Patroni users, join us on channel `#patroni <https://postgresteam.slack.com/archives/C9XPYG92A>`__ in the `PostgreSQL Slack <https://pgtreats.info/slack-invite>`__.

+.. _reporting_bugs:
+
 Reporting bugs
 --------------

@@ -0,0 +1,329 @@
+.. _faq:
+
+FAQ
+===
+
+In this section you will find answers for the most frequently asked questions about Patroni.
+Each sub-section attempts to focus on different kinds of questions.
+
+We hope that this helps you to clarify most of your questions.
+If you still have further concerns or find yourself facing an unexpected issue, please refer to :ref:`chatting` and :ref:`reporting_bugs` for instructions on how to get help or report issues.
+
+Comparison with other HA solutions
+----------------------------------
+
+Why does Patroni require a separate cluster of DCS nodes while other solutions like ``repmgr`` do not?
+    There are different ways of implementing HA solutions, each of them with their pros and cons.
+
+    Software like ``repmgr`` performs communication among the nodes to decide when actions should be taken.
+
+    Patroni on the other hand relies on the state stored in the DCS. The DCS acts as a source of truth for Patroni to decide what it should do.
+
+    While having a separate DCS cluster can make you bloat your architecture, this approach also makes it less likely for split-brain scenarios to happen in your Postgres cluster.
+
+What is the difference between Patroni and other HA solutions in regards to Postgres management?
+    Patroni does not just manage the high availability of the Postgres cluster but also manages Postgres itself.
+
+    If Postgres nodes do not exist yet, it takes care of bootstrapping the primary and the standby nodes, and also manages Postgres configuration of the nodes. If the Postgres nodes already exist, Patroni will take over management of the cluster.
+
+    Besides the above, Patroni also has self-healing capabilities. In other words, if a primary node fails, Patroni will not only fail over to a replica, but also attempt to rejoin the former primary as a replica of the new primary. Similarly, if a replica fails, Patroni will attempt to rejoin that replica.
+
+    That is way we call Patroni as a "template for HA solutions". It goes further than just managing physical replication: it manages Postgres as a whole.
+
+DCS
+---
+
+Can I use the same ``etcd`` cluster to store data from two or more Patroni clusters?
+    Yes, you can!
+
+    Information about a Patroni cluster is stored in the DCS under a path prefixed with the ``namespace`` and ``scope`` Patroni settings.
+
+    As long as you do not have conflicting namespace and scope across different Patroni clusters, you should be able to use the same DCS cluster to store information from multiple Patroni clusters.
+
+What occurs if I attempt to use the same combination of ``namespace`` and ``scope`` for different Patroni clusters that point to the same DCS cluster?
+    The second Patroni cluster that attempts to use the same ``namespace`` and ``scope`` will not be able to manage Postgres because it will find information related with that same combination in the DCS, but with an incompatible Postgres system identifier.
+    The mismatch on the system identifier causes Patroni to abort the management of the second cluster, as it assumes that refers to a different cluster and that the user has misconfigured Patroni.
+
+    Make sure to use different ``namespace`` / ``scope`` when dealing with different Patroni clusters that share the same DCS cluster.
+
+What occurs if I lose my DCS cluster?
+    The DCS is used to store basically status and the dynamic configuration of the Patroni cluster.
+
+    They very first consequence is that all the Patroni clusters that rely on that DCS will go to read-only mode -- unless :ref:`dcs_failsafe_mode` is enabled.
+
+What should I do if I lose my DCS cluster?
+    There are three possible outcomes upon losing your DCS cluster:
+
+    1. The DCS cluster is fully recovered: this requires no action from the Patroni side. Once the DCS cluster is recovered, Patroni should be able to recover too;
+    2. The DCS cluster is re-created in place, and the endpoints remain the same. No changes are required on the Patroni side;
+    3. A new DCS cluster is created with different endpoints. You will need to update the DCS endpoints in the Patroni configuration of each Patroni node.
+
+    If you face scenario ``2.`` or ``3.`` Patroni will take care of creating the status information again based on the current status of the cluster, and recreate the dynamic configuration on the DCS based on a backup file named ``patroni.dynamic.json`` which is stored inside the Postgres data directory of each member of the Patroni cluster.
+
+What occurs if I lose majority in my DCS cluster?
+    The DCS will become unresponsive, which will cause Patroni to demote the current read/write Postgres node.
+
+    Remember: Patroni relies on the state of the DCS to take actions on the cluster.
+
+    You can use the :ref:`dcs_failsafe_mode` to alleviate that situation.
+
+patronictl
+----------
+
+Do I need to run :ref:`patronictl` in the Patroni host?
+    No, you do not need to do that.
+
+    Running :ref:`patronictl` in the Patroni host is handy if you have access to the Patroni host because you can use the very same configuration file from the ``patroni`` agent for the :ref:`patronictl` application.
+
+    However, :ref:`patronictl` is basically a client and it can be executed from remote machines. You just need to provide it with enough configuration so it can reach the DCS and the REST API of the Patroni member(s).
+
+Why did the information from one of my Patroni members disappear from the output of :ref:`patronictl_list` command?
+    Information shown by :ref:`patronictl_list` is based on the contents of the DCS.
+
+    If information about a member disappeared from the DCS it is very likely that the Patroni agent on that node is not running anymore, or it is not able to communicate with the DCS.
+
+    As the member is not able to update the information, the information eventually expires from the DCS, and consequently the member is not shown anymore in the output of :ref:`patronictl_list`.
+
+Why is the information about one of my Patroni members not up-to-date in the output of :ref:`patronictl_list` command?
+    Information shown by :ref:`patronictl_list` is based on the contents of the DCS.
+
+    By default, that information is updated by Patroni roughly every ``loop_wait`` seconds.
+    In other words, even if everything is normally functional you may still see a "delay" of up to ``loop_wait`` seconds in the information stored in the DCS.
+
+    Be aware that that is not a rule, though. Some operations performed by Patroni cause it to immediately update the DCS information.
+
+Configuration
+-------------
+
+What is the difference between dynamic configuration and local configuration?
+    Dynamic configuration (or global configuration) is the configuration stored in the DCS, and which is applied to all members of the Patroni cluster.
+    This is primarily where you should store your configuration.
+
+    Settings that are specific to a node, or settings that you would like to overwrite the global configuration with, you should set only on the desired Patroni member as a local configuration.
+    That local configuration can be specified either through the configuration file or through environment variables.
+
+    See more in :ref:`patroni_configuration`.
+
+What are the types of configuration in Patroni, and what is the precedence?
+    The types are:
+
+    * Dynamic configuration: applied to all members;
+    * Local configuration: applied to the local member, overrides dynamic configuration;
+    * Environment configuration: applied to the local member, overrides both dynamic and local configuration.
+
+    **Note:** some Postgres GUCs can only be set globally, i.e., through dynamic configuration. Besides that, there are GUCs which Patroni enforces a hard-coded value.
+
+    See more in :ref:`patroni_configuration`.
+
+Is there any facility to help me create my Patroni configuration file?
+    Yes, there is.
+
+    You can use ``patroni --generate-sample-config`` or ``patroni --generate-config`` commands to generate a sample Patroni configuration or a Patroni configuration based on an existing Postgres instance, respectively.
+
+    Please refer to :ref:`generate_sample_config` and :ref:`generate_config` for more details.
+
+I changed my parameters under ``bootstrap.dcs`` configuration but Patroni is not applying the changes to the cluster members. What is wrong?
+    The values configured under ``bootstrap.dcs`` are only used when bootstrapping a fresh cluster. Those values will be written to the DCS during the bootstrap.
+
+    After the bootstrap phase finishes, you will only be able to change the dynamic configuration through the DCS.
+
+    Refer to the next question for more details.
+
+How can I change my dynamic configuration?
+    You need to change the configuration in the DCS. That is accomplished either through:
+
+    * :ref:`patronictl_edit_config`; or
+    * A ``PATCH`` request to :ref:`config_endpoint`.
+
+How can I change my local configuration?
+    You need to change the configuration file of the corresponding Patroni member and signal the Patroni agent with ``SIHGUP``. You can do that using either of these approaches:
+
+    * Send a ``POST`` request to the REST API :ref:`reload_endpoint`; or
+    * Run :ref:`patronictl_reload`; or
+    * Locally signal the Patroni process with ``SIGHUP``:
+
+        * If you started Patroni through systemd, you can use the command ``systemctl reload PATRONI_UNIT.service``, ``PATRONI_UNIT`` being the name of the Patroni service; or
+        * If you started Patroni through other means, you will need to identify the ``patroni`` process and run ``kill -s HUP PID``, ``PID`` being the process ID of the ``patroni`` process.
+
+    **Note:** there are cases where a reload through the :ref:`patronictl_reload` may not work:
+
+    * Expired REST API certificates: you can mitigate that by using the ``-k`` option of the :ref:`patronictl`;
+    * Wrong credentials: for example when changing ``restapi`` or ``ctl`` credentials in the configuration file, and using that same configuration file for Patroni and :ref:`patronictl`.
+
+How can I change my environment configuration?
+    The environment configuration is only read by Patroni during startup.
+
+    With that in mind, if you change the environment configuration you will need to restart the corresponding Patroni agent.
+
+    Take care to not cause a failover in the cluster! You might be interested in checking :ref:`patronictl_pause`.
+
+What occurs if I change a Postgres GUC that requires a reload?
+    When you change the dynamic or the local configuration as explained in the previous questions, Patroni will take care of reloading the Postgres configuration for you.
+
+What occurs if I change a Postgres GUC that requires a restart?
+    Patroni will mark the affected members with a flag of ``pending restart``.
+
+    It is up to you to determine when and how to restart the members. That can be accomplished either through:
+
+    * :ref:`patronictl_restart`; or
+    * A ``POST`` request to :ref:`restart_endpoint`.
+
+    **Note:** some Postgres GUCs require a special management in terms of the order for restarting the Postgres nodes. Refer to :ref:`shared_memory_gucs` for more details.
+
+What is the difference between ``etcd`` and ``etcd3`` in Patroni configuration?
+    ``etcd`` uses the API version 2 of ``etcd``, while ``etcd3`` uses the API version 3 of ``etcd``.
+
+    Be aware that information stored by the API version 2 is not manageable by API version 3 and vice-versa.
+
+    We recommend that you configure ``etcd3`` instead of ``etcd`` because:
+
+    * API version 2 is disabled by default from Etcd v3.4 onward;
+    * API version 2 will be completely removed on Etcd v3.6.
+
+I have ``use_slots`` enabled in my Patroni configuration, but when a cluster member goes offline for some time, the replication slot used by that member is dropped on the upstream node. What can I do to avoid that issue?
+    You can configure a permanent physical replication slot for the members.
+
+    Since Patroni ``3.2.0`` it is now possible to have member slots as permanent slots managed by Patroni.
+
+    Patroni will create the permanent physical slots on all nodes, and make sure to not remove the slots, as well as to advance the slots' LSN on all nodes according to the LSN that has been consumed by the member.
+
+    Later, if you decide to remove the corresponding member, it's **your responsability** to adjust the permanent slots configuration, otherwise Patroni will keep the slots around forever.
+
+    **Note:** on Patroni older than ``3.2.0`` you could still have member slots configured as permanent physical slots, however they would be managed only on the current leader. That is, in case of failover/switchover these slots would be created on the new leader, but that wouldn't guarantee that it had all WAL segments for the absent node.
+
+    **Note:** even with Patroni ``3.2.0`` there might be a small race condition. In the very beginning, when the slot is created on the replica it could be ahead of the same slot on the leader and in case if nobody is consuming the slot there is still a chance that some files could be missing after failover. With that in mind, it is recommended that you configure continuous archiving, which makes it possible to restore required WALs or perform PITR.
+
+What is the difference between ``loop_wait``, ``retry_timeout`` and ``ttl``?
+    Patroni performs what we call a HA cycle from time to time. On each HA cycle it takes care of performing a series of checks on the cluster to determine its healthiness, and depending on the status it may take actions, like failing over to a standby.
+
+    ``loop_wait`` determines for how long, in seconds, Patroni should sleep before performing a new cycle of HA checks.
+
+    ``retry_timeout`` sets the timeout for retry operations on the DCS and on Postgres. For example: if the DCS is unresponsive for more than ``retry_timeout`` seconds, Patroni might demote the primary node as a security action.
+
+    ``ttl`` sets the lease time on the ``leader`` lock in the DCS. If the current leader of the cluster is not able to renew the lease during its HA cycles for longer than ``ttl``, then the lease will expire and that will trigger a ``leader race`` in the cluster.
+
+    **Note:** when modifying these settings, please keep in mind that Patroni enforces the rule and minimal values described in :ref:`dynamic_configuration` section of the docs.
+
+Postgres management
+-------------------
+
+Can I change Postgres GUCs directly in Postgres configuration?
+    You can, but you should avoid that.
+
+    Postgres configuration is managed by Patroni, and attempts to edit the configuration files may end up being frustrated by Patroni as it may eventually overwrite them.
+
+    There are a few options available to overcome the management performed by Patroni:
+
+    * Change Postgres GUCs through ``$PGDATA/postgresql.base.conf``; or
+    * Define a ``postgresql.custom_conf`` which will be used instead of ``postgresql.base.conf`` so you can manage that externally; or
+    * Change GUCs using ``ALTER SYSTEM`` / ``ALTER DATABASE`` / ``ALTER USER``.
+
+    You can find more information about that in the section :ref:`important_configuration_rules`.
+
+    In any case we recommend that you manage all the Postgres configuration through Patroni. That will centralize the management and make it easier to debug Patroni when needed.
+
+Can I restart Postgres nodes directly?
+    No, you should **not** attempt to manage Postgres directly!
+
+    Any attempt of bouncing the Postgres server without Patroni can lead your cluster to face failovers.
+
+    If you need to manage the Postgres server, do that through the ways exposed by Patroni.
+
+Is Patroni able to take over management of an already existing Postgres cluster?
+    Yes, it can!
+
+    Please refer to :ref:`existing_data` for detailed instructions.
+
+How does Patroni manage Postgres?
+    Patroni takes care of bringing Postgres up and down by running the Postgres binaries, like ``pg_ctl`` and ``postgres``.
+
+    With that in mind you **MUST** disable any other sources that could manage the Postgres clusters, like the systemd units, e.g. ``postgresql.service``. Only Patroni should be able to start, stop and promote Postgres instances in the cluster. Not doing so may result in split-brain scenarios. For example: if the node running as a primary failed and the unit ``postgresql.service`` is enabled, it may bring Postgres back up and cause a split-brain.
+
+Concepts and requirements
+-------------------------
+
+Which are the applications that make part of Patroni?
+    Patroni basically ships a couple applications:
+
+    * ``patroni``: This is the Patroni agent, which takes care of managing a Postgres node;
+    * ``patronictl``: This is a command-line utility used to interact with a Patroni cluster (perform switchovers, restarts, changes in the configuration, etc.). Please find more information in :ref:`patronictl`.
+
+What is a ``standby cluster`` in Patroni?
+    It is a cluster that does not have any primary Postgres node running, i.e., there is no read/write member in the cluster.
+
+    These kinds of clusters exist to replicate data from another cluster and are usually useful when you want to replicate data across data centers.
+
+    There will be a leader in the cluster which will be a standby in charge of replicating changes from a remote Postgres node.
+    Then, there will be a set of standbys configured with cascading replication from such leader member.
+
+    **Note:** the standby cluster doesn't know anything about the source cluster which it is replicating from -- it can even use ``restore_command`` instead of WAL streaming, and may use an absolutely independent DCS cluster.
+
+    Refer to :ref:`standby_cluster` for more details.
+
+What is a ``leader`` in Patroni?
+    A ``leader`` in Patroni is like a coordinator of the cluster.
+
+    In a regular Patroni cluster, the ``leader`` will be the read/write node.
+
+    In a standby Patroni cluster, the ``leader`` (AKA ``standby leader``) will be in charge of replicating from a remote Postgres node, and cascading those changes to the other members of the standby cluster.
+
+Does Patroni require a minimum number of Postgres nodes in the cluster?
+    No, you can run Patroni with any number of Postgres nodes.
+
+    Remember: Patroni is decoupled from the DCS.
+
+What does ``pause`` mean in Patroni?
+    Pause is an operation exposed by Patroni so the user can ask Patroni to step back in regards to Postgres management.
+
+    That is mainly useful when you want to perform maintenance on the cluster, and would like to avoid that Patroni takes decisions related with HA, like failing over to a standby when you stop the primary.
+
+    You can find more information about that in :ref:`pause`.
+
+Automatic failover
+------------------
+
+How does the automatic failover mechanism of Patroni work?
+    Patroni automatic failover is based on what we call ``leader race``.
+
+    Patroni stores the cluster's status in the DCS, among them a ``leader`` lock which holds the name of the Patroni member which is the current ``leader`` of the cluster.
+
+    That ``leader`` lock has a time-to-live associated with it. If the leader node fails to update the lease of the ``leader`` lock in time, the key will eventually expire from the DCS.
+
+    When the ``leader`` lock expires, it triggers what Patroni calls a ``leader race``: all nodes start performing checks to determine if they are the best candidates for taking over the ``leader`` role.
+    Some of these checks include calls to the REST API of all other Patroni members.
+
+    All Patroni members that find themselves as the best candidate for taking over the ``leader`` lock will attempt to do so.
+    The first Patroni member that is able to take the ``leader`` lock will promote itself to a read/write node (or ``standby leader``), and the others will be configured to follow it.
+
+Can I temporarily disable automatic failover in the Patroni cluster?
+    Yes, you can!
+
+    You can achieve that by temporarily pausing the cluster.
+    This is typically useful for performing maintenance.
+
+    When you want to resume the automatic failover of the cluster, you just need to unpause it.
+
+    You can find more information about that in :ref:`pause`.
+
+Bootstrapping and standbys creation
+-----------------------------------
+
+How does Patroni create a primary Postgres node? What about a standby Postgres node?
+    By default Patroni will use ``initdb`` to bootstrap a fresh cluster, and ``pg_basebackup`` to create standby nodes from a copy of the ``leader`` member.
+
+    You can customize that behavior by writing your custom bootstrap methods, and your custom replica creation methods.
+
+    Custom methods are usually useful when you want to restore backups created by backup tools like pgBackRest or Barman, for example.
+
+    For detailed information please refer to :ref:`custom_bootstrap` and :ref:`custom_replica_creation`.
+
+Monitoring
+----------
+
+How can I monitor my Patroni cluster?
+    Patroni exposes a couple handy endpoints in its :ref:`rest_api`:
+
+    * ``/metrics``: exposes monitoring metrics in a format that can be consumed by Prometheus;
+    * ``/patroni``: exposes the status of the cluster in a JSON format. The information shown here is very similar to what is shown by the ``/metrics`` endpoint.
+
+    You can use those endpoints to implement monitoring checks.
@@ -36,6 +36,7 @@ Currently supported PostgreSQL versions: 9.3 to 16.
   existing_data
   security
   ha_multi_dc
+   faq
   releases
   CONTRIBUTING

@@ -30,6 +30,7 @@ There are 3 types of Patroni configuration:
 	It is possible to set/override some of the "Local" configuration parameters with environment variables.
 	Environment configuration is very useful when you are running in a dynamic environment and you don't know some of the parameters in advance (for example it's not possible to know your external IP address when you are running inside ``docker``).

+.. _important_configuration_rules:

 Important rules
 ---------------
@@ -90,6 +91,7 @@ The parameters would be applied in the following order (run-time are given the h

 This allows configuration for all the nodes (2), configuration for a specific node using ``ALTER SYSTEM`` (3) and ensures that parameters essential to the running of Patroni are enforced (4), as well as leaves room for configuration tools that manage `postgresql.conf` directly without involving Patroni (1).

+.. _shared_memory_gucs:

 PostgreSQL parameters that touch shared memory
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
@@ -156,6 +158,7 @@ Patroni provides command-line interfaces for a Patroni :ref:`local configuration
 - Create a Patroni configuration file for the locally running PostgreSQL instance (e.g. as a preparation step for the :ref:`Patroni integration <existing_data>`);
 - Validate a given Patroni configuration file.

+.. _generate_sample_config:

 Sample Patroni configuration
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
@@ -183,6 +186,7 @@ Parameters

 ``configfile`` - full path to the configuration file used to store the result. If not provided, the result is sent to ``stdout``.

+.. _generate_config:

 Patroni configuration for a running instance
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
@@ -3,6 +3,94 @@
 Release notes
 =============

+Version 3.2.2
+-------------
+
+**Bugfixes**
+
+- Don't let replica restore initialize key when DCS was wiped (Alexander Kukushkin)
+
+  It was happening in the method where Patroni was supposed to take over a standalone PG cluster.
+
+- Use consistent read when fetching just updated sync key from Consul (Alexander Kukushkin)
+
+  Consul doesn't provide any interface to immediately get ``ModifyIndex`` for the key that we just updated, therefore we have to perform an explicit read operation. Since stale reads are allowed by default, we sometimes used to get an outdated version of the key.
+
+- Reload Postgres config if a parameter that requires restart was reset to the original value (Polina Bungina)
+
+  Previously Patroni wasn't updating the config, but only resetting the ``pending_restart``.
+
+- Fix erroneous inverted logic of the confirmation prompt message when doing a failover to an async candidate in synchronous mode (Polina Bungina)
+
+  The problem existed only in ``patronictl``.
+
+- Exclude leader from failover candidates in ``patronictl`` (Polina Bungina)
+
+  If the cluster is healthy, failing over to an existing leader is no-op.
+
+- Create Citus database and extension idempotently (Alexander Kukushkin, Zhao Junwang)
+
+  It will allow to create them in the ``post_bootstrap`` script in case if there is a need to add some more dependencies to the Citus database.
+
+- Don't filter our contradictory ``nofailover`` tag (Polina Bungina)
+
+  The configuration ``{nofailover: false, failover_priority: 0}`` set on a node didn't allow it to participate in the race, while it should, because ``nofailover`` tag should take precedence.
+
+- Fixed PyInstaller frozen issue (Sophia Ruan)
+
+  The ``freeze_support()`` was called after ``argparse`` and as a result, Patroni wasn't able to start Postgres.
+
+- Fixed bug in the config generator for ``patronictl`` and ``Citus`` configuration (Israel Barth Rubio)
+
+  It prevented ``patronictl`` and ``Citus`` configuration parameters set via environment variables from being written into the generated config.
+
+- Restore recovery GUCs and some Patroni-managed parameters when joining a running standby (Alexander Kukushkin)
+
+  Patroni was failing to restart Postgres v12 onwards with an error about missing ``port`` in one of the internal structures.
+
+- Fixes around ``pending_restart`` flag (Polina Bungina)
+
+  Don't expose ``pending_restart`` when in custom bootstrap with ``recovery_target_action = promote`` or when someone changed ``hot_standby`` or ``wal_log_hints`` using for example ``ALTER SYSTEM``.
+
+
+Version 3.2.1
+-------------
+
+**Bugfixes**
+
+- Limit accepted values for ``--format`` argument in ``patronictl`` (Alexander Kukushkin)
+
+  It used to accept any arbitrary string and produce no output if the value wasn't recognized.
+
+- Verify that replica nodes received checkpoint LSN on shutdown before releasing the leader key (Alexander Kukushkin)
+
+  Previously in some cases, we were using LSN of the SWITCH record that is followed by CHECKPOINT (if archiving mode is enabled). As a result the former primary sometimes had to do ``pg_rewind``, but there would be no data loss involved.
+
+- Do a real HTTP request when performing node name uniqueness check (Alexander Kukushkin)
+
+  When running Patroni in containers it is possible that the traffic is routed using ``docker-proxy``, which listens on the port and accepts incoming connections. It was causing false positives.
+
+- Fixed Citus support with Etcd v2 (Alexander Kukushkin)
+
+  Patroni was failing to deploy a new Citus cluster with Etcd v2.
+
+- Fixed ``pg_rewind`` behavior with Postgres v16+ (Alexander Kukushkin)
+
+  The error message format of ``pg_waldump`` changed in v16 which caused ``pg_rewind`` to be called by Patroni even when it was not necessary.
+
+- Fixed bug with custom bootstrap (Alexander Kukushkin)
+
+  Patroni was falsely applying ``--command`` argument, which is a bootstrap command itself.
+
+- Fixed the issue with REST API health check endpoints (Sophia Ruan)
+
+  There were chances that after Postgres restart it could return ``unknown`` state for Postgres because connections were not properly closed.
+
+- Cache ``postgres --describe-config`` output results (Waynerv)
+
+  They are used to figure out which GUCs are available to validate PostgreSQL configuration and we don't expect this list to change while Patroni is running.
+
+
 Version 3.2.0
 -------------

@@ -426,6 +426,7 @@ Cluster status endpoints
      ]
    ]

+.. _config_endpoint:

 Config endpoint
 ---------------
@@ -666,6 +667,7 @@ There are a couple of checks that a member of a cluster should pass to be able t
 	- its lag exceeds the maximum replication lag allowed;
 	- it has the timeline number smaller than the last known cluster timeline.

+.. _restart_endpoint:

 Restart endpoint
 ----------------
@@ -682,6 +684,7 @@ Restart endpoint

 ``POST /restart`` and ``DELETE /restart`` endpoints are used by :ref:`patronictl_restart` and :ref:`patronictl flush cluster-name restart <patronictl_flush_parameters>` respectively.

+.. _reload_endpoint:

 Reload endpoint
 ---------------
@@ -1073,6 +1073,8 @@ def before_all(context):
    context.keyfile = os.path.join(context.pctl.output_dir, 'patroni.key')
    context.certfile = os.path.join(context.pctl.output_dir, 'patroni.crt')
    try:
+        if sys.platform == 'darwin' and 'GITHUB_ACTIONS' in os.environ:
+            raise Exception
        with open(os.devnull, 'w') as null:
            ret = subprocess.call(['openssl', 'req', '-nodes', '-new', '-x509', '-subj', '/CN=batman.patroni',
                                   '-addext', 'subjectAltName=IP:127.0.0.1', '-keyout', context.keyfile,
@@ -6,10 +6,9 @@ Feature: priority replication
    And I configure and start postgres1 with a tag failover_priority 0
    Then replication works from postgres0 to postgres1 after 20 seconds
    When I shut down postgres0
-    And I sleep for 5 seconds
-    Then postgres1 role is the secondary after 10 seconds
    And there is one of ["following a different leader because I am not allowed to promote"] INFO in the postgres1 patroni log after 5 seconds
-    Given I start postgres0
+    Then postgres1 role is the secondary after 10 seconds
+    When I start postgres0
    Then postgres0 role is the primary after 10 seconds

  Scenario: check higher failover priority is respected
@@ -18,6 +17,23 @@ Feature: priority replication
    Then replication works from postgres0 to postgres2 after 20 seconds
    And replication works from postgres0 to postgres3 after 20 seconds
    When I shut down postgres0
-    And I sleep for 5 seconds
    Then postgres3 role is the primary after 10 seconds
    And there is one of ["postgres3 has equally tolerable WAL position and priority 2, while this node has priority 1","Wal position of postgres3 is ahead of my wal position"] INFO in the postgres2 patroni log after 5 seconds
+
+ Scenario: check conflicting configuration handling
+    When I set nofailover tag in postgres2 config
+    And I issue an empty POST request to http://127.0.0.1:8010/reload
+    Then I receive a response code 202
+    And there is one of ["Conflicting configuration between nofailover: True and failover_priority: 1. Defaulting to nofailover: True"] WARNING in the postgres2 patroni log after 5 seconds
+    And "members/postgres2" key in DCS has tags={'failover_priority': '1', 'nofailover': True} after 10 seconds
+    When I issue a POST request to http://127.0.0.1:8010/failover with {"candidate": "postgres2"}
+    Then I receive a response code 412
+    And I receive a response text "failover is not possible: no good candidates have been found"
+    When I reset nofailover tag in postgres1 config
+    And I issue an empty POST request to http://127.0.0.1:8009/reload
+    Then I receive a response code 202
+    And there is one of ["Conflicting configuration between nofailover: False and failover_priority: 0. Defaulting to nofailover: False"] WARNING in the postgres1 patroni log after 5 seconds
+    And "members/postgres1" key in DCS has tags={'failover_priority': '0', 'nofailover': False} after 10 seconds
+    And I issue a POST request to http://127.0.0.1:8009/failover with {"candidate": "postgres1"}
+    Then I receive a response code 200
+    And postgres1 role is the primary after 10 seconds
@@ -114,7 +114,7 @@ def replication_works(context, primary, replica, time_limit):
    """.format(str(time()).replace('.', '_').replace(',', '_'), primary, replica, time_limit))


-@then('there is one of {message_list} {level:w} in the {node} patroni log after {timeout:d} seconds')
+@step('there is one of {message_list} {level:w} in the {node} patroni log after {timeout:d} seconds')
 def check_patroni_log(context, message_list, level, node, timeout):
    timeout *= context.timeout_multiplier
    message_list = json.loads(message_list)
@@ -123,6 +123,6 @@ def check_patroni_log(context, message_list, level, node, timeout):
        messsages_of_level = context.pctl.read_patroni_log(node, level)
        if any(any(message in line for line in messsages_of_level) for message in message_list):
            break
-        time.sleep(1)
+        sleep(1)
    else:
        assert False, f"There were none of {message_list} {level} in the {node} patroni log after {timeout} seconds"
@@ -131,5 +131,5 @@ def check_transaction(context, name, time_limit):

@step("a transaction finishes in {timeout:d} seconds")
 def check_transaction_timeout(context, timeout):
-    assert (datetime.now(tzutc) - context.xact_start).seconds > timeout, \
+    assert (datetime.now(tzutc) - context.xact_start).seconds >= timeout, \
        "a transaction finished earlier than in {0} seconds".format(timeout)
@@ -128,6 +128,12 @@ def scheduled_restart(context, url, in_seconds, data):
    context.execute_steps(u"""Given I issue a POST request to {0}/restart with {1}""".format(url, json.dumps(data)))


+@step('I {action:w} {tag:w} tag in {pg_name:w} config')
+def add_bool_tag_to_config(context, action, tag, pg_name):
+    value = action == 'set'
+    context.pctl.add_tag_to_config(pg_name, tag, value)
+
+
@step('I add tag {tag:w} {value:w} to {pg_name:w} config')
 def add_tag_to_config(context, tag, value, pg_name):
    context.pctl.add_tag_to_config(pg_name, tag, value)
@@ -107,8 +107,6 @@ class Patroni(AbstractPatroniDaemon, Tags):

    def ensure_unique_name(self) -> None:
        """A helper method to prevent splitbrain from operator naming error."""
-        from urllib.parse import urlparse
-        from urllib3.connection import HTTPConnection
        from patroni.dcs import Member

        cluster = self.dcs.get_cluster()
@@ -118,14 +116,14 @@ class Patroni(AbstractPatroniDaemon, Tags):
        if not isinstance(member, Member):
            return
        try:
-            parts = urlparse(member.api_url)
-            if isinstance(parts.hostname, str):
-                connection = HTTPConnection(parts.hostname, port=parts.port or 80, timeout=3)
-                connection.connect()
-                logger.fatal("Can't start; there is already a node named '%s' running", self.config['name'])
-                sys.exit(1)
+            # Silence annoying WARNING: Retrying (...) messages when Patroni is quickly restarted.
+            # At this moment we don't have custom log levels configured and hence shouldn't lose anything useful.
+            self.logger.update_loggers({'urllib3.connectionpool': 'ERROR'})
+            _ = self.request(member, endpoint="/liveness", timeout=3)
+            logger.fatal("Can't start; there is already a node named '%s' running", self.config['name'])
+            sys.exit(1)
        except Exception:
-            return
+            self.logger.update_loggers({})

    def _get_tags(self) -> Dict[str, Any]:
        """Get tags configured for this node, if any.
@@ -231,11 +229,6 @@ def patroni_main(configfile: str) -> None:

    :param configfile: path to Patroni configuration file.
    """
-    from multiprocessing import freeze_support
-
-    # Windows executables created by PyInstaller are frozen, thus we need to enable frozen support for
-    # :mod:`multiprocessing` to avoid :class:`RuntimeError` exceptions.
-    freeze_support()
    abstract_main(Patroni, configfile)


@@ -337,6 +330,12 @@ def main() -> None:
        ``patroni`` daemon as another process. In that case relevant signals received by the main process and forwarded
        to ``patroni`` daemon process.
    """
+    from multiprocessing import freeze_support
+
+    # Executables created by PyInstaller are frozen, thus we need to enable frozen support for
+    # :mod:`multiprocessing` to avoid :class:`RuntimeError` exceptions.
+    freeze_support()
+
    check_psycopg()

    args = process_arguments()
@@ -37,7 +37,7 @@ from .utils import deep_compare, enable_keepalive, parse_bool, patch_config, Ret
 logger = logging.getLogger(__name__)


-def check_access(func: Callable[['RestApiHandler'], None]) -> Callable[..., None]:
+def check_access(func: Callable[..., None]) -> Callable[..., None]:
    """Check the source ip, authorization header, or client certificates.

    .. note::
@@ -290,10 +290,10 @@ class Config(object):
        self.__effective_configuration = self._build_effective_configuration({}, self._local_configuration)
        self._data_dir = self.__effective_configuration.get('postgresql', {}).get('data_dir', "")
        self._cache_file = os.path.join(self._data_dir, self.__CACHE_FILENAME)
-        if validator:  # patronictl uses validator=None and we don't want to load anything from local cache in this case
-            self._load_cache()
+        if validator:  # patronictl uses validator=None
+            self._load_cache()  # we don't want to load anything from local cache for ctl
+            self._validate_failover_tags()  # irrelevant for ctl
        self._cache_needs_saving = False
-        self._validate_failover_tags()

    @property
    def config_file(self) -> Optional[str]:
@@ -504,6 +504,7 @@ class Config(object):
                    new_configuration = self._build_effective_configuration(self._dynamic_configuration, configuration)
                    self._local_configuration = configuration
                    self.__effective_configuration = new_configuration
+                    self._validate_failover_tags()
                    return True
                else:
                    logger.info('No local configuration items changed.')
@@ -974,10 +975,12 @@ class Config(object):
          bedrock source of truth)
        """
        tags = self.get('tags', {})
+        if 'nofailover' not in tags:
+            return
        nofailover_tag = tags.get('nofailover')
        failover_priority_tag = parse_int(tags.get('failover_priority'))
        if failover_priority_tag is not None \
-                and (nofailover_tag is True and failover_priority_tag > 0
-                     or nofailover_tag is False and failover_priority_tag <= 0):
+                and (bool(nofailover_tag) is True and failover_priority_tag > 0
+                     or bool(nofailover_tag) is False and failover_priority_tag <= 0):
            logger.warning('Conflicting configuration between nofailover: %s and failover_priority: %s. '
                           'Defaulting to nofailover: %s', nofailover_tag, failover_priority_tag, nofailover_tag)
@@ -178,7 +178,7 @@ class AbstractConfigGenerator(abc.ABC):

        :yields: formatted lines or blocks that represent a text output of the YAML document.
        """
-        for name in ('scope', 'namespace', 'name', 'log', 'restapi', 'ctl' 'citus',
+        for name in ('scope', 'namespace', 'name', 'log', 'restapi', 'ctl', 'citus',
                     'consul', 'etcd', 'etcd3', 'exhibitor', 'kubernetes', 'raft', 'zookeeper'):
            yield from self._format_config_section(name)

@@ -255,7 +255,8 @@ def load_config(path: str, dcs_url: Optional[str]) -> Dict[str, Any]:
    return config


-option_format = click.option('--format', '-f', 'fmt', help='Output format (pretty, tsv, json, yaml)', default='pretty')
+option_format = click.option('--format', '-f', 'fmt', help='Output format', default='pretty',
+                             type=click.Choice(['pretty', 'tsv', 'json', 'yaml', 'yml']))
 option_watchrefresh = click.option('-w', '--watch', type=float, help='Auto update the screen every X seconds')
 option_watch = click.option('-W', is_flag=True, help='Auto update the screen every 2 seconds')
 option_force = click.option('--force', is_flag=True, help='Do not ask for confirmation at any point')
@@ -811,7 +812,7 @@ def query(
            raise PatroniCtlException('You need to specify either --command or --file')
        sql = command

-    connect_parameters = {}
+    connect_parameters: Dict[str, str] = {}
    if username:
        connect_parameters['username'] = username
    if password:
@@ -1094,7 +1095,7 @@ def restart(obj: Dict[str, Any], cluster_name: str, group: Optional[int], member
        version = click.prompt('Restart if the PostgreSQL version is less than provided (e.g. 9.5.2) ',
                               type=str, default='')

-    content = {}
+    content: Dict[str, Any] = {}
    if pending:
        content['restart_pending'] = True

@@ -1189,7 +1190,7 @@ def reinit(obj: Dict[str, Any], cluster_name: str, group: Optional[int],


 def _do_failover_or_switchover(obj: Dict[str, Any], action: str, cluster_name: str,
-                               group: Optional[int], leader: Optional[str], candidate: Optional[str],
+                               group: Optional[int], switchover_leader: Optional[str], candidate: Optional[str],
                               force: bool, scheduled: Optional[str] = None) -> None:
    """Perform a failover or a switchover operation in the cluster.

@@ -1204,7 +1205,7 @@ def _do_failover_or_switchover(obj: Dict[str, Any], action: str, cluster_name: s
    :param cluster_name: name of the Patroni cluster.
    :param group: filter Citus group within we should perform a failover or switchover. If ``None``, user will be
        prompted for filling it -- unless *force* is ``True``, in which case an exception is raised.
-    :param leader: name of the current leader member.
+    :param switchover_leader: name of the leader member passed as switchover option.
    :param candidate: name of a standby member to be promoted. Nodes that are tagged with ``nofailover`` cannot be used.
    :param force: perform the failover or switchover without asking for confirmations.
    :param scheduled: timestamp when the switchover should be scheduled to occur. If ``now`` perform immediately.
@@ -1213,10 +1214,11 @@ def _do_failover_or_switchover(obj: Dict[str, Any], action: str, cluster_name: s
        :class:`PatroniCtlException`: if:
            * Patroni is running on a Citus cluster, but no *group* was specified; or
            * a switchover was requested by the cluster has no leader; or
-            * *leader* does not match the current leader of the cluster; or
+            * *switchover_leader* does not match the current leader of the cluster; or
            * cluster has no candidates available for the operation; or
            * no *candidate* is given for a failover operation; or
-            * *leader* and *candidate* are the same; or
+            * current leader and *candidate* are the same; or
+            * *candidate* is tagged as nofailover; or
            * *candidate* is not a member of the cluster; or
            * trying to schedule a switchover in a cluster that is in maintenance mode; or
            * user aborts the operation.
@@ -1236,23 +1238,24 @@ def _do_failover_or_switchover(obj: Dict[str, Any], action: str, cluster_name: s

    global_config = get_global_config(cluster)

+    cluster_leader = cluster.leader and cluster.leader.name
    # leader has to be be defined for switchover only
    if action == 'switchover':
-        if cluster.leader is None or not cluster.leader.name:
+        if not cluster_leader:
            raise PatroniCtlException('This cluster has no leader')

-        if leader is None:
+        if switchover_leader is None:
            if force:
-                leader = cluster.leader.name
+                switchover_leader = cluster_leader
            else:
                prompt = 'Standby Leader' if global_config.is_standby_cluster else 'Primary'
-                leader = click.prompt(prompt, type=str, default=(cluster.leader and cluster.leader.name))
+                switchover_leader = click.prompt(prompt, type=str, default=cluster_leader)

-        if cluster.leader.name != leader:
-            raise PatroniCtlException(f'Member {leader} is not the leader of cluster {cluster_name}')
+        if cluster_leader != switchover_leader:
+            raise PatroniCtlException(f'Member {switchover_leader} is not the leader of cluster {cluster_name}')

    # excluding members with nofailover tag
-    candidate_names = [str(m.name) for m in cluster.members if m.name != leader and not m.nofailover]
+    candidate_names = [str(m.name) for m in cluster.members if m.name != cluster_leader and not m.nofailover]
    # We sort the names for consistent output to the client
    candidate_names.sort()

@@ -1265,10 +1268,10 @@ def _do_failover_or_switchover(obj: Dict[str, Any], action: str, cluster_name: s
    if action == 'failover' and not candidate:
        raise PatroniCtlException('Failover could be performed only to a specific candidate')

-    if candidate == leader:
-        raise PatroniCtlException(action.title() + ' target and source are the same.')
-
    if candidate and candidate not in candidate_names:
+        if candidate == cluster_leader:
+            raise PatroniCtlException(
+                f'Member {candidate} is already the leader of cluster {cluster_name}')
        raise PatroniCtlException(
            f'Member {candidate} does not exist in cluster {cluster_name} or is tagged as nofailover')

@@ -1277,7 +1280,7 @@ def _do_failover_or_switchover(obj: Dict[str, Any], action: str, cluster_name: s
            global_config.is_synchronous_mode,
            not cluster.sync.is_empty,
            not cluster.sync.matches(candidate, True))):
-        if click.confirm(f'Are you sure you want to failover to the asynchronous node {candidate}'):
+        if not click.confirm(f'Are you sure you want to failover to the asynchronous node {candidate}?'):
            raise PatroniCtlException('Aborting ' + action)

    scheduled_at_str = None
@@ -1297,7 +1300,7 @@ def _do_failover_or_switchover(obj: Dict[str, Any], action: str, cluster_name: s

    failover_value = {'candidate': candidate}
    if action == 'switchover':
-        failover_value['leader'] = leader
+        failover_value['leader'] = switchover_leader
    if scheduled_at_str:
        failover_value['scheduled_at'] = scheduled_at_str

@@ -1305,7 +1308,7 @@ def _do_failover_or_switchover(obj: Dict[str, Any], action: str, cluster_name: s

    # By now we have established that the leader exists and the candidate exists
    if not force:
-        demote_msg = f', demoting current leader {cluster.leader.name}' if cluster.leader else ''
+        demote_msg = f', demoting current leader {cluster_leader}' if cluster_leader else ''
        if scheduled_at_str:
            # only switchover can be scheduled
            if not click.confirm(f'Are you sure you want to schedule switchover of cluster '
@@ -1339,7 +1342,7 @@ def _do_failover_or_switchover(obj: Dict[str, Any], action: str, cluster_name: s
        logging.exception(r)
        logging.warning('Failing over to DCS')
        click.echo('{0} Could not {1} using Patroni api, falling back to DCS'.format(timestamp(), action))
-        dcs.manual_failover(leader, candidate, scheduled_at=scheduled_at)
+        dcs.manual_failover(switchover_leader, candidate, scheduled_at=scheduled_at)

    output_members(obj, cluster, cluster_name, group=group)

@@ -1415,7 +1418,7 @@ def switchover(obj: Dict[str, Any], cluster_name: str, group: Optional[int],


 def generate_topology(level: int, member: Dict[str, Any],
-                      topology: Dict[str, List[Dict[str, Any]]]) -> Iterator[Dict[str, Any]]:
+                      topology: Dict[Optional[str], List[Dict[str, Any]]]) -> Iterator[Dict[str, Any]]:
    """Recursively yield members with their names adjusted according to their *level* in the cluster topology.

    .. note::
@@ -1478,7 +1481,7 @@ def topology_sort(members: List[Dict[str, Any]]) -> Iterator[Dict[str, Any]]:
    :yields: *members* sorted by level in the topology, and with a new ``name`` value according to their level
        in the topology.
    """
-    topology: Dict[str, List[Dict[str, Any]]] = defaultdict(list)
+    topology: Dict[Optional[str], List[Dict[str, Any]]] = defaultdict(list)
    leader = next((m for m in members if m['role'].endswith('leader')), {'name': None})
    replicas = set(member['name'] for member in members if not member['role'].endswith('leader'))
    for member in members:
@@ -1564,7 +1564,7 @@ class AbstractDCS(abc.ABC):
        """

    @abc.abstractmethod
-    def _citus_cluster_loader(self, path: Any) -> Union[Cluster, Dict[int, Cluster]]:
+    def _citus_cluster_loader(self, path: Any) -> Dict[int, Cluster]:
        """Load and build all Patroni clusters from a single Citus cluster.

        :param path: the path in DCS where to load Cluster(s) from.
@@ -422,8 +422,8 @@ class Consul(AbstractDCS):
    def _cluster_loader(self, path: str) -> Cluster:
        _, results = self.retry(self._client.kv.get, path, recurse=True, consistency=self._consistency)
        if results is None:
-            raise NotFound
-        nodes = {}
+            return Cluster.empty()
+        nodes: Dict[str, Dict[str, Any]] = {}
        for node in results:
            node['Value'] = (node['Value'] or b'').decode('utf-8')
            nodes[node['Key'][len(path):]] = node
@@ -445,8 +445,6 @@ class Consul(AbstractDCS):
    ) -> Union[Cluster, Dict[int, Cluster]]:
        try:
            return loader(path)
-        except NotFound:
-            return Cluster.empty()
        except Exception:
            logger.exception('get_cluster')
            raise ConsulError('Consul is not responding properly')
@@ -668,7 +666,7 @@ class Consul(AbstractDCS):
        if ret:  # We have no other choise, only read after write :(
            if not retry.ensure_deadline(0.5):
                return False
-            _, ret = self.retry(self._client.kv.get, self.sync_path)
+            _, ret = self.retry(self._client.kv.get, self.sync_path, consistency='consistent')
            if ret and (ret.get('Value') or b'').decode('utf-8') == value:
                return ret['ModifyIndex']
        return False
@@ -710,13 +710,20 @@ class Etcd(AbstractEtcd):
        return Cluster(initialize, config, leader, status, members, failover, sync, history, failsafe)

    def _cluster_loader(self, path: str) -> Cluster:
-        result = self.retry(self._client.read, path, recursive=True, quorum=self._ctl)
+        try:
+            result = self.retry(self._client.read, path, recursive=True, quorum=self._ctl)
+        except etcd.EtcdKeyNotFound:
+            return Cluster.empty()
        nodes = {node.key[len(result.key):].lstrip('/'): node for node in result.leaves}
        return self._cluster_from_nodes(result.etcd_index, nodes)

    def _citus_cluster_loader(self, path: str) -> Dict[int, Cluster]:
+        try:
+            result = self.retry(self._client.read, path, recursive=True, quorum=self._ctl)
+        except etcd.EtcdKeyNotFound:
+            return {}
+
        clusters: Dict[int, Dict[str, etcd.EtcdResult]] = defaultdict(dict)
-        result = self.retry(self._client.read, path, recursive=True, quorum=self._ctl)
        for node in result.leaves:
            key = node.key[len(result.key):].lstrip('/').split('/', 1)
            if len(key) == 2 and citus_group_re.match(key[0]):
@@ -729,8 +736,6 @@ class Etcd(AbstractEtcd):
        cluster = None
        try:
            cluster = loader(path)
-        except etcd.EtcdKeyNotFound:
-            cluster = Cluster.empty()
        except Exception as e:
            self._handle_exception(e, 'get_cluster', raise_ex=EtcdError('Etcd is not responding properly'))
        self._has_failed = False
@@ -1227,15 +1227,16 @@ class Ha(object):

        status = {'released': False}

-        def on_shutdown(checkpoint_location: int) -> None:
+        def on_shutdown(checkpoint_location: int, prev_location: int) -> None:
            # Postmaster is still running, but pg_control already reports clean "shut down".
            # It could happen if Postgres is still archiving the backlog of WAL files.
            # If we know that there are replicas that received the shutdown checkpoint
            # location, we can remove the leader key and allow them to start leader race.
+            time.sleep(1)  # give replicas some more time to catch up
            if self.is_failover_possible(cluster_lsn=checkpoint_location):
                self.state_handler.set_role('demoted')
                with self._async_executor:
-                    self.release_leader_key_voluntarily(checkpoint_location)
+                    self.release_leader_key_voluntarily(prev_location)
                    status['released'] = True

        def before_shutdown() -> None:
@@ -1850,10 +1851,9 @@ class Ha(object):
                            logger.fatal('system ID mismatch, node %s belongs to a different cluster: %s != %s',
                                         self.state_handler.name, self.cluster.initialize, data_sysid)
                            sys.exit(1)
-                elif self.cluster.is_unlocked() and not self.is_paused():
+                elif self.cluster.is_unlocked() and not self.is_paused() and not self.state_handler.cb_called:
                    # "bootstrap", but data directory is not empty
-                    if not self.state_handler.cb_called and self.state_handler.is_running() \
-                            and not self.state_handler.is_primary():
+                    if self.state_handler.is_running() and not self.state_handler.is_primary():
                        self._join_aborted = True
                        logger.error('No initialize key in DCS and PostgreSQL is running as replica, aborting start')
                        logger.error('Please first start Patroni on the node running as primary')
@@ -1990,18 +1990,18 @@ class Ha(object):

            status = {'deleted': False}

-            def _on_shutdown(checkpoint_location: int) -> None:
+            def _on_shutdown(checkpoint_location: int, prev_location: int) -> None:
                if self.is_leader():
                    # Postmaster is still running, but pg_control already reports clean "shut down".
                    # It could happen if Postgres is still archiving the backlog of WAL files.
                    # If we know that there are replicas that received the shutdown checkpoint
                    # location, we can remove the leader key and allow them to start leader race.
-
+                    time.sleep(1)  # give replicas some more time to catch up
                    if self.is_failover_possible(cluster_lsn=checkpoint_location):
-                        self.dcs.delete_leader(self.cluster.leader, checkpoint_location)
+                        self.dcs.delete_leader(self.cluster.leader, prev_location)
                        status['deleted'] = True
                    else:
-                        self.dcs.write_leader_optime(checkpoint_location)
+                        self.dcs.write_leader_optime(prev_location)

            def _before_shutdown() -> None:
                self.notify_citus_coordinator('before_demote')
@@ -202,24 +202,37 @@ class PatroniLogger(Thread):
        self._proxy_handler = ProxyHandler(self)
        self._root_logger.addHandler(self._proxy_handler)

-    def update_loggers(self) -> None:
-        """Configure loggers' log level as defined in ``log.loggers`` section of Patroni configuration.
+    def update_loggers(self, config: Dict[str, Any]) -> None:
+        """Configure custom loggers' log levels.

        .. note::
            It creates logger objects that are not defined yet in the log manager.
+
+        :param config: :class:`dict` object with custom loggers configuration, is set either from:
+
+                       * ``log.loggers`` section of Patroni configuration; or
+
+                       * from the method that is trying to make sure that the node name
+                         isn't duplicated (to silence annoying ``urllib3`` WARNING's).
+
+        :Example:
+
+            .. code-block:: python
+
+                update_loggers({'urllib3.connectionpool': 'WARNING'})
        """
-        loggers = deepcopy((self._config or {}).get('loggers') or {})
+        loggers = deepcopy(config)
        for name, logger in self._root_logger.manager.loggerDict.items():
            # ``Placeholder`` is a node in the log manager for which no logger has been defined. We are interested only
            # in the ones that were defined
            if not isinstance(logger, logging.PlaceHolder):
-                # if this logger is present in ``log.loggers`` Patroni configuration, use the configured level,
-                # otherwise use ``logging.NOTSET``, which means it will inherit the level from any parent node up to
-                # the root for which log level is defined.
+                # if this logger is present in *config*, use the configured level, otherwise
+                # use ``logging.NOTSET``, which means it will inherit the level
+                # from any parent node up to the root for which log level is defined.
                level = loggers.pop(name, logging.NOTSET)
                logger.setLevel(level)

-        # define loggers that do not exist yet and set level as configured in ``log.loggers`` section of configuration.
+        # define loggers that do not exist yet and set level as configured in the *config*
        for name, level in loggers.items():
            logger = self._root_logger.manager.getLogger(name)
            logger.setLevel(level)
@@ -274,7 +287,7 @@ class PatroniLogger(Thread):
                    self.log_handler = new_handler

            self._config = config.copy()
-            self.update_loggers()
+            self.update_loggers(config.get('loggers') or {})

    def _close_old_handlers(self) -> None:
        """Close old log handlers.
@@ -119,6 +119,8 @@ class Postgresql(object):
        # Last known running process
        self._postmaster_proc = None

+        self._available_gucs = None
+
        if self.is_running():
            # If we found postmaster process we need to figure out whether postgres is accepting connections
            self.set_state('starting')
@@ -241,7 +243,9 @@ class Postgresql(object):
    @property
    def available_gucs(self) -> CaseInsensitiveSet:
        """GUCs available in this Postgres server."""
-        return self._get_gucs()
+        if not self._available_gucs:
+            self._available_gucs = self._get_gucs()
+        return self._available_gucs

    def _version_file_exists(self) -> bool:
        return not self.data_directory_empty() and os.path.isfile(self._version_file)
@@ -591,14 +595,17 @@ class Postgresql(object):
                return match.group(1), match.group(2), match.group(3), match.group(4)
        return None, None, None, None

-    def latest_checkpoint_location(self) -> Optional[int]:
-        """Returns checkpoint location for the cleanly shut down primary.
-           But, if we know that the checkpoint was written to the new WAL
-           due to the archive_mode=on, we will return the LSN of prev wal record (SWITCH)."""
+    def _checkpoint_locations_from_controldata(self, data: Dict[str, str]) -> Optional[Tuple[int, int]]:
+        """Get shutdown checkpoint location.

-        data = self.controldata()
+        :param data: :class:`dict` object with values returned by `pg_controldata` tool.
+
+        :returns: a tuple of checkpoint LSN for the cleanly shut down primary, and LSN of prev wal record (SWITCH)
+                  if we know that the checkpoint was written to the new WAL file due to the archive_mode=on.
+        """
        timeline = data.get("Latest checkpoint's TimeLineID")
        lsn = checkpoint_lsn = data.get('Latest checkpoint location')
+        prev_lsn = None
        if data.get('Database cluster state') == 'shut down' and lsn and timeline and checkpoint_lsn:
            try:
                checkpoint_lsn = parse_lsn(checkpoint_lsn)
@@ -609,13 +616,26 @@ class Postgresql(object):
                    _, lsn, _, desc = self.parse_wal_record(timeline, prev)
                    prev = parse_lsn(prev)
                    # If the cluster is shutdown with archive_mode=on, WAL is switched before writing the checkpoint.
-                    # In this case we want to take the LSN of previous record (switch) as the last known WAL location.
+                    # In this case we want to take the LSN of previous record (SWITCH) as the last known WAL location.
                    if lsn and parse_lsn(lsn) == prev and str(desc).strip() in ('xlog switch', 'SWITCH'):
-                        return prev
+                        prev_lsn = prev
            except Exception as e:
                logger.error('Exception when parsing WAL pg_%sdump output: %r', self.wal_name, e)
            if isinstance(checkpoint_lsn, int):
-                return checkpoint_lsn
+                return checkpoint_lsn, (prev_lsn or checkpoint_lsn)
+
+    def latest_checkpoint_location(self) -> Optional[int]:
+        """Get shutdown checkpoint location.
+
+        .. note::
+            In case if checkpoint was written to the new WAL file due to the archive_mode=on
+            we return LSN of the previous wal record (SWITCH).
+
+        :returns: checkpoint LSN for the cleanly shut down primary.
+        """
+        checkpoint_locations = self._checkpoint_locations_from_controldata(self.controldata())
+        if checkpoint_locations:
+            return checkpoint_locations[1]

    def is_running(self) -> Optional[PostmasterProcess]:
        """Returns PostmasterProcess if one is running on the data directory or None. If most recently seen process
@@ -801,7 +821,7 @@ class Postgresql(object):
            return 'not accessible or not healty'

    def stop(self, mode: str = 'fast', block_callbacks: bool = False, checkpoint: Optional[bool] = None,
-             on_safepoint: Optional[Callable[..., Any]] = None, on_shutdown: Optional[Callable[[int], Any]] = None,
+             on_safepoint: Optional[Callable[..., Any]] = None, on_shutdown: Optional[Callable[[int, int], Any]] = None,
             before_shutdown: Optional[Callable[..., Any]] = None, stop_timeout: Optional[int] = None) -> bool:
        """Stop PostgreSQL

@@ -831,7 +851,7 @@ class Postgresql(object):
        return success

    def _do_stop(self, mode: str, block_callbacks: bool, checkpoint: bool,
-                 on_safepoint: Optional[Callable[..., Any]], on_shutdown: Optional[Callable[..., Any]],
+                 on_safepoint: Optional[Callable[..., Any]], on_shutdown: Optional[Callable[[int, int], Any]],
                 before_shutdown: Optional[Callable[..., Any]], stop_timeout: Optional[int]) -> Tuple[bool, bool]:
        postmaster = self.is_running()
        if not postmaster:
@@ -871,7 +891,9 @@ class Postgresql(object):
            while postmaster.is_running():
                data = self.controldata()
                if data.get('Database cluster state', '') == 'shut down':
-                    on_shutdown(self.latest_checkpoint_location())
+                    checkpoint_locations = self._checkpoint_locations_from_controldata(data)
+                    if checkpoint_locations:
+                        on_shutdown(*checkpoint_locations)
                    break
                elif data.get('Database cluster state', '').startswith('shut down'):  # shut down in recovery
                    break
@@ -188,10 +188,9 @@ class Bootstrap(object):
        params = [] if config.get('no_params') else ['--scope=' + self._postgresql.scope,
                                                     '--datadir=' + self._postgresql.data_dir]
        # Add custom parameters specified by the user
-        reserved_args = {'no_params', 'keep_existing_recovery_conf', 'recovery_conf', 'scope', 'datadir'}
-        for arg, val in config.items():
-            if arg not in reserved_args:
-                params.append(f"--{arg}={val}")
+        reserved_args = {'command', 'no_params', 'keep_existing_recovery_conf', 'recovery_conf', 'scope', 'datadir'}
+        params += [f"--{arg}={val}" for arg, val in config.items() if arg not in reserved_args]
+
        try:
            logger.info('Running custom bootstrap script: %s', config['command'])
            if self._postgresql.cancellable.call(shlex.split(config['command']) + params) != 0:
@@ -7,7 +7,7 @@ from urllib.parse import urlparse
 from typing import Any, Dict, List, Optional, Union, Tuple, TYPE_CHECKING

 from ..dcs import CITUS_COORDINATOR_GROUP_ID, Cluster
-from ..psycopg import connect, quote_ident
+from ..psycopg import connect, quote_ident, ProgrammingError

 if TYPE_CHECKING:  # pragma: no cover
    from . import Postgresql
@@ -100,7 +100,8 @@ class CitusHandler(Thread):
    def on_demote(self) -> None:
        with self._condition:
            self._pg_dist_node.clear()
-            self._tasks[:] = []
+            empty_tasks: List[PgDistNode] = []
+            self._tasks[:] = empty_tasks
            self._in_flight = None

    def query(self, sql: str, *params: Any) -> List[Tuple[Any, ...]]:
@@ -363,6 +364,11 @@ class CitusHandler(Thread):
                with conn.cursor() as cur:
                    cur.execute('CREATE DATABASE {0}'.format(
                        quote_ident(self._config['database'], conn)).encode('utf-8'))
+            except ProgrammingError as exc:
+                if exc.diag.sqlstate == '42P04':  # DuplicateDatabase
+                    logger.debug('Exception when creating database: %r', exc)
+                else:
+                    raise exc
            finally:
                conn.close()

@@ -370,7 +376,7 @@ class CitusHandler(Thread):
        conn = connect(**conn_kwargs)
        try:
            with conn.cursor() as cur:
-                cur.execute('CREATE EXTENSION citus')
+                cur.execute('CREATE EXTENSION IF NOT EXISTS citus')

                superuser = self._postgresql.config.superuser
                params = {k: superuser[k] for k in ('password', 'sslcert', 'sslkey') if k in superuser}
@@ -336,12 +336,24 @@ class ConfigHandler(object):

    def load_current_server_parameters(self) -> None:
        """Read GUC's values from ``pg_settings`` when Patroni is joining the the postgres that is already running."""
-        exclude = [name.lower() for name, value in self.CMDLINE_OPTIONS.items() if value[1] == _false_validator] \
-            + [name.lower() for name in self._RECOVERY_PARAMETERS]
-        self._server_parameters = CaseInsensitiveDict({r[0]: r[1] for r in self._postgresql.query(
+        exclude = [name.lower() for name, value in self.CMDLINE_OPTIONS.items() if value[1] == _false_validator]
+        keep_values = {k: self._server_parameters[k] for k in exclude}
+        server_parameters = CaseInsensitiveDict({r[0]: r[1] for r in self._postgresql.query(
            "SELECT name, pg_catalog.current_setting(name) FROM pg_catalog.pg_settings"
            " WHERE (source IN ('command line', 'environment variable') OR sourcefile = %s)"
            " AND pg_catalog.lower(name) != ALL(%s)", self._postgresql_conf, exclude)})
+        recovery_params = CaseInsensitiveDict({k: server_parameters.pop(k) for k in self._RECOVERY_PARAMETERS
+                                               if k in server_parameters})
+        # We also want to load current settings of recovery parameters, including primary_conninfo
+        # and primary_slot_name, otherwise patronictl restart will update postgresql.conf
+        # and remove them, what in the worst case will cause another restart.
+        # We are doing it only for PostgresSQL v12 onwards, because older version still have recovery.conf
+        if not self._postgresql.is_primary() and self._postgresql.major_version >= 120000:
+            # primary_conninfo is expected to be a dict, therefore we need to parse it
+            recovery_params['primary_conninfo'] = parse_dsn(recovery_params.pop('primary_conninfo', '')) or {}
+            self._recovery_params = recovery_params
+
+        self._server_parameters = CaseInsensitiveDict({**server_parameters, **keep_values})

    def setup_server_parameters(self) -> None:
        self._server_parameters = self.get_server_parameters(self._config)
@@ -1064,13 +1076,14 @@ class ConfigHandler(object):
    def reload_config(self, config: Dict[str, Any], sighup: bool = False) -> None:
        self._superuser = config['authentication'].get('superuser', {})
        server_parameters = self.get_server_parameters(config)
+        params_skip_changes = CaseInsensitiveSet((*self._RECOVERY_PARAMETERS, 'hot_standby', 'wal_log_hints'))

        conf_changed = hba_changed = ident_changed = local_connection_address_changed = pending_restart = False
        if self._postgresql.state == 'running':
            changes = CaseInsensitiveDict({p: v for p, v in server_parameters.items()
-                                           if p.lower() not in self._RECOVERY_PARAMETERS})
+                                           if p not in params_skip_changes})
            changes.update({p: None for p in self._server_parameters.keys()
-                            if not (p in changes or p.lower() in self._RECOVERY_PARAMETERS)})
+                            if not (p in changes or p in params_skip_changes)})
            if changes:
                undef = []
                if 'wal_buffers' in changes:  # we need to calculate the default value of wal_buffers
@@ -1097,6 +1110,12 @@ class ConfigHandler(object):
                                    local_connection_address_changed = True
                            else:
                                logger.info('Changed %s from %s to %s', r[0], r[1], new_value)
+                        elif r[0] in self._server_parameters \
+                                and not compare_values(r[3], r[2], r[1], self._server_parameters[r[0]]):
+                            # Check if any parameter was set back to the current pg_settings value
+                            # We can use pg_settings value here, as it is proved to be equal to new_value
+                            logger.info('Changed %s from %s to %s', r[0], self._server_parameters[r[0]], r[1])
+                            conf_changed = True
                for param, value in changes.items():
                    if '.' in param:
                        # Check that user-defined-paramters have changed (parameters with period in name)
@@ -1150,7 +1169,7 @@ class ConfigHandler(object):
                    pending_restart = self._postgresql.query(
                        'SELECT COUNT(*) FROM pg_catalog.pg_settings'
                        ' WHERE pg_catalog.lower(name) != ALL(%s) AND pending_restart',
-                        [n.lower() for n in self._RECOVERY_PARAMETERS])[0][0] > 0
+                        [n.lower() for n in params_skip_changes])[0][0] > 0
                    self._postgresql.set_pending_restart(pending_restart)
                except Exception as e:
                    logger.warning('Exception %r when running query', e)
@@ -1225,7 +1244,6 @@ class ConfigHandler(object):

            if disable_hot_standby:
                effective_configuration['hot_standby'] = 'off'
-                self._postgresql.set_pending_restart(True)

        return effective_configuration

@@ -147,7 +147,8 @@ class ConnectionPool:
    def close(self) -> None:
        """Close all named connections from Patroni to PostgreSQL registered in the pool."""
        with self._lock:
-            if any(conn.close(True) for conn in self._connections.values()):
+            closed_connections = [conn.close(True) for conn in self._connections.values()]
+            if any(closed_connections):
                logger.info("closed patroni connections to postgres")


@@ -101,12 +101,26 @@ class Rewind(object):
            return 'not accessible or not healty'

    def _get_checkpoint_end(self, timeline: int, lsn: int) -> int:
-        """The checkpoint record size in WAL depends on postgres major version and platform (memory alignment).
-        Hence, the only reliable way to figure out where it ends, read the record from file with the help of pg_waldump
-        and parse the output. We are trying to read two records, and expect that it will fail to read the second one:
-        `pg_waldump: fatal: error in WAL record at 0/182E220: invalid record length at 0/182E298: wanted 24, got 0`
-        The error message contains information about LSN of the next record, which is exactly where checkpoint ends."""
+        """Get the end of checkpoint record from WAL.

+        .. note::
+            The checkpoint record size in WAL depends on postgres major version and platform (memory alignment).
+            Hence, the only reliable way to figure out where it ends, is to read the record from file with the
+            help of ``pg_waldump`` and parse the output.
+
+            We are trying to read two records, and expect that it will fail to read the second record with message:
+
+                fatal: error in WAL record at 0/182E220: invalid record length at 0/182E298: wanted 24, got 0; or
+
+                fatal: error in WAL record at 0/182E220: invalid record length at 0/182E298: expected at least 24, got 0
+
+            The error message contains information about LSN of the next record, which is exactly where checkpoint ends.
+
+        :param timeline: the checkpoint *timeline* from ``pg_controldata``.
+        :param lsn: the checkpoint *location* as :class:`int` from ``pg_controldata``.
+
+        :returns: the end of checkpoint record as :class:`int` or ``0`` if failed to parse ``pg_waldump`` output.
+        """
        lsn8 = format_lsn(lsn, True)
        lsn_str = format_lsn(lsn)
        out, err = self._postgresql.waldump(timeline, lsn_str, 2)
@@ -117,12 +131,17 @@ class Rewind(object):

            if len(out) == 1 and len(err) == 1 and ', lsn: {0}, prev '.format(lsn8) in out[0] and pattern in err[0]:
                i = err[0].find(pattern) + len(pattern)
-                j = err[0].find(": wanted ", i)
-                if j > -1:
-                    try:
-                        return parse_lsn(err[0][i:j])
-                    except Exception as e:
-                        logger.error('Failed to parse lsn %s: %r', err[0][i:j], e)
+                # Message format depends on the major version:
+                # * expected at least -- starting from v16
+                # * wanted -- before v16
+                # We will simply check all possible combinations.
+                for pattern in (': expected at least ', ': wanted '):
+                    j = err[0].find(pattern, i)
+                    if j > -1:
+                        try:
+                            return parse_lsn(err[0][i:j])
+                        except Exception as e:
+                            logger.error('Failed to parse lsn %s: %r', err[0][i:j], e)
            logger.error('Failed to parse pg_%sdump output', self._postgresql.wal_name)
            logger.error(' stdout=%s', '\n'.join(out))
            logger.error(' stderr=%s', '\n'.join(err))
@@ -22,14 +22,18 @@ class Tags(abc.ABC):
            A custom tag is any tag added to the configuration ``tags`` section that is not one of ``clonefrom``,
            ``nofailover``, ``noloadbalance`` or ``nosync``.

-            For the Patroni predefined tags, the returning object will only contain them if they are enabled as they
-            all are boolean values that default to disabled.
+            For most of the Patroni predefined tags, the returning object will only contain them if they are enabled as
+            they all are boolean values that default to disabled.
+            However ``nofailover`` tag is always returned if ``failover_priority`` tag is defined. In this case, we need
+            both values to see if they are contradictory and the ``nofailover`` value should be used.

        :returns: a dictionary of tags set for this node. The key is the tag name, and the value is the corresponding
            tag value.
        """
        return {tag: value for tag, value in tags.items()
-                if tag not in ('clonefrom', 'nofailover', 'noloadbalance', 'nosync') or value}
+                if any((tag not in ('clonefrom', 'nofailover', 'noloadbalance', 'nosync'),
+                        value,
+                        tag == 'nofailover' and 'failover_priority' in tags))}

    @property
    @abc.abstractmethod
@@ -401,22 +401,23 @@ def parse_real(value: Any, base_unit: Optional[str] = None) -> Optional[float]:
        return convert_to_base_unit(val, unit, base_unit)


-def compare_values(vartype: str, unit: Optional[str], old_value: Any, new_value: Any) -> bool:
-    """Check if *old_value* and *new_value* are equivalent after parsing them as *vartype*.
+def compare_values(vartype: str, unit: Optional[str], settings_value: Any, config_value: Any) -> bool:
+    """Check if the value from ``pg_settings`` and from Patroni config are equivalent after parsing them as *vartype*.

-    :param vartpe: the target type to parse *old_value* and *new_value* before comparing them. Accepts any among of the
-        following (case sensitive):
+    :param vartype: the target type to parse *settings_value* and *config_value* before comparing them.
+        Accepts any among of the following (case sensitive):

        * ``bool``: parse values using :func:`parse_bool`; or
        * ``integer``: parse values using :func:`parse_int`; or
        * ``real``: parse values using :func:`parse_real`; or
        * ``enum``: parse values as lowercase strings; or
        * ``string``: parse values as strings. This one is used by default if no valid value is passed as *vartype*.
-    :param unit: base unit to be used as argument when calling :func:`parse_int` or :func:`parse_real` for *new_value*.
-    :param old_value: value to be compared with *new_value*.
-    :param new_value: value to be compared with *old_value*.
+    :param unit: base unit to be used as argument when calling :func:`parse_int` or :func:`parse_real`
+        for *config_value*.
+    :param settings_value: value to be compared with *config_value*.
+    :param config_value: value to be compared with *settings_value*.

-    :returns: ``True`` if *old_value* is equivalent to *new_value* when both are parsed as *vartype*.
+    :returns: ``True`` if *settings_value* is equivalent to *config_value* when both are parsed as *vartype*.

    :Example:

@@ -456,8 +457,8 @@ def compare_values(vartype: str, unit: Optional[str], old_value: Any, new_value:
    }

    converter = converters.get(vartype) or converters['string']
-    old_converted = converter(old_value, None)
-    new_converted = converter(new_value, unit)
+    old_converted = converter(settings_value, None)
+    new_converted = converter(config_value, unit)

    return old_converted is not None and new_converted is not None and old_converted == new_converted

@@ -9,7 +9,7 @@ import os
 import shutil
 import socket

-from typing import Any, Dict, Union, Iterator, List, Optional as OptionalType, Tuple
+from typing import Any, Dict, Union, Iterator, List, Optional as OptionalType, Tuple, TYPE_CHECKING

 from .collections import CaseInsensitiveSet

@@ -200,6 +200,8 @@ def get_bin_name(bin_name: str) -> str:

    :returns: value of ``postgresql.bin_name[*bin_name*]``, if present, otherwise *bin_name*.
    """
+    if TYPE_CHECKING:  # pragma: no cover
+        assert isinstance(schema.data, dict)
    return (schema.data.get('postgresql', {}).get('bin_name', {}) or {}).get(bin_name, bin_name)


@@ -239,6 +241,8 @@ def validate_data_dir(data_dir: str) -> bool:
            if not os.path.isdir(os.path.join(data_dir, waldir)):
                raise ConfigParseError("data dir for the cluster is not empty, but doesn't contain"
                                       " \"{}\" directory".format(waldir))
+            if TYPE_CHECKING:  # pragma: no cover
+                assert isinstance(schema.data, dict)
            bin_dir = schema.data.get("postgresql", {}).get("bin_dir", None)
            major_version = get_major_version(bin_dir, get_bin_name('postgres'))
            if pgversion != major_version:
@@ -274,6 +278,8 @@ def validate_binary_name(bin_name: str) -> bool:
    """
    if not bin_name:
        raise ConfigParseError("is an empty string")
+    if TYPE_CHECKING:  # pragma: no cover
+        assert isinstance(schema.data, dict)
    bin_dir = schema.data.get('postgresql', {}).get('bin_dir', None)
    if not shutil.which(bin_name, path=bin_dir):
        raise ConfigParseError(f"does not contain '{bin_name}' in '{bin_dir or '$PATH'}'")
@@ -523,7 +529,7 @@ class Schema(object):
        * :class:`dict`: dictionary representing the YAML configuration tree.
    """

-    def __init__(self, validator: Any) -> None:
+    def __init__(self, validator: Union[Dict[Any, Any], List[Any], Any]) -> None:
        """Create a :class:`Schema` object.

        .. note::
@@ -614,7 +620,7 @@ class Schema(object):
                errors.append(str(i))
        return errors

-    def validate(self, data: Any) -> Iterator[Result]:
+    def validate(self, data: Union[Dict[Any, Any], Any]) -> Iterator[Result]:
        """Perform all validations from the schema against the given configuration.

        It first checks that *data* argument type is compliant with the type of ``validator`` attribute.
@@ -638,11 +644,8 @@ class Schema(object):
        # iterable objects in the structure, until we eventually reach a leaf node to validate its value.
        if isinstance(self.validator, str):
            yield Result(isinstance(self.data, str), "is not a string", level=1, data=self.data)
-        elif issubclass(type(self.validator), type):
-            validator = self.validator
-            if self.validator == str:
-                validator = str
-            yield Result(isinstance(self.data, validator),
+        elif isinstance(self.validator, type):
+            yield Result(isinstance(self.data, self.validator),
                         "is not {}".format(_get_type_name(self.validator)), level=1, data=self.data)
        elif callable(self.validator):
            if hasattr(self.validator, "expected_type"):
@@ -689,7 +692,7 @@ class Schema(object):
                    for v in Schema(self.validator[0]).validate(value):
                        yield Result(v.status, v.error,
                                     path=(str(key) + ("." + v.path if v.path else "")), level=v.level, data=value)
-        elif isinstance(self.validator, Directory):
+        elif isinstance(self.validator, Directory) and isinstance(self.data, str):
            yield from self.validator.validate(self.data)
        elif isinstance(self.validator, Or):
            yield from self.iter_or()
@@ -701,6 +704,9 @@ class Schema(object):
        """
        # One key in `validator` attribute (`key` variable) can be mapped to one or more keys in `data` attribute (`d`
        # variable), depending on the `key` type.
+        if TYPE_CHECKING:  # pragma: no cover
+            assert isinstance(self.validator, dict)
+            assert isinstance(self.data, dict)
        for key in self.validator.keys():
            if isinstance(key, AtMostOne) and len(list(self._data_key(key))) > 1:
                yield Result(False, f"Multiple of {key.args} provided")
@@ -730,6 +736,8 @@ class Schema(object):

        :yields: objects with the error message related to the failure, if any check fails.
        """
+        if TYPE_CHECKING:  # pragma: no cover
+            assert isinstance(self.validator, Or)
        results: List[Result] = []
        for a in self.validator.args:
            r: List[Result] = []
@@ -766,7 +774,7 @@ class Schema(object):
            yield key.name
        # If the key was defined as an `Or` object in `validator` attribute, then each of its values are the keys to
        # access the `data` dictionary.
-        elif isinstance(key, Or):
+        elif isinstance(key, Or) and isinstance(self.data, dict):
            # At least one of the `Or` entries should be available in the `data` dictionary. If we find at least one of
            # them in `data`, then we return all found entries so the caller method can validate them all.
            if any([item in self.data for item in key.args]):
@@ -780,7 +788,7 @@ class Schema(object):
                    yield item
        # If the key was defined as a `AtMostOne` object in `validator` attribute, then each of its values
        # are the keys to access the `data` dictionary.
-        elif isinstance(key, AtMostOne):
+        elif isinstance(key, AtMostOne) and isinstance(self.data, dict):
            # Yield back all of the entries from the `data` dictionary, each will be validated and then counted
            # to inform us if we've provided too many
            for item in key.args:
@@ -2,4 +2,4 @@

 :var __version__: the current Patroni version.
 """
-__version__ = '3.2.0'
+__version__ = '3.2.2'
@@ -132,7 +132,7 @@ postgresql:
 #  safety_margin: 5

 tags:
-    nofailover: false
+    # failover_priority: 1
    noloadbalance: false
    clonefrom: false
    nosync: false
@@ -124,6 +124,6 @@ postgresql:
  #pre_promote: /path/to/pre_promote.sh

 tags:
-    nofailover: false
+    # failover_priority: 1
    noloadbalance: false
    clonefrom: false
@@ -114,7 +114,7 @@ postgresql:
 #    krb_server_keyfile: /var/spool/keytabs/postgres
    unix_socket_directories: '..'  # parent directory of data_dir
 tags:
-    nofailover: false
+    # failover_priority: 1
    noloadbalance: false
    clonefrom: false
 #    replicatefrom: postgresql1
@@ -25,8 +25,41 @@ mock_available_gucs = PropertyMock(return_value={
    'max_wal_senders', 'max_worker_processes', 'port', 'search_path', 'shared_preload_libraries',
    'stats_temp_directory', 'synchronous_standby_names', 'track_commit_timestamp', 'unix_socket_directories',
    'vacuum_cost_delay', 'vacuum_cost_limit', 'wal_keep_size', 'wal_level', 'wal_log_hints', 'zero_damaged_pages',
+    'autovacuum', 'wal_segment_size', 'wal_block_size', 'shared_buffers', 'wal_buffers',
 })

+GET_PG_SETTINGS_RESULT = [
+    ('wal_segment_size', '2048', '8kB', 'integer', 'internal'),
+    ('wal_block_size', '8192', None, 'integer', 'internal'),
+    ('shared_buffers', '16384', '8kB', 'integer', 'postmaster'),
+    ('wal_buffers', '-1', '8kB', 'integer', 'postmaster'),
+    ('max_connections', '100', None, 'integer', 'postmaster'),
+    ('max_prepared_transactions', '200', None, 'integer', 'postmaster'),
+    ('max_worker_processes', '8', None, 'integer', 'postmaster'),
+    ('max_locks_per_transaction', '64', None, 'integer', 'postmaster'),
+    ('max_wal_senders', '5', None, 'integer', 'postmaster'),
+    ('search_path', 'public', None, 'string', 'user'),
+    ('port', '5432', None, 'integer', 'postmaster'),
+    ('listen_addresses', '127.0.0.2, 127.0.0.3', None, 'string', 'postmaster'),
+    ('autovacuum', 'on', None, 'bool', 'sighup'),
+    ('unix_socket_directories', '/tmp', None, 'string', 'postmaster'),
+    ('shared_preload_libraries', 'citus', None, 'string', 'postmaster'),
+    ('wal_keep_size', '128', 'MB', 'integer', 'sighup'),
+    ('cluster_name', 'batman', None, 'string', 'postmaster'),
+    ('vacuum_cost_delay', '200', 'ms', 'real', 'user'),
+    ('vacuum_cost_limit', '-1', None, 'integer', 'user'),
+    ('max_stack_depth', '2048', 'kB', 'integer', 'superuser'),
+    ('constraint_exclusion', '', None, 'enum', 'user'),
+    ('force_parallel_mode', '1', None, 'enum', 'user'),
+    ('zero_damaged_pages', 'off', None, 'bool', 'superuser'),
+    ('stats_temp_directory', '/tmp', None, 'string', 'sighup'),
+    ('track_commit_timestamp', 'off', None, 'bool', 'postmaster'),
+    ('wal_log_hints', 'on', None, 'bool', 'postmaster'),
+    ('hot_standby', 'on', None, 'bool', 'postmaster'),
+    ('max_replication_slots', '5', None, 'integer', 'postmaster'),
+    ('wal_level', 'logical', None, 'enum', 'postmaster'),
+]
+

 class MockResponse(object):

@@ -133,22 +166,9 @@ class MockCursor(object):
                            ('archive_command', 'my archive command'),
                            ('cluster_name', 'my_cluster')]
        elif sql.startswith('SELECT name, setting'):
-            self.results = [('wal_segment_size', '2048', '8kB', 'integer', 'internal'),
-                            ('wal_block_size', '8192', None, 'integer', 'internal'),
-                            ('shared_buffers', '16384', '8kB', 'integer', 'postmaster'),
-                            ('wal_buffers', '-1', '8kB', 'integer', 'postmaster'),
-                            ('max_connections', '100', None, 'integer', 'postmaster'),
-                            ('max_prepared_transactions', '0', None, 'integer', 'postmaster'),
-                            ('max_worker_processes', '8', None, 'integer', 'postmaster'),
-                            ('max_locks_per_transaction', '64', None, 'integer', 'postmaster'),
-                            ('max_wal_senders', '5', None, 'integer', 'postmaster'),
-                            ('search_path', 'public', None, 'string', 'user'),
-                            ('port', '5433', None, 'integer', 'postmaster'),
-                            ('listen_addresses', '*', None, 'string', 'postmaster'),
-                            ('autovacuum', 'on', None, 'bool', 'sighup'),
-                            ('unix_socket_directories', '/tmp', None, 'string', 'postmaster')]
+            self.results = GET_PG_SETTINGS_RESULT
        elif sql.startswith('SELECT COUNT(*) FROM pg_catalog.pg_settings'):
-            self.results = [(1,)]
+            self.results = [(0,)]
        elif sql.startswith('IDENTIFY_SYSTEM'):
            self.results = [('1', 3, '0/402EEC0', '')]
        elif sql.startswith('TIMELINE_HISTORY '):
@@ -218,11 +238,11 @@ class PostgresInit(unittest.TestCase):
    _PARAMETERS = {'wal_level': 'hot_standby', 'max_replication_slots': 5, 'f.oo': 'bar',
                   'search_path': 'public', 'hot_standby': 'on', 'max_wal_senders': 5,
                   'wal_keep_segments': 8, 'wal_log_hints': 'on', 'max_locks_per_transaction': 64,
-                   'max_worker_processes': 8, 'max_connections': 100, 'max_prepared_transactions': 0,
+                   'max_worker_processes': 8, 'max_connections': 100, 'max_prepared_transactions': 200,
                   'track_commit_timestamp': 'off', 'unix_socket_directories': '/tmp',
-                   'trigger_file': 'bla', 'stats_temp_directory': '/tmp', 'zero_damaged_pages': '',
+                   'trigger_file': 'bla', 'stats_temp_directory': '/tmp', 'zero_damaged_pages': 'off',
                   'force_parallel_mode': '1', 'constraint_exclusion': '',
-                   'max_stack_depth': 'Z', 'vacuum_cost_limit': -1, 'vacuum_cost_delay': 200}
+                   'max_stack_depth': 2048, 'vacuum_cost_limit': -1, 'vacuum_cost_delay': 200}

    @patch('patroni.psycopg._connect', psycopg_connect)
    @patch('patroni.postgresql.CallbackExecutor', Mock())
@@ -179,10 +179,17 @@ class TestBootstrap(BaseTestPostgresql):
    @patch.object(Postgresql, 'controldata', Mock(return_value={'Database cluster state': 'in production'}))
    def test_custom_bootstrap(self, mock_cancellable_subprocess_call):
        self.p.config._config.pop('pg_hba')
-        config = {'method': 'foo', 'foo': {'command': 'bar'}}
+        config = {'method': 'foo', 'foo': {'command': 'bar --arg1=val1'}}

        mock_cancellable_subprocess_call.return_value = 1
        self.assertFalse(self.b.bootstrap(config))
+        self.assertEqual(mock_cancellable_subprocess_call.call_args_list[0][0][0],
+                         ['bar', '--arg1=val1', '--scope=batman', '--datadir=' + os.path.join('data', 'test0')])
+
+        mock_cancellable_subprocess_call.reset_mock()
+        config['foo']['no_params'] = 1
+        self.assertFalse(self.b.bootstrap(config))
+        self.assertEqual(mock_cancellable_subprocess_call.call_args_list[0][0][0], ['bar', '--arg1=val1'])

        mock_cancellable_subprocess_call.return_value = 0
        with patch('multiprocessing.Process', Mock(side_effect=Exception("42"))), \
@@ -1,6 +1,7 @@
 import time
-from mock import Mock, patch
+from mock import Mock, patch, PropertyMock
 from patroni.postgresql.citus import CitusHandler
+from patroni.psycopg import ProgrammingError

 from . import BaseTestPostgresql, MockCursor, psycopg_connect, SleepException
 from .test_ha import get_cluster_initialized_with_leader
@@ -161,3 +162,16 @@ class TestCitus(BaseTestPostgresql):
                                                         'type': 'logical', 'database': 'citus', 'plugin': 'pgoutput'}))
        self.assertTrue(self.c.ignore_replication_slot({'name': 'citus_shard_split_slot_1_2_3',
                                                        'type': 'logical', 'database': 'citus', 'plugin': 'citus'}))
+
+    @patch('patroni.postgresql.citus.logger.debug')
+    @patch('patroni.postgresql.citus.connect', psycopg_connect)
+    @patch('patroni.postgresql.citus.quote_ident', Mock())
+    def test_bootstrap_duplicate_database(self, mock_logger):
+        with patch.object(MockCursor, 'execute', Mock(side_effect=ProgrammingError)):
+            self.assertRaises(ProgrammingError, self.c.bootstrap)
+        with patch.object(MockCursor, 'execute', Mock(side_effect=[ProgrammingError, None, None, None])), \
+                patch.object(ProgrammingError, 'diag') as mock_diag:
+            type(mock_diag).sqlstate = PropertyMock(return_value='42P04')
+            self.c.bootstrap()
+        mock_logger.assert_called_once()
+        self.assertTrue(mock_logger.call_args[0][0].startswith('Exception when creating database'))
@@ -155,48 +155,40 @@ class TestConfig(unittest.TestCase):
    @patch('patroni.config.logger')
    def test__validate_failover_tags(self, mock_logger, mock_get):
        """Ensures that only one of `nofailover` or `failover_priority` can be provided"""
-        mock_logger.warning.reset_mock()
        config = Config("postgres0.yml")
+
        # Providing one of `nofailover` or `failover_priority` is fine
-        just_nofailover = {"nofailover": True}
-        mock_get.side_effect = [just_nofailover] * 2
-        self.assertIsNone(config._validate_failover_tags())
-        mock_logger.warning.assert_not_called()
-        just_failover_priority = {"failover_priority": 1}
-        mock_get.side_effect = [just_failover_priority] * 2
-        self.assertIsNone(config._validate_failover_tags())
-        mock_logger.warning.assert_not_called()
+        for single_param in ({"nofailover": True}, {"failover_priority": 1}, {"failover_priority": 0}):
+            mock_get.side_effect = [single_param] * 2
+            self.assertIsNone(config._validate_failover_tags())
+            mock_logger.warning.assert_not_called()
+
        # Providing both `nofailover` and `failover_priority` is fine if consistent
-        consistent_false = {"nofailover": False, "failover_priority": 1}
-        mock_get.side_effect = [consistent_false] * 2
-        self.assertIsNone(config._validate_failover_tags())
-        mock_logger.warning.assert_not_called()
-        consistent_true = {"nofailover": True, "failover_priority": 0}
-        mock_get.side_effect = [consistent_true] * 2
-        self.assertIsNone(config._validate_failover_tags())
-        mock_logger.warning.assert_not_called()
+        for consistent_state in (
+            {"nofailover": False, "failover_priority": 1},
+            {"nofailover": True, "failover_priority": 0},
+            {"nofailover": "False", "failover_priority": 0}
+        ):
+            mock_get.side_effect = [consistent_state] * 2
+            self.assertIsNone(config._validate_failover_tags())
+            mock_logger.warning.assert_not_called()
+
        # Providing both inconsistently should log a warning
-        inconsistent_false = {"nofailover": False, "failover_priority": 0}
-        mock_get.side_effect = [inconsistent_false] * 2
-        self.assertIsNone(config._validate_failover_tags())
-        mock_logger.warning.assert_called_once_with(
-            'Conflicting configuration between nofailover: %s and failover_priority: %s.'
-            + ' Defaulting to nofailover: %s',
-            False,
-            0,
-            False
-        )
-        mock_logger.warning.reset_mock()
-        inconsistent_true = {"nofailover": True, "failover_priority": 1}
-        mock_get.side_effect = [inconsistent_true] * 2
-        self.assertIsNone(config._validate_failover_tags())
-        mock_logger.warning.assert_called_once_with(
-            'Conflicting configuration between nofailover: %s and failover_priority: %s.'
-            + ' Defaulting to nofailover: %s',
-            True,
-            1,
-            True
-        )
+        for inconsistent_state in (
+            {"nofailover": False, "failover_priority": 0},
+            {"nofailover": True, "failover_priority": 1},
+            {"nofailover": "False", "failover_priority": 1},
+            {"nofailover": "", "failover_priority": 0}
+        ):
+            mock_get.side_effect = [inconsistent_state] * 2
+            self.assertIsNone(config._validate_failover_tags())
+            mock_logger.warning.assert_called_once_with(
+                'Conflicting configuration between nofailover: %s and failover_priority: %s.'
+                + ' Defaulting to nofailover: %s',
+                inconsistent_state['nofailover'],
+                inconsistent_state['failover_priority'],
+                inconsistent_state['nofailover'])
+            mock_logger.warning.reset_mock()

    def test__process_postgresql_parameters(self):
        expected_params = {
@@ -156,7 +156,8 @@ class TestCtl(unittest.TestCase):
        # Target and source are equal
        result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'], input='leader\nleader\n\ny')
        self.assertEqual(result.exit_code, 1)
-        self.assertIn('Switchover target and source are the same', result.output)
+        self.assertIn("Candidate ['other']", result.output)
+        self.assertIn('Member leader is already the leader of cluster dummy', result.output)

        # Candidate is not a member of the cluster
        result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'], input='leader\nReality\n\ny')
@@ -223,7 +224,10 @@ class TestCtl(unittest.TestCase):
        result = self.runner.invoke(ctl, ['failover', 'dummy'], input='0\n')
        self.assertIn('Failover could be performed only to a specific candidate', result.output)

-        cluster = get_cluster_initialized_with_leader(sync=('leader', 'other'))
+        # Candidate is the same as the leader
+        result = self.runner.invoke(ctl, ['failover', 'dummy', '--group', '0'], input='leader\n')
+        self.assertIn("Candidate ['other']", result.output)
+        self.assertIn('Member leader is already the leader of cluster dummy', result.output)

        # Temp test to check a fallback to switchover if leader is specified
        with patch('patroni.ctl._do_failover_or_switchover') as failover_func_mock:
@@ -232,17 +236,20 @@ class TestCtl(unittest.TestCase):
            failover_func_mock.assert_called_once_with(
                DEFAULT_CONFIG, 'switchover', 'dummy', None, 'leader', None, False)

-        # Failover to an async member in sync mode (confirm)
+        cluster = get_cluster_initialized_with_leader(sync=('leader', 'other'))
        cluster.members.append(Member(0, 'async', 28, {'api_url': 'http://127.0.0.1:8012/patroni'}))
        cluster.config.data['synchronous_mode'] = True
        mock_get_dcs.return_value.get_cluster = Mock(return_value=cluster)
-        result = self.runner.invoke(ctl, ['failover', 'dummy', '--group', '0', '--candidate', 'async'], input='y\ny')
+        # Failover to an async member in sync mode (confirm)
+        result = self.runner.invoke(ctl,
+                                    ['failover', 'dummy', '--group', '0', '--candidate', 'async'], input='y\ny')
        self.assertIn('Are you sure you want to failover to the asynchronous node async', result.output)
+        self.assertEqual(result.exit_code, 0)

        # Failover to an async member in sync mode (abort)
-        mock_get_dcs.return_value.get_cluster = Mock(return_value=cluster)
        result = self.runner.invoke(ctl, ['failover', 'dummy', '--group', '0', '--candidate', 'async'], input='N')
        self.assertEqual(result.exit_code, 1)
+        self.assertIn('Aborting failover', result.output)

    @patch('patroni.dcs.dcs_modules', Mock(return_value=['patroni.dcs.dummy', 'patroni.dcs.etcd']))
    def test_get_dcs(self):
@@ -274,6 +274,8 @@ class TestEtcd(unittest.TestCase):
        cluster = self.etcd.get_cluster()
        self.assertIsInstance(cluster, Cluster)
        self.assertIsInstance(cluster.workers[1], Cluster)
+        self.etcd._base_path = '/service/nocluster'
+        self.assertTrue(self.etcd.get_cluster().is_empty())

    def test_touch_member(self):
        self.assertFalse(self.etcd.touch_member(''))
@@ -1532,7 +1532,7 @@ class TestHa(PostgresInit):
        self.ha.is_leader = true

        def stop(*args, **kwargs):
-            kwargs['on_shutdown'](123)
+            kwargs['on_shutdown'](123, 120)

        self.p.stop = stop
        self.ha.shutdown()
@@ -1581,6 +1581,11 @@ class TestHa(PostgresInit):
        self.p.is_primary = false
        self.ha.run_cycle()
        exit_mock.assert_called_once_with(1)
+        self.p.set_role('replica')
+        self.ha.dcs.initialize = Mock()
+        with patch.object(Postgresql, 'cb_called', PropertyMock(return_value=True)):
+            self.assertEqual(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock')
+        self.ha.dcs.initialize.assert_not_called()

    @patch.object(Cluster, 'is_unlocked', Mock(return_value=False))
    def test_after_pause(self):
@@ -45,7 +45,7 @@ class MockFrozenImporter(object):
@patch('time.sleep', Mock())
@patch('subprocess.call', Mock(return_value=0))
@patch('patroni.psycopg.connect', psycopg_connect)
-@patch('urllib3.connection.HTTPConnection.connect', Mock(side_effect=Exception))
+@patch('urllib3.PoolManager.request', Mock(side_effect=Exception))
@patch.object(ConfigHandler, 'append_pg_hba', Mock())
@patch.object(ConfigHandler, 'write_postgresql_conf', Mock())
@patch.object(ConfigHandler, 'write_recovery_conf', Mock())
@@ -69,7 +69,7 @@ class TestPatroni(unittest.TestCase):
            self.assertRaises(SystemExit, _main)

    @patch('pkgutil.iter_importers', Mock(return_value=[MockFrozenImporter()]))
-    @patch('urllib3.connection.HTTPConnection.connect', Mock(side_effect=Exception))
+    @patch('urllib3.PoolManager.request', Mock(side_effect=Exception))
    @patch('sys.frozen', Mock(return_value=True), create=True)
    @patch.object(HTTPServer, '__init__', Mock())
    @patch.object(etcd.Client, 'read', etcd_read)
@@ -174,6 +174,20 @@ class TestPatroni(unittest.TestCase):
        self.p.next_run = time.time() - self.p.dcs.loop_wait - 1
        self.p.schedule_next_run()

+    def test__filter_tags(self):
+        tags = {'noloadbalance': False, 'clonefrom': False, 'nosync': False, 'smth': 'random'}
+        self.assertEqual(self.p._filter_tags(tags), {'smth': 'random'})
+
+        tags['clonefrom'] = True
+        tags['smth'] = False
+        self.assertEqual(self.p._filter_tags(tags), {'clonefrom': True, 'smth': False})
+
+        tags = {'nofailover': False, 'failover_priority': 0}
+        self.assertEqual(self.p._filter_tags(tags), tags)
+
+        tags = {'nofailover': True, 'failover_priority': 1}
+        self.assertEqual(self.p._filter_tags(tags), tags)
+
    def test_noloadbalance(self):
        self.p.tags['noloadbalance'] = True
        self.assertTrue(self.p.noloadbalance)
@@ -185,9 +199,11 @@ class TestPatroni(unittest.TestCase):
            # Setting `nofailover: True` has precedence
            (True, 0, True),
            (True, 1, True),
+            ('False', 1, True),  # because we use bool() for the value
            # Similarly, setting `nofailover: False` has precedence
            (False, 0, False),
            (False, 1, False),
+            ('', 0, False),
            # Only when we have `nofailover: None` should we got based on priority
            (None, 0, True),
            (None, 1, False),
@@ -273,8 +289,8 @@ class TestPatroni(unittest.TestCase):
        )
        with patch('patroni.dcs.AbstractDCS.get_cluster', Mock(return_value=bad_cluster)):
            # If the api of the running node cannot be reached, this implies unique name
-            with patch('urllib3.connection.HTTPConnection.connect', Mock(side_effect=ConnectionError)):
+            with patch('urllib3.PoolManager.request', Mock(side_effect=ConnectionError)):
                self.assertIsNone(self.p.ensure_unique_name())
            # Only if the api of the running node is reachable do we throw an error
-            with patch('urllib3.connection.HTTPConnection.connect', Mock()):
+            with patch('urllib3.PoolManager.request', Mock()):
                self.assertRaises(SystemExit, self.p.ensure_unique_name)
@@ -5,6 +5,7 @@ import re
 import subprocess
 import time

+from copy import deepcopy
 from mock import Mock, MagicMock, PropertyMock, patch, mock_open

 import patroni.psycopg as psycopg
@@ -17,6 +18,7 @@ from patroni.exceptions import PostgresConnectionException, PatroniException
 from patroni.postgresql import Postgresql, STATE_REJECT, STATE_NO_RESPONSE
 from patroni.postgresql.bootstrap import Bootstrap
 from patroni.postgresql.callback_executor import CallbackAction
+from patroni.postgresql.config import _false_validator
 from patroni.postgresql.postmaster import PostmasterProcess
 from patroni.postgresql.validator import (ValidatorFactoryNoType, ValidatorFactoryInvalidType,
                                          ValidatorFactoryInvalidSpec, ValidatorFactory, InvalidGucValidatorsFile,
@@ -25,7 +27,8 @@ from patroni.postgresql.validator import (ValidatorFactoryNoType, ValidatorFacto
 from patroni.utils import RetryFailedError
 from threading import Thread, current_thread

-from . import BaseTestPostgresql, MockCursor, MockPostmaster, psycopg_connect, mock_available_gucs
+from . import (BaseTestPostgresql, MockCursor, MockPostmaster, psycopg_connect, mock_available_gucs,
+               GET_PG_SETTINGS_RESULT)


 mtime_ret = {}
@@ -237,7 +240,10 @@ class TestPostgresql(BaseTestPostgresql):
    @patch.object(Postgresql, 'latest_checkpoint_location', Mock(return_value='7'))
    def test__do_stop(self):
        mock_callback = Mock()
-        with patch.object(Postgresql, 'controldata', Mock(return_value={'Database cluster state': 'shut down'})):
+        with patch.object(Postgresql, 'controldata',
+                          Mock(return_value={'Database cluster state': 'shut down',
+                                             "Latest checkpoint's TimeLineID": '1',
+                                             'Latest checkpoint location': '1/1'})):
            self.assertTrue(self.p.stop(on_shutdown=mock_callback, stop_timeout=3))
            mock_callback.assert_called()
        with patch.object(Postgresql, 'controldata',
@@ -556,31 +562,111 @@ class TestPostgresql(BaseTestPostgresql):

    @patch('time.sleep', Mock())
    @patch.object(Postgresql, 'is_running', Mock(return_value=True))
-    def test_reload_config(self):
-        parameters = self._PARAMETERS.copy()
-        parameters.pop('f.oo')
-        parameters['wal_buffers'] = '512'
-        config = {'pg_hba': [''], 'pg_ident': [''], 'use_unix_socket': True, 'use_unix_socket_repl': True,
-                  'authentication': {},
-                  'retry_timeout': 10, 'listen': '*', 'krbsrvname': 'postgres', 'parameters': parameters}
+    @patch('patroni.postgresql.config.logger.info')
+    @patch('patroni.postgresql.config.logger.warning')
+    def test_reload_config(self, mock_warning, mock_info):
+        config = deepcopy(self.p.config._config)
+
+        # Nothing changed
        self.p.reload_config(config)
-        parameters['b.ar'] = 'bar'
-        with patch.object(MockCursor, 'fetchall',
-                          Mock(side_effect=[[('wal_block_size', '8191', None, 'integer', 'internal'),
-                                             ('wal_segment_size', '2048', '8kB', 'integer', 'internal'),
-                                             ('shared_buffers', '16384', '8kB', 'integer', 'postmaster'),
-                                             ('wal_buffers', '-1', '8kB', 'integer', 'postmaster'),
-                                             ('port', '5433', None, 'integer', 'postmaster')], Exception])):
+        mock_info.assert_called_once_with('No PostgreSQL configuration items changed, nothing to reload.')
+        mock_warning.assert_not_called()
+        self.assertEqual(self.p.pending_restart, False)
+
+        mock_info.reset_mock()
+
+        # Ignored params changed
+        config['parameters']['archive_cleanup_command'] = 'blabla'
+        self.p.reload_config(config)
+        mock_info.assert_called_once_with('No PostgreSQL configuration items changed, nothing to reload.')
+        self.assertEqual(self.p.pending_restart, False)
+
+        mock_info.reset_mock()
+
+        # Handle wal_buffers
+        self.p.config._config['parameters']['wal_buffers'] = '512'
+        self.p.reload_config(config)
+        mock_info.assert_called_once_with('No PostgreSQL configuration items changed, nothing to reload.')
+        self.assertEqual(self.p.pending_restart, False)
+
+        mock_info.reset_mock()
+        config = deepcopy(self.p.config._config)
+
+        # hba/ident_changed
+        config['pg_hba'] = ['']
+        config['pg_ident'] = ['']
+        self.p.reload_config(config)
+        mock_info.assert_called_once_with('Reloading PostgreSQL configuration.')
+        self.assertEqual(self.p.pending_restart, False)
+
+        mock_info.reset_mock()
+
+        # Postmaster parameter change (pending_restart)
+        init_max_worker_processes = config['parameters']['max_worker_processes']
+        config['parameters']['max_worker_processes'] *= 2
+        with patch('patroni.postgresql.Postgresql._query', Mock(side_effect=[GET_PG_SETTINGS_RESULT, [(1,)]])):
            self.p.reload_config(config)
-        parameters['autovacuum'] = 'on'
+            self.assertEqual(mock_info.call_args_list[0][0], ('Changed %s from %s to %s (restart might be required)',
+                                                              'max_worker_processes', str(init_max_worker_processes),
+                                                              config['parameters']['max_worker_processes']))
+            self.assertEqual(mock_info.call_args_list[1][0], ('Reloading PostgreSQL configuration.',))
+            self.assertEqual(self.p.pending_restart, True)
+
+        mock_info.reset_mock()
+
+        # Reset to the initial value without restart
+        config['parameters']['max_worker_processes'] = init_max_worker_processes
        self.p.reload_config(config)
-        parameters['autovacuum'] = 'off'
-        parameters.pop('search_path')
-        config['listen'] = '*:5433'
+        self.assertEqual(mock_info.call_args_list[0][0], ('Changed %s from %s to %s', 'max_worker_processes',
+                                                          init_max_worker_processes * 2,
+                                                          str(config['parameters']['max_worker_processes'])))
+        self.assertEqual(mock_info.call_args_list[1][0], ('Reloading PostgreSQL configuration.',))
+        self.assertEqual(self.p.pending_restart, False)
+
+        mock_info.reset_mock()
+
+        # User-defined parameter changed (removed)
+        config['parameters'].pop('f.oo')
        self.p.reload_config(config)
-        parameters['unix_socket_directories'] = '.'
+        self.assertEqual(mock_info.call_args_list[0][0], ('Changed %s from %s to %s', 'f.oo', 'bar', None))
+        self.assertEqual(mock_info.call_args_list[1][0], ('Reloading PostgreSQL configuration.',))
+        self.assertEqual(self.p.pending_restart, False)
+
+        mock_info.reset_mock()
+
+        # Non-postmaster parameter change
+        config['parameters']['autovacuum'] = 'off'
        self.p.reload_config(config)
-        self.p.config.resolve_connection_addresses()
+        self.assertEqual(mock_info.call_args_list[0][0], ("Changed %s from %s to %s", 'autovacuum', 'on', 'off'))
+        self.assertEqual(mock_info.call_args_list[1][0], ('Reloading PostgreSQL configuration.',))
+        self.assertEqual(self.p.pending_restart, False)
+
+        config['parameters']['autovacuum'] = 'on'
+        mock_info.reset_mock()
+
+        # Remove invalid parameter
+        config['parameters']['invalid'] = 'value'
+        self.p.reload_config(config)
+        self.assertEqual(mock_warning.call_args_list[0][0],
+                         ('Removing invalid parameter `%s` from postgresql.parameters', 'invalid'))
+        config['parameters'].pop('invalid')
+
+        mock_warning.reset_mock()
+        mock_info.reset_mock()
+
+        # Non-empty result (outside changes) and exception while querying pending_restart parameters
+        with patch('patroni.postgresql.Postgresql._query',
+                   Mock(side_effect=[GET_PG_SETTINGS_RESULT, [(1,)], GET_PG_SETTINGS_RESULT, Exception])):
+            self.p.reload_config(config, True)
+            self.assertEqual(mock_info.call_args_list[0][0], ('Reloading PostgreSQL configuration.',))
+            self.assertEqual(self.p.pending_restart, True)
+
+            # Invalid values, just to increase silly coverage in postgresql.validator.
+            # One day we will have proper tests there.
+            config['parameters']['autovacuum'] = 'of'  # Bool.transform()
+            config['parameters']['vacuum_cost_limit'] = 'smth'  # Number.transform()
+            self.p.reload_config(config, True)
+            self.assertEqual(mock_warning.call_args_list[-1][0][0], 'Exception %r when running query')

    def test_resolve_connection_addresses(self):
        self.p.config._config['use_unix_socket'] = self.p.config._config['use_unix_socket_repl'] = True
@@ -723,21 +809,28 @@ class TestPostgresql(BaseTestPostgresql):
    @patch.object(Postgresql, 'get_postgres_role_from_data_directory', Mock(return_value='replica'))
    @patch.object(Postgresql, 'is_running', Mock(return_value=False))
    @patch.object(Bootstrap, 'running_custom_bootstrap', PropertyMock(return_value=True))
-    @patch.object(Postgresql, 'controldata', Mock(return_value={'max_connections setting': '200',
-                                                                'max_worker_processes setting': '20',
-                                                                'max_locks_per_xact setting': '100',
-                                                                'max_wal_senders setting': 10}))
-    @patch('patroni.postgresql.config.logger.warning')
+    @patch('patroni.postgresql.config.logger')
    def test_effective_configuration(self, mock_logger):
-        self.p.cancellable.cancel()
-        self.p.config.write_recovery_conf({'pause_at_recovery_target': 'false'})
-        self.assertFalse(self.p.start())
-        mock_logger.assert_called_once()
-        self.assertTrue('is missing from pg_controldata output' in mock_logger.call_args[0][0])
+        controldata = {'max_connections setting': '100', 'max_worker_processes setting': '8',
+                       'max_locks_per_xact setting': '64', 'max_wal_senders setting': 5}

-        self.assertTrue(self.p.pending_restart)
-        with patch.object(Bootstrap, 'keep_existing_recovery_conf', PropertyMock(return_value=True)):
+        with patch.object(Postgresql, 'controldata', Mock(return_value=controldata)), \
+             patch.object(Bootstrap, 'keep_existing_recovery_conf', PropertyMock(return_value=True)):
+            self.p.cancellable.cancel()
            self.assertFalse(self.p.start())
+            self.assertFalse(self.p.pending_restart)
+            mock_logger.warning.assert_called_once()
+            self.assertEqual(mock_logger.warning.call_args[0],
+                             ('%s is missing from pg_controldata output', 'max_prepared_xacts setting'))
+
+        mock_logger.reset_mock()
+        controldata['max_prepared_xacts setting'] = 0
+        controldata['max_wal_senders setting'] *= 2
+
+        with patch.object(Postgresql, 'controldata', Mock(return_value=controldata)):
+            self.p.config.write_recovery_conf({'pause_at_recovery_target': 'false'})
+            self.assertFalse(self.p.start())
+            mock_logger.warning.assert_not_called()
            self.assertTrue(self.p.pending_restart)

    @patch('os.path.exists', Mock(return_value=True))
@@ -981,3 +1074,14 @@ class TestPostgresql2(BaseTestPostgresql):
        self.assertIn('diff(pg_catalog.pg_current_xlog_flush_location(', self.p.cluster_info_query)
        self.p._major_version = 90500
        self.assertIn('diff(pg_catalog.pg_current_xlog_location(', self.p.cluster_info_query)
+
+    @patch.object(Postgresql, 'is_primary', Mock(return_value=False))
+    @patch.object(Postgresql, '_query', Mock(return_value=[('primary_conninfo', 'host=a port=5433 passfile=/blabla')]))
+    def test_load_current_server_parameters(self):
+        keep_values = {name: self.p.config._server_parameters[name]
+                       for name, value in self.p.config.CMDLINE_OPTIONS.items() if value[1] == _false_validator}
+        self.p.config.load_current_server_parameters()
+        self.assertTrue(all(self.p.config._server_parameters[name] == value for name, value in keep_values.items()))
+        self.assertEqual(dict(self.p.config._recovery_params),
+                         {'primary_conninfo': {'host': 'a', 'port': '5433', 'passfile': '/blabla',
+                          'gssencmode': 'prefer', 'sslmode': 'prefer', 'channel_binding': 'prefer'}})
@@ -180,6 +180,16 @@ class TestRewind(BaseTestPostgresql):
            self.r.trigger_check_diverged_lsn()
            mock_get_local_timeline_lsn.return_value = (False, 2, 67197377)
            self.assertTrue(self.r.rewind_or_reinitialize_needed_and_possible(self.leader))
+
+            mock_popen.return_value.communicate.return_value = (
+                b'0, lsn: 0/040159C1, prev 0/\n',
+                b'pg_waldump: fatal: error in WAL record at 0/40159C1: invalid record '
+                b'length at 0/402DD98: expected at least 24, got 0\n'
+            )
+            self.r.reset_state()
+            self.r.trigger_check_diverged_lsn()
+            self.assertFalse(self.r.rewind_or_reinitialize_needed_and_possible(self.leader))
+
            self.r.reset_state()
            self.r.trigger_check_diverged_lsn()
            mock_popen.side_effect = Exception
@@ -1,4 +1,4 @@
-from typing import Any
+from typing import Any, MutableMapping
 class HTTPHeaderDict(MutableMapping[str, str]):
    def __init__(self, headers=None, **kwargs) -> None: ...
    def __setitem__(self, key, val) -> None: ...
Author	SHA1	Message	Date
Alexander KukushkinandPolina Bungina	c8e32775df	Release v3.2.2 (#3007 ) - update release notes - bump Patroni version - bump pyright version and fix reported issues - improve compatibility with legacy psycopg2 Co-authored-by: Polina Bungina <[email protected]>	2024-01-17 08:35:35 +01:00
Polina BunginaandAlexander Kukushkin	f2919f9c2f	Fixes around pending_restart flag (#3003 ) * Do not set pending_restart flag if hot_standby is set to 'off' during a custom bootstrap (even though we will have this flag actually set in PG, this configuration parameter is irrelevant on primary and there is no actual need for restart) * Skip hot_standby and wal_log_hints when querying parameters pending restart on config reload. They actually can be changed manually (e.g. via ALTER SYSTEM) and it will cause the pending_restart state in PG but Patroni anyway always passes those params to postmaster as command line options. And there they only can have one value - 'on' (except on primary when performing custom bootstrap)	2024-01-16 10:44:30 +01:00
Alexander Kukushkin	f59c79740f	Optimize priority failover behave tests (#3004 ) 1. get rid of useless sleep calls 2. call `POST /failover` on the node where we want to failover to	2024-01-15 12:24:42 +01:00
Alexander Kukushkin	2a64bfd459	Restore recovery GUCs when joining running standby (#2998 ) Close https://github.com/zalando/patroni/issues/2993	2024-01-08 09:17:17 +01:00
IsraelandAlexander Kukushkin	23067d7ea7	Close the doors for a possible future bug in the config generator (#3000 ) The `AbstractConfigGenerator._format_config` method was missing a comma in the declaration of a tuple. As a consequence it was concatenating the strings `ctl` and `citus` instead of creating two separate items in the tuple. There is currently no observed bug from that issue in the code because the template configuration created by the method `AbstractConfigGenerator.get_template_config` doesn't include either of `ctl` or `citus` keys. However, it is still important that we close the doors for possible future bugs that would come up if we ever attempt to use either of those keys in the template, for example. References: PAT-231.	2024-01-05 10:17:07 +01:00
Sophia RuanandAlexander Kukushkin	47063de46d	call freeze_support in main module to solve pyinstaller frozen issue (#2996 ) Close #2995	2024-01-05 10:17:01 +01:00
Polina BunginaandAlexander Kukushkin	3e9bceac11	Don't filter out contradictory nofailover tag (#2992 ) * Ensure that nofailover will always be used if both nofailover and failover_priority tags are provided * Call _validate_failover_tags from reload_local_configuration() as well * Properly check values in the _validate_failover_tags(): nofailover value should be casted to boolean like it is done when accessed in other places	2024-01-05 10:16:52 +01:00
zhjwpkuandAlexander Kukushkin	9cc1f8e763	Fix Citus bootstrap - CREATE DATABASE cannot be executed from a function (#2994 ) This was introduced by #2990: pod cannot be started and show the following logs: ``` 2023-12-26 03:29:25.569 UTC [47] CONTEXT: SQL statement "CREATE DATABASE "citus"" PL/pgSQL function inline_code_block line 5 at SQL statement 2023-12-26 03:29:25.569 UTC [47] STATEMENT: DO $$ BEGIN PERFORM * FROM pg_catalog.pg_database WHERE datname = 'citus'; IF NOT FOUND THEN CREATE DATABASE "citus"; END IF; END;$$ 2023-12-26 03:29:25,570 ERROR: post_bootstrap Traceback (most recent call last): File "/usr/local/lib/python3.11/dist-packages/patroni/postgresql/bootstrap.py", line 474, in post_bootstrap self._postgresql.citus_handler.bootstrap() File "/usr/local/lib/python3.11/dist-packages/patroni/postgresql/mpp/citus.py", line 401, in bootstrap cur.execute(sql.encode('utf-8')) psycopg2.errors.ActiveSqlTransaction: CREATE DATABASE cannot be executed from a function CONTEXT: SQL statement "CREATE DATABASE "citus"" PL/pgSQL function inline_code_block line 5 at SQL statement ``` --------- Signed-off-by: Zhao Junwang <[email protected]>	2024-01-05 10:16:26 +01:00
Alexander Kukushkin	d00f5a645b	Create citus database and extension idempotently (#2990 ) Consider a task: we want to create an extension _before_ citus in a database. Currently `post_bootstrab` script is executed before `CitusHandler.bootstrap()` method, which seems to allow doing that, but in fact `CitusHandler.bootstrap()` will fail to create already existing database and as a result the whole bootstrap will fail. Changing the order of execution of `post_bootstrab` hook and `CitusHandler.bootstrap()` seems to be useless, because it will not allow creating another extension _before_ citus. Therefore the only way of solving it is making CREATE DATABASE and CREATE EXTENSION idempotent. It will allow to create citus database and all dependencies from the `post_bootstrab` hook.	2024-01-05 10:14:45 +01:00
Polina BunginaandAlexander Kukushkin	15b57c5bdc	Exclude leader from failover candidates in ctl (#2983 ) Exclude actual leader (not the passed leader argument) from the candidates list in the `patronictl failover` prompt. Abort `patronictl failover` execution if candidate specified is the same as the current cluster leader	2024-01-05 10:12:33 +01:00
Polina BunginaandAlexander Kukushkin	f10e4805db	Actually allow failover to an async candidate in sync mode (#2980 )	2024-01-05 10:05:28 +01:00
Polina BunginaandAlexander Kukushkin	3e0e91f905	Reload postgres config if a server param was reset (#2975 ) Fix the case when a parameter value was changed and then reset back to the initial value without restart - before this fix, the second change was not reflected in the Postgres config. This commit also includes the related unit test refactoring.	2024-01-05 09:56:01 +01:00
Alexander Kukushkin	51a148fcf3	Use consistent read when fetching just updated sync key (#2974 ) Consul doesn't provide any interface to immediately get `ModifyIndex` for the key that we just updated, therefore we have to perform an explicit read operation. By default stale reads are allowed and sometimes we may read stale data. As a result write_sync_state() call was considered as failed. To mitigate the problem we switch to `consistent` reads when that executed after update of the `/sync` key. Close #2972	2024-01-05 09:55:43 +01:00
Alexander Kukushkin	c3697738b1	Disable SSL for MacOS GH action runners (#2976 ) Latest runners release (20231127.1) somehow broke our tests. Connections to postgres somehow failing with strange error: ``` could not accept SSL connection: Socket operation on non-socket ```	2024-01-05 09:55:27 +01:00
Alexander Kukushkin	722b4b72a8	Don't let replica restore initialize key when DCS was wiped (#2970 ) It was happening from the branch where Patroni was supposed to be complain about converting standalone PG cluster to be governed by Patroni and exit.	2024-01-05 09:55:10 +01:00
Alexander Kukushkin	65b43c39fa	Release/v3.2.1 (#2968 ) - bump version - bump pyright - update release notes	2023-11-30 16:51:21 +01:00
WaynervandAlexander Kukushkin	aea9a2b0ca	Cache `postgres --describe-config` output results (#2967 ) We don't expect GUCs list to change for the same major version and don't expect major version to change while Patroni is running.	2023-11-30 12:07:06 +01:00
Sophia RuanandAlexander Kukushkin	71ccd41915	Fix the issue that REST API returns unknown after postgres restart (#2956 ) Close #2955	2023-11-30 10:16:51 +01:00
Alexander Kukushkin	49e4a6ed7d	Fix Citus transaction rollback condition check (#2964 ) It seems that sometimes we get an exact match, what makes behave tests to fail.	2023-11-30 09:02:50 +01:00
Alexander Kukushkin	ebd05871d9	Bump pyright to 1.1.336 (#2952 ) and fix newly reported issues	2023-11-30 09:02:16 +01:00
Alexander Kukushkin	42cd803619	Fix bug with custom bootstrap (#2948 ) Patroni was falsely applying `--command` argument. Close https://github.com/zalando/patroni/issues/2947	2023-11-30 09:01:47 +01:00
Alexander Kukushkin	bae72df5b1	Fix pg_rewind behavior with Postgres v16+ (#2944 ) The error message format was changed in https://github.com/postgres/postgres/commit/4ac30ba4f29d4b586b131404b0d514f16501272a, what caused `pg_rewind` being called by Patroni even when it was not necessary.	2023-11-30 09:01:41 +01:00
Alexander Kukushkin	f2a129f209	Fix Etcd v2 with Citus (#2943 ) When deploying a new Citus cluster with Etcd v2 Patroni was failing to start with the following exception: ```python 2023-11-09 10:51:41,246 INFO: Selected new etcd server http://localhost:2379 Traceback (most recent call last): File "/home/akukushkin/git/patroni/./patroni.py", line 6, in <module> main() File "/home/akukushkin/git/patroni/patroni/__main__.py", line 343, in main return patroni_main(args.configfile) File "/home/akukushkin/git/patroni/patroni/__main__.py", line 237, in patroni_main abstract_main(Patroni, configfile) File "/home/akukushkin/git/patroni/patroni/daemon.py", line 172, in abstract_main controller = cls(config) File "/home/akukushkin/git/patroni/patroni/__main__.py", line 66, in __init__ self.ensure_unique_name() File "/home/akukushkin/git/patroni/patroni/__main__.py", line 112, in ensure_unique_name cluster = self.dcs.get_cluster() File "/home/akukushkin/git/patroni/patroni/dcs/__init__.py", line 1654, in get_cluster cluster = self._get_citus_cluster() if self.is_citus_coordinator() else self.__get_patroni_cluster() File "/home/akukushkin/git/patroni/patroni/dcs/__init__.py", line 1638, in _get_citus_cluster cluster = groups.pop(CITUS_COORDINATOR_GROUP_ID, Cluster.empty()) AttributeError: 'Cluster' object has no attribute 'pop' ``` It is broken since #2909. In addition to that fix `_citus_cluster_loader()` interface by allowing it to return only dict obj.	2023-11-30 09:01:19 +01:00
Alexander Kukushkin	df0fd91614	Do a real http request when performing name uniqueness check (#2942 ) When running in containers it is possible that the traffic is routed using `docker-proxy`, which listens on the port and accepting incoming connections. This commit effectively sticks to the original solution from #2878	2023-11-30 09:01:11 +01:00
Alexander Kukushkin	43f23df974	Verify that replica nodes received checkpoint LSN on shutdown (#2939 ) In case if archiving is enabled the `Postgresql.latest_checkpoint_location()` method returns LSN of the prev (SWITCH) record, which points to the beginning of the WAL file. It is done in order to make it possible to safely promote replica which recovers WAL files from the archive and wasn't streaming when the primary was stopped (primary doesn't archive this WAL file). But, in certain cases using the LSN pointing to SWITCH record was causing unnecessary pg_rewind, if replica didn't managed to replay shutdown checkpoint record before it was promoted. In order to mitigate the problem we need to check that replica received/replayed exactly the shutdown checkpoint LSN. But, at the same time we will still write LSN of the SWITCH record to the `/status` key when releasing the leader lock.	2023-11-30 09:01:05 +01:00
Alexander Kukushkin	42bf1f95a3	Limit accepted values for --format argument (#2938 ) It used to accept any arbitrary string Close https://github.com/zalando/patroni/issues/2936	2023-11-30 09:00:39 +01:00
IsraelandAlexander Kukushkin	23200daada	Add a FAQ page to the docs (#2933 ) This commit introduces a FAQ page to the docs. The idea is to get most frequently asked questions answered before-hand, so the user is able to get them answered quickly without going into detail in the docs or having to go to Slack/GitHub to clarify questions. --------- Signed-off-by: Israel Barth Rubio <[email protected]>	2023-11-30 09:00:22 +01:00