Merge branch 'release/v3.3.0' of github.com:zalando/patroni into feature/citus-secondaries

Most of them are related to partially unknown types of values from empty
dict or list. To solve it for the empty dict we use `EMPTY_DICT` object of
newly introduced `_FrozenDict` class.

.github/workflows/tests.yaml

@@ -174,7 +174,7 @@ jobs:
 
     - uses: jakebailey/pyright-action@v1
       with:
-        version: 1.1.347
+        version: 1.1.356
 
   docs:
     runs-on: ubuntu-latest

docs/releases.rst

@@ -3,6 +3,67 @@
 Release notes
 =============
 
+Version 3.3.0
+-------------
+
+**New features**
+
+- Add ability to pass ``auth_data`` to Zookeeper client (Aras Mumcuyan)
+
+  It allows to specify the authentication credentials to use for the connection.
+
+- Add a contrib script for ``Barman`` integration (Israel Barth Rubio)
+
+  Provide an application ``patroni_barman`` that allows to perform ``Barman`` operations remotely and can be used as a custom bootstrap/custom replica method or as an ``on_role_change`` callback. Please check :ref:`here <tools_integration>` for more information.
+
+- Support JSON log format (alisalemmi)
+
+  Apart from ``plain``, Patroni now also supports ``json`` log format. Requires ``python-json-logger`` library to be installed.
+
+- Show ``pending_restart_reason`` information (Polina Bungina)
+
+  Provide extended information about the PostgreSQL parameters that caused ``pending_restart`` flag to be set. Both ``patronictl list`` and ``/patroni`` REST API endpoint now show the parameters names and their "diff" as ``pending_restart_reason``.
+
+- Implement ``nostream`` tag (Grigory Smolkin)
+
+  If ``nostream`` tag is set to ``true``, the node will not use replication protocol to stream WAL but instead rely on archive recovery (if ``restore_command`` is configured). It also disables copying and synchronization of permanent logical replication slots on the node itself and all its cascading replicas.
+
+
+**Improvements**
+
+- Implement validation of the log section (Alexander Kukushkin)
+
+  Until now validator was not checking the correctness of the logging configuration provided.
+
+- Improve logging for PostgreSQL parameters change (Polina Bungina)
+
+  Convert old values to a human-readable format and log information about the ``pg_controldata`` vs Patroni global configuration mismatch.
+
+
+**Bugfixes**
+
+- Properly filter out not allowed ``pg_basebackup`` options (Israel Barth Rubio)
+
+  Due to a bug, Patroni was not properly filtering out the not allowed options configured for the ``basebackup`` replica bootstrap method, when provided in the ``- setting: value`` format.
+
+- Fix ``etcd3`` authentication error handling (Alexander Kukushkin)
+
+  Always retry one time on ``etcd3`` authentication error if authentication was not done right before executing the request. Also, do not restart watchers on reauthentication.
+
+- Improve logic of the validator files discovery (Waynerv)
+
+  Use ``importlib`` library to discover the files with available configuration parameters when possible (for Python 3.9+). This implementation is more stable and doesn't break the Patroni distributions based on ``zip`` archives.
+
+- Use ``target_session_attrs`` only when multiple hosts are specified in the ``standby_cluster`` section (Alexander Kukushkin)
+
+  ``target_session_attrs=read-write`` is now added to the ``primary_conninfo`` on the standby leader node only when ``standby_cluster.host`` section contains multiple hosts separated by commas.
+
+- Add compatibility code for ``ydiff`` library version 1.3+ (Alexander Kukushkin)
+
+.. warning::
+    All older Partoni versions are not compatible with ``ydiff`` 1.3+. Please upgrade Patroni, use ``ydiff`` version <1.3, or install ``cdiff``.
+
+
 Version 3.2.2
 -------------
 

docs/tools_integration.rst

@@ -1,3 +1,5 @@
+.. _tools_integration:
+
 Integration with other tools
 ============================
 

features/citus.feature

@@ -11,7 +11,9 @@ Feature: citus
     Then replication works from postgres0 to postgres1 after 15 seconds
     Then replication works from postgres2 to postgres3 after 15 seconds
     And postgres0 is registered in the postgres0 as the primary in group 0 after 5 seconds
+    And postgres1 is registered in the postgres0 as the secondary in group 0 after 5 seconds
     And postgres2 is registered in the postgres0 as the primary in group 1 after 5 seconds
+    And postgres3 is registered in the postgres0 as the secondary in group 1 after 5 seconds
 
   Scenario: coordinator failover updates pg_dist_node
     Given I run patronictl.py failover batman --group 0 --candidate postgres1 --force
@@ -19,11 +21,13 @@ Feature: citus
     And "members/postgres0" key in a group 0 in DCS has state=running after 15 seconds
     And replication works from postgres1 to postgres0 after 15 seconds
     And postgres1 is registered in the postgres2 as the primary in group 0 after 5 seconds
+    And postgres0 is registered in the postgres2 as the secondary in group 0 after 15 seconds
     And "sync" key in a group 0 in DCS has sync_standby=postgres0 after 15 seconds
     When I run patronictl.py switchover batman --group 0 --candidate postgres0 --force
     Then postgres0 role is the primary after 10 seconds
     And replication works from postgres0 to postgres1 after 15 seconds
     And postgres0 is registered in the postgres2 as the primary in group 0 after 5 seconds
+    And postgres1 is registered in the postgres2 as the secondary in group 0 after 15 seconds
     And "sync" key in a group 0 in DCS has sync_standby=postgres1 after 15 seconds
 
   Scenario: worker switchover doesn't break client queries on the coordinator
@@ -35,6 +39,7 @@ Feature: citus
     And "members/postgres2" key in a group 1 in DCS has state=running after 15 seconds
     And replication works from postgres3 to postgres2 after 15 seconds
     And postgres3 is registered in the postgres0 as the primary in group 1 after 5 seconds
+    And postgres2 is registered in the postgres0 as the secondary in group 1 after 15 seconds
     And "sync" key in a group 1 in DCS has sync_standby=postgres2 after 15 seconds
     And a thread is still alive
     When I run patronictl.py switchover batman --group 1 --force
@@ -42,6 +47,7 @@ Feature: citus
     And postgres2 role is the primary after 10 seconds
     And replication works from postgres2 to postgres3 after 15 seconds
     And postgres2 is registered in the postgres0 as the primary in group 1 after 5 seconds
+    And postgres3 is registered in the postgres0 as the secondary in group 1 after 15 seconds
     And "sync" key in a group 1 in DCS has sync_standby=postgres3 after 15 seconds
     And a thread is still alive
     When I stop a thread
@@ -55,6 +61,7 @@ Feature: citus
     And postgres2 role is the primary after 10 seconds
     And replication works from postgres2 to postgres3 after 15 seconds
     And postgres2 is registered in the postgres0 as the primary in group 1 after 5 seconds
+    And postgres3 is registered in the postgres0 as the secondary in group 1 after 15 seconds
     And a thread is still alive
     When I stop a thread
     Then a distributed table on postgres0 has expected rows

patroni/collections.py

@@ -1,9 +1,10 @@
 """Patroni custom object types somewhat like :mod:`collections` module.
 
-Provides a case insensitive :class:`dict` and :class:`set` object types.
+Provides a case insensitive :class:`dict` and :class:`set` object types, and `EMPTY_DICT` frozen dictionary object.
 """
 from collections import OrderedDict
-from typing import Any, Collection, Dict, Iterator, KeysView, MutableMapping, MutableSet, Optional
+from copy import deepcopy
+from typing import Any, Collection, Dict, Iterator, KeysView, Mapping, MutableMapping, MutableSet, Optional
 
 
 class CaseInsensitiveSet(MutableSet[str]):
@@ -48,7 +49,7 @@ class CaseInsensitiveSet(MutableSet[str]):
         """
         return str(set(self._values.values()))
 
-    def __contains__(self, value: str) -> bool:
+    def __contains__(self, value: object) -> bool:
         """Check if set contains *value*.
 
         The check is performed case-insensitively.
@@ -57,7 +58,7 @@ class CaseInsensitiveSet(MutableSet[str]):
 
         :returns: ``True`` if *value* is already in the set, ``False`` otherwise.
         """
-        return value.lower() in self._values
+        return isinstance(value, str) and value.lower() in self._values
 
     def __iter__(self) -> Iterator[str]:
         """Iterate over the values in this set.
@@ -207,3 +208,47 @@ class CaseInsensitiveDict(MutableMapping[str, Any]):
             "<CaseInsensitiveDict{'A': 'B', 'c': 'd'} at ..."
         """
         return '<{0}{1} at {2:x}>'.format(type(self).__name__, dict(self.items()), id(self))
+
+
+class _FrozenDict(Mapping[str, Any]):
+    """Frozen dictionary object."""
+
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        """Create a new instance of :class:`_FrozenDict` with given data."""
+        self.__values: Dict[str, Any] = dict(*args, **kwargs)
+
+    def __iter__(self) -> Iterator[str]:
+        """Iterate over keys of this dict.
+
+        :yields: each key present in the dict. Yields each key with its last case that has been stored.
+        """
+        return iter(self.__values)
+
+    def __len__(self) -> int:
+        """Get the length of this dict.
+
+        :returns: number of keys in the dict.
+
+        :Example:
+
+        >>> len(_FrozenDict())
+        0
+        """
+        return len(self.__values)
+
+    def __getitem__(self, key: str) -> Any:
+        """Get the value corresponding to *key*.
+
+        :returns: value corresponding to *key*.
+        """
+        return self.__values[key]
+
+    def copy(self) -> Dict[str, Any]:
+        """Create a copy of this dict.
+
+        :return: a new dict object with the same keys and values of this dict.
+        """
+        return deepcopy(self.__values)
+
+
+EMPTY_DICT = _FrozenDict()

patroni/config.py

@@ -12,7 +12,7 @@ from copy import deepcopy
 from typing import Any, Callable, Collection, Dict, List, Optional, Union, TYPE_CHECKING
 
 from . import PATRONI_ENV_PREFIX
-from .collections import CaseInsensitiveDict
+from .collections import CaseInsensitiveDict, EMPTY_DICT
 from .dcs import ClusterConfig
 from .exceptions import ConfigParseError
 from .file_perm import pg_perm
@@ -445,14 +445,14 @@ class Config(object):
 
         for name, value in dynamic_configuration.items():
             if name == 'postgresql':
-                for name, value in (value or {}).items():
+                for name, value in (value or EMPTY_DICT).items():
                     if name == 'parameters':
                         config['postgresql'][name].update(self._process_postgresql_parameters(value))
                     elif name not in ('connect_address', 'proxy_address', 'listen',
                                       'config_dir', 'data_dir', 'pgpass', 'authentication'):
                         config['postgresql'][name] = deepcopy(value)
             elif name == 'standby_cluster':
-                for name, value in (value or {}).items():
+                for name, value in (value or EMPTY_DICT).items():
                     if name in self.__DEFAULT_CONFIG['standby_cluster']:
                         config['standby_cluster'][name] = deepcopy(value)
             elif name in config:  # only variables present in __DEFAULT_CONFIG allowed to be overridden from DCS

patroni/config_generator.py

@@ -15,6 +15,7 @@ if TYPE_CHECKING:  # pragma: no cover
     from psycopg2 import cursor
 
 from . import psycopg
+from .collections import EMPTY_DICT
 from .config import Config
 from .exceptions import PatroniException
 from .log import PatroniLogger
@@ -244,7 +245,8 @@ class SampleConfigGenerator(AbstractConfigGenerator):
                   See :func:`~patroni.postgresql.misc.postgres_major_version_to_int` and
                   :func:`~patroni.utils.get_major_version`.
         """
-        postgres_bin = ((self.config.get('postgresql') or {}).get('bin_name') or {}).get('postgres', 'postgres')
+        postgres_bin = ((self.config.get('postgresql')
+                         or EMPTY_DICT).get('bin_name') or EMPTY_DICT).get('postgres', 'postgres')
         return postgres_major_version_to_int(get_major_version(self.config['postgresql'].get('bin_dir'), postgres_bin))
 
     def generate(self) -> None:
@@ -411,8 +413,10 @@ class RunningClusterConfigGenerator(AbstractConfigGenerator):
             val = self.parsed_dsn.get(conn_param, os.getenv(env_var))
             if val:
                 su_params[conn_param] = val
-        patroni_env_su_username = ((self.config.get('authentication') or {}).get('superuser') or {}).get('username')
+        patroni_env_su_username = ((self.config.get('authentication')
-        patroni_env_su_pwd = ((self.config.get('authentication') or {}).get('superuser') or {}).get('password')
+                                    or EMPTY_DICT).get('superuser') or EMPTY_DICT).get('username')
+        patroni_env_su_pwd = ((self.config.get('authentication')
+                               or EMPTY_DICT).get('superuser') or EMPTY_DICT).get('password')
         # because we use "username" in the config for some reason
         su_params['username'] = su_params.pop('user', patroni_env_su_username) or getuser()
         su_params['password'] = su_params.get('password', patroni_env_su_pwd) or \

patroni/dcs/__init__.py

@@ -85,6 +85,8 @@ def dcs_modules() -> List[str]:
 
     :returns: list of known module names with absolute python module path namespace, e.g. ``patroni.dcs.etcd``.
     """
+    if TYPE_CHECKING:  # pragma: no cover
+        assert isinstance(__package__, str)
     return iter_modules(__package__)
 
 
@@ -101,6 +103,8 @@ def iter_dcs_classes(
 
     :returns: an iterator of tuples, each containing the module ``name`` and the imported DCS class object.
     """
+    if TYPE_CHECKING:  # pragma: no cover
+        assert isinstance(__package__, str)
     return iter_classes(__package__, AbstractDCS, config)
 
 

patroni/dcs/consul.py

@@ -444,8 +444,9 @@ class Consul(AbstractDCS):
 
         :returns: all MPP groups as :class:`dict`, with group IDs as keys and :class:`Cluster` objects as values.
         """
+        results: Optional[List[Dict[str, Any]]]
         _, results = self.retry(self._client.kv.get, path, recurse=True, consistency=self._consistency)
-        clusters: Dict[int, Dict[str, Cluster]] = defaultdict(dict)
+        clusters: Dict[int, Dict[str, Dict[str, Any]]] = defaultdict(dict)
         for node in results or []:
             key = node['Key'][len(path):].split('/', 1)
             if len(key) == 2 and self._mpp.group_re.match(key[0]):

patroni/dcs/kubernetes.py

@@ -20,6 +20,7 @@ from threading import Condition, Lock, Thread
 from typing import Any, Callable, Collection, Dict, List, Optional, Tuple, Type, Union, TYPE_CHECKING
 
 from . import AbstractDCS, Cluster, ClusterConfig, Failover, Leader, Member, Status, SyncState, TimelineHistory
+from ..collections import EMPTY_DICT
 from ..exceptions import DCSError
 from ..postgresql.mpp import AbstractMPP
 from ..utils import deep_compare, iter_response_objects, keepalive_socket_options, \
@@ -470,7 +471,7 @@ class K8sClient(object):
                 if len(args) == 3:  # name, namespace, body
                     body = args[2]
                 elif action == 'create':  # namespace, body
-                    body = args[1]
+                    body = args[1]  # pyright: ignore [reportGeneralTypeIssues]
                 elif action == 'delete':  # name, namespace
                     body = kwargs.pop('body', None)
                 else:
@@ -509,7 +510,7 @@ class KubernetesRetriableException(k8s_client.rest.ApiException):
     @property
     def sleeptime(self) -> Optional[int]:
         try:
-            return int((self.headers or {}).get('retry-after', ''))
+            return int((self.headers or EMPTY_DICT).get('retry-after', ''))
         except Exception:
             return None
 
@@ -654,7 +655,7 @@ class ObjectCache(Thread):
             obj = K8sObject(obj)
             success, old_value = self.set(name, obj)
             if success:
-                new_value = (obj.metadata.annotations or {}).get(self._annotations_map.get(name))
+                new_value = (obj.metadata.annotations or EMPTY_DICT).get(self._annotations_map.get(name, ''))
         elif ev_type == 'DELETED':
             success, old_value = self.delete(name, obj['metadata']['resourceVersion'])
         else:
@@ -662,7 +663,7 @@ class ObjectCache(Thread):
 
         if success and obj.get('kind') != 'Pod':
             if old_value:
-                old_value = (old_value.metadata.annotations or {}).get(self._annotations_map.get(name))
+                old_value = (old_value.metadata.annotations or EMPTY_DICT).get(self._annotations_map.get(name, ''))
 
             value_changed = old_value != new_value and \
                 (name != self._dcs.config_path or old_value is not None and new_value is not None)
@@ -844,7 +845,7 @@ class Kubernetes(AbstractDCS):
 
     @staticmethod
     def member(pod: K8sObject) -> Member:
-        annotations = pod.metadata.annotations or {}
+        annotations = pod.metadata.annotations or EMPTY_DICT
         member = Member.from_node(pod.metadata.resource_version, pod.metadata.name, None, annotations.get('status', ''))
         member.data['pod_labels'] = pod.metadata.labels
         return member
@@ -925,7 +926,7 @@ class Kubernetes(AbstractDCS):
         failover = nodes.get(path + self._FAILOVER)
         metadata = failover and failover.metadata
         failover = metadata and Failover.from_node(metadata.resource_version,
-                                                   (metadata.annotations or {}).copy())
+                                                   (metadata.annotations or EMPTY_DICT).copy())
 
         # get synchronization state
         sync = nodes.get(path + self._SYNC)
@@ -1047,8 +1048,9 @@ class Kubernetes(AbstractDCS):
 
     def __target_ref(self, leader_ip: str, latest_subsets: List[K8sObject], pod: K8sObject) -> K8sObject:
         # we want to re-use existing target_ref if possible
+        empty_addresses: List[K8sObject] = []
         for subset in latest_subsets:
-            for address in subset.addresses or []:
+            for address in subset.addresses or empty_addresses:
                 if address.ip == leader_ip and address.target_ref and address.target_ref.name == self._name:
                     return address.target_ref
         return k8s_client.V1ObjectReference(kind='Pod', uid=pod.metadata.uid, namespace=self._namespace,
@@ -1056,7 +1058,8 @@ class Kubernetes(AbstractDCS):
 
     def _map_subsets(self, endpoints: Dict[str, Any], ips: List[str]) -> None:
         leader = self._kinds.get(self.leader_path)
-        latest_subsets = leader and leader.subsets or []
+        empty_addresses: List[K8sObject] = []
+        latest_subsets = leader and leader.subsets or empty_addresses
         if not ips:
             # We want to have subsets empty
             if latest_subsets:
@@ -1212,7 +1215,7 @@ class Kubernetes(AbstractDCS):
         if not retry.ensure_deadline(0.5):
             return False
 
-        kind_annotations = kind and kind.metadata.annotations or {}
+        kind_annotations = kind and kind.metadata.annotations or EMPTY_DICT
         kind_resource_version = kind and kind.metadata.resource_version
 
         # There is different leader or resource_version in cache didn't change
@@ -1225,7 +1228,7 @@ class Kubernetes(AbstractDCS):
     def update_leader(self, leader: Leader, last_lsn: Optional[int],
                       slots: Optional[Dict[str, int]] = None, failsafe: Optional[Dict[str, str]] = None) -> bool:
         kind = self._kinds.get(self.leader_path)
-        kind_annotations = kind and kind.metadata.annotations or {}
+        kind_annotations = kind and kind.metadata.annotations or EMPTY_DICT
 
         if kind and kind_annotations.get(self._LEADER) != self._name:
             return False
@@ -1346,7 +1349,7 @@ class Kubernetes(AbstractDCS):
     def delete_leader(self, leader: Optional[Leader], last_lsn: Optional[int] = None) -> bool:
         ret = False
         kind = self._kinds.get(self.leader_path)
-        if kind and (kind.metadata.annotations or {}).get(self._LEADER) == self._name:
+        if kind and (kind.metadata.annotations or EMPTY_DICT).get(self._LEADER) == self._name:
             annotations: Dict[str, Optional[str]] = {self._LEADER: None}
             if last_lsn:
                 annotations[self._OPTIME] = str(last_lsn)

patroni/global_config.py

@@ -10,6 +10,7 @@ import types
 from copy import deepcopy
 from typing import Any, Dict, List, Optional, Union, TYPE_CHECKING
 
+from .collections import EMPTY_DICT
 from .utils import parse_bool, parse_int
 
 if TYPE_CHECKING:  # pragma: no cover
@@ -214,7 +215,7 @@ class GlobalConfig(types.ModuleType):
     @property
     def use_slots(self) -> bool:
         """``True`` if cluster is configured to use replication slots."""
-        return bool(parse_bool((self.get('postgresql') or {}).get('use_slots', True)))
+        return bool(parse_bool((self.get('postgresql') or EMPTY_DICT).get('use_slots', True)))
 
     @property
     def permanent_slots(self) -> Dict[str, Any]:
@@ -222,7 +223,7 @@ class GlobalConfig(types.ModuleType):
         return deepcopy(self.get('permanent_replication_slots')
                         or self.get('permanent_slots')
                         or self.get('slots')
-                        or {})
+                        or EMPTY_DICT.copy())
 
 
 sys.modules[__name__] = GlobalConfig()

patroni/log.py

@@ -413,7 +413,8 @@ class PatroniLogger(Thread):
                 if not isinstance(handler, RotatingFileHandler):
                     handler = RotatingFileHandler(os.path.join(config['dir'], __name__))
 
-                handler.maxBytes = int(config.get('file_size', 25000000))  # pyright: ignore [reportGeneralTypeIssues]
+                max_file_size = int(config.get('file_size', 25000000))
+                handler.maxBytes = max_file_size  # pyright: ignore [reportAttributeAccessIssue]
                 handler.backupCount = int(config.get('file_num', 4))
             # we can't use `if not isinstance(handler, logging.StreamHandler)` below,
             # because RotatingFileHandler is a child of StreamHandler!!!

patroni/postgresql/__init__.py

@@ -26,7 +26,7 @@ from .slots import SlotsHandler
 from .sync import SyncHandler
 from .. import global_config, psycopg
 from ..async_executor import CriticalTask
-from ..collections import CaseInsensitiveSet, CaseInsensitiveDict
+from ..collections import CaseInsensitiveSet, CaseInsensitiveDict, EMPTY_DICT
 from ..dcs import Cluster, Leader, Member, SLOT_ADVANCE_AVAILABLE_VERSION
 from ..exceptions import PostgresConnectionException
 from ..utils import Retry, RetryFailedError, polling_loop, data_directory_is_empty, parse_int
@@ -272,7 +272,7 @@ class Postgresql(object):
 
         :returns: path to Postgres binary named *cmd*.
         """
-        return os.path.join(self._bin_dir, (self.config.get('bin_name', {}) or {}).get(cmd, cmd))
+        return os.path.join(self._bin_dir, (self.config.get('bin_name', {}) or EMPTY_DICT).get(cmd, cmd))
 
     def pg_ctl(self, cmd: str, *args: str, **kwargs: Any) -> bool:
         """Builds and executes pg_ctl command
@@ -414,7 +414,7 @@ class Postgresql(object):
         return data_directory_is_empty(self._data_dir)
 
     def replica_method_options(self, method: str) -> Dict[str, Any]:
-        return deepcopy(self.config.get(method, {}) or {})
+        return deepcopy(self.config.get(method, {}) or EMPTY_DICT.copy())
 
     def replica_method_can_work_without_replication_connection(self, method: str) -> bool:
         return method != 'basebackup' and bool(self.replica_method_options(method).get('no_master')

patroni/postgresql/available_parameters/__init__.py

@@ -1,12 +1,13 @@
 import logging
 import sys
 
-from pathlib import Path
 from typing import Iterator
 
 logger = logging.getLogger(__name__)
 
-if sys.version_info < (3, 9):
+if sys.version_info < (3, 9):  # pragma: no cover
+    from pathlib import Path
+
     PathLikeObj = Path
     conf_dir = Path(__file__).parent
 else:

patroni/postgresql/bootstrap.py

@@ -7,6 +7,7 @@ import time
 from typing import Any, Callable, Dict, List, Optional, Union, Tuple, TYPE_CHECKING
 
 from ..async_executor import CriticalTask
+from ..collections import EMPTY_DICT
 from ..dcs import Leader, Member, RemoteMember
 from ..psycopg import quote_ident, quote_literal
 from ..utils import deep_compare, unquote
@@ -146,7 +147,7 @@ class Bootstrap(object):
 
         # make sure there is no trigger file or postgres will be automatically promoted
         trigger_file = self._postgresql.config.triggerfile_good_name
-        trigger_file = (self._postgresql.config.get('recovery_conf') or {}).get(trigger_file) or 'promote'
+        trigger_file = (self._postgresql.config.get('recovery_conf') or EMPTY_DICT).get(trigger_file) or 'promote'
         trigger_file = os.path.abspath(os.path.join(self._postgresql.data_dir, trigger_file))
         if os.path.exists(trigger_file):
             os.unlink(trigger_file)
@@ -441,7 +442,7 @@ END;$$""".format(f, quote_ident(rewind['username'], postgresql.connection()))
                 if config.get('users'):
                     logger.warning('User creation via "bootstrap.users" will be removed in v4.0.0')
 
-                for name, value in (config.get('users') or {}).items():
+                for name, value in (config.get('users') or EMPTY_DICT).items():
                     if all(name != a.get('username') for a in (superuser, replication, rewind)):
                         self.create_or_update_role(name, value.get('password'), value.get('options', []))
 

patroni/postgresql/cancellable.py

@@ -100,7 +100,8 @@ class CancellableSubprocess(CancellableExecutor):
 
             if started and self._process is not None:
                 if isinstance(communicate, dict):
-                    communicate['stdout'], communicate['stderr'] = self._process.communicate(input_data)
+                    communicate['stdout'], communicate['stderr'] = \
+                        self._process.communicate(input_data)  # pyright: ignore [reportGeneralTypeIssues]
                 return self._process.wait()
         finally:
             with self._lock:

patroni/postgresql/config.py

@@ -13,7 +13,7 @@ from typing import Any, Callable, Collection, Dict, Iterator, List, Optional, Un
 
 from .validator import recovery_parameters, transform_postgresql_parameter_value, transform_recovery_parameter_value
 from .. import global_config
-from ..collections import CaseInsensitiveDict, CaseInsensitiveSet
+from ..collections import CaseInsensitiveDict, CaseInsensitiveSet, EMPTY_DICT
 from ..dcs import Leader, Member, RemoteMember, slot_name_from_member_name
 from ..exceptions import PatroniFatalException, PostgresConnectionException
 from ..file_perm import pg_perm
@@ -619,7 +619,8 @@ class ConfigHandler(object):
             fd.write_param(name, value)
 
     def build_recovery_params(self, member: Union[Leader, Member, None]) -> CaseInsensitiveDict:
-        recovery_params = CaseInsensitiveDict({p: v for p, v in (self.get('recovery_conf') or {}).items()
+        default: Dict[str, Any] = {}
+        recovery_params = CaseInsensitiveDict({p: v for p, v in (self.get('recovery_conf') or default).items()
                                                if not p.lower().startswith('recovery_target')
                                                and p.lower() not in ('primary_conninfo', 'primary_slot_name')})
         recovery_params.update({'standby_mode': 'on', 'recovery_target_timeline': 'latest'})
@@ -845,7 +846,7 @@ class ConfigHandler(object):
             required['restart' if mtype else 'reload'] += 1
 
         wanted_recovery_params = self.build_recovery_params(member)
-        for param, value in (self._current_recovery_params or {}).items():
+        for param, value in (self._current_recovery_params or EMPTY_DICT).items():
             # Skip certain parameters defined in the included postgres config files
             # if we know that they are not specified in the patroni configuration.
             if len(value) > 2 and value[2] not in (self._postgresql_conf, self._auto_conf) and \
@@ -1324,4 +1325,4 @@ class ConfigHandler(object):
         return self._config.get(key, default)
 
     def restore_command(self) -> Optional[str]:
-        return (self.get('recovery_conf') or {}).get('restore_command')
+        return (self.get('recovery_conf') or EMPTY_DICT).get('restore_command')

patroni/postgresql/mpp/__init__.py

@@ -299,6 +299,8 @@ def iter_mpp_classes(
 
     :yields: tuples, each containing the module ``name`` and the imported MPP class object.
     """
+    if TYPE_CHECKING:  # pragma: no cover
+        assert isinstance(__package__, str)
     yield from iter_classes(__package__, AbstractMPP, config)
 
 

patroni/postgresql/mpp/citus.py

@@ -4,7 +4,7 @@ import time
 
 from threading import Condition, Event, Thread
 from urllib.parse import urlparse
-from typing import Any, Dict, List, Optional, Union, Tuple, TYPE_CHECKING
+from typing import Any, Collection, Dict, Iterator, List, Optional, Union, Set, Tuple, TYPE_CHECKING
 
 from . import AbstractMPP, AbstractMPPHandler
 from ...dcs import Cluster
@@ -19,19 +19,312 @@ CITUS_SLOT_NAME_RE = re.compile(r'^citus_shard_(move|split)_slot(_[1-9][0-9]*){2
 logger = logging.getLogger(__name__)
 
 
-class PgDistNode(object):
+class PgDistNode:
-    """Represents a single row in the `pg_dist_node` table"""
+    """Represents a single row in "pg_dist_node" table.
 
-    def __init__(self, group: int, host: str, port: int, event: str, nodeid: Optional[int] = None,
+    .. note::
-                 timeout: Optional[float] = None, cooldown: Optional[float] = None) -> None:
+
-        self.group = group
+        Unlike "noderole" possible values of ``role`` are ``primary``, ``secondary``, and ``demoted``.
-        # A weird way of pausing client connections by adding the `-demoted` suffix to the hostname
+        The last one is used to pause client connections on the coordinator to the worker by
-        self.host = host + ('-demoted' if event == 'before_demote' else '')
+        appending ``-demoted`` suffix to the "nodename". The actual "noderole" in DB remains ``primary``.
+
+    :ivar host: "nodename" value
+    :ivar port: "nodeport" value
+    :ivar role: "noderole" value
+    :ivar nodeid: "nodeid" value
+    """
+
+    def __init__(self, host: str, port: int, role: str, nodeid: Optional[int] = None) -> None:
+        """Create a :class:`PgDistNode` object based on given arguments.
+
+        :param host: "nodename" of the Citus coordinator or worker.
+        :param port: "nodeport" of the Citus coordinator or worker.
+        :param role: "noderole" value.
+        :param nodeid: id of the row in the "pg_dist_node".
+        """
+        self.host = host
         self.port = port
+        self.role = role
+        self.nodeid = nodeid
+
+    def __hash__(self) -> int:
+        """Defines a hash function to put :class:`PgDistNode` objects to :class:`PgDistGroup` set-like object.
+
+        .. note::
+            We use (:attr:`host`, :attr:`port`) tuple here because it is one of the UNIQUE constraints on the
+            "pg_dist_node" table. The :attr:`role` value is irrelevant here because nodes may change their roles.
+        """
+        return hash((self.host, self.port))
+
+    def __eq__(self, other: Any) -> bool:
+        """Defines a comparison function.
+
+        :returns: ``True`` if :attr:`host` and :attr:`port` between two instances are the same.
+        """
+        return isinstance(other, PgDistNode) and self.host == other.host and self.port == other.port
+
+    def __str__(self) -> str:
+        return ('PgDistNode(nodeid={0},host={1},port={2},role={3})'
+                .format(self.nodeid, self.host, self.port, self.role))
+
+    def __repr__(self) -> str:
+        return str(self)
+
+    def is_primary(self) -> bool:
+        """Checks whether this object represents "primary" in a corresponding group.
+
+        :returns: ``True`` if this object represents the ``primary``.
+        """
+        return self.role in ('primary', 'demoted')
+
+    def as_tuple(self, include_nodeid: bool = False) -> Tuple[str, int, str, Optional[int]]:
+        """Helper method to compare two :class:`PgDistGroup` objects.
+
+        .. note::
+
+            *include_nodeid* is set to ``True`` only in unit-tests.
+
+        :param include_nodeid: whether :attr:`nodeid` should be taken into account when comparison is performed.
+
+        :returns: :class:`tuple` object with :attr:`host`, :attr:`port`, :attr:`role`, and optionally :attr:`nodeid`.
+        """
+        return self.host, self.port, self.role, (self.nodeid if include_nodeid else None)
+
+
+class PgDistGroup(Set[PgDistNode]):
+    """A :class:`set`-like object that represents a Citus group in "pg_dist_node" table.
+
+    This class implements a set of methods to compare topology and if it is necessary
+    to transition from the old to the new topology in a "safe" manner:
+
+        * register new primary/secondaries
+        * replace gone secondaries with added secondaries
+        * failover and switchover
+
+    Typically there will be at least one :class:`PgDistNode` object registered (``primary``).
+    In addition to that there could be one or more ``secondary`` nodes.
+
+    :ivar failover: whether the ``primary`` row should be updated as a result of :func:`transition` method call.
+    :ivar groupid: the "groupid" from "pg_dist_node".
+    """
+
+    def __init__(self, groupid: int, nodes: Optional[Collection[PgDistNode]] = None) -> None:
+        """Creates a :class:`PgDistGroup` object based on given arguments.
+
+        :param groupid: the groupid from "pg_dist_node".
+        :param nodes: a collection of :class:`PgDistNode` objects that belog to a *groupid*.
+        """
+        self.failover = False
+        self.groupid = groupid
+
+        if nodes:
+            self.update(nodes)
+
+    def equals(self, other: 'PgDistGroup', check_nodeid: bool = False) -> bool:
+        """Compares two :class:`PgDistGroup` objects.
+
+        :param other: what we want to compare with.
+        :param check_nodeid: whether :attr:`PgDistNode.nodeid` should be compared in addition to
+                             :attr:`PgDistNode.host`, :attr:`PgDistNode.port`, and :attr:`PgDistNode.role`.
+
+        :returns: ``True`` if two :class:`PgDistGroup` objects are fully identical.
+        """
+        return self.groupid == other.groupid\
+            and set(v.as_tuple(check_nodeid) for v in self) == set(v.as_tuple(check_nodeid) for v in other)
+
+    def primary(self) -> Optional[PgDistNode]:
+        """Finds and returns :class:`PgDistNode` object that represents the "primary".
+
+        :returns: :class:`PgDistNode` object which represents the "primary" or ``None`` if not found.
+        """
+        return next(iter(v for v in self if v.is_primary()), None)
+
+    def get(self, value: PgDistNode) -> Optional[PgDistNode]:
+        """Performs a lookup of the actual value in a set.
+
+        .. note::
+            It is necessary because :func:`__hash__` and :func:`__eq__` methods in :class:`PgDistNode` are
+            redefined and effectively they check only :attr:`PgDistNode.host` and :attr:`PgDistNode.port` attributes.
+
+        :param value: the key we search for.
+        :returns: the actual :class:`PgDistNode` value from this :class:`PgDistGroup` object or ``None`` if not found.
+        """
+        return next(iter(v for v in self if v == value), None)
+
+    def transition(self, old: 'PgDistGroup') -> Iterator[PgDistNode]:
+        """Compares this topology with the old one and yields transitions that transform the old to the new one.
+
+        .. note::
+            The actual yielded object is :class:`PgDistNode` that will be passed to
+            the :meth:`CitusHandler.update_node` to execute all transitions in a transaction.
+
+            In addition to the yielding transactions this method fills up :attr:`PgDistNode.nodeid`
+            attribute for nodes that are presented in the old and in the new topology.
+
+            There are a few simple rules/constraints that are imposed by Citus and must be followed:
+            - adding/removing nodes is only possible when metadata is synced to all registered "priorities".
+
+            - the "primary" row in "pg_dist_node" always keeps the nodeid (unless it is
+              removed, but it is not supported by Patroni).
+
+            - "nodename", "nodeport" must be unique across all rows in the "pg_dist_node". This means that
+              every time we want to change the nodeid of an existing node (i.e. to change it from secondary
+              to primary), we should first write some other "nodename"/"nodeport" to the row it's currently in.
+
+            - updating "broken" nodes always works and metadata is synced asynchnonously after the commit.
+
+        Following these rules below is an example of the switchover between node1 (primary, nodeid=4)
+        and node2 (secondary, nodeid=5).
+
+        .. code-block:: SQL
+
+            BEGIN;
+                SELECT citus_update_node(4, 'node1-demoted', 5432);
+                SELECT citus_update_node(5, 'node1', 5432);
+                SELECT citus_update_node(4, 'node2', 5432);
+            COMMIT;
+
+        :param old: the last known topology registered in "pg_dist_node" for a given :attr:`groupid`.
+
+        :yields: :class:`PgDistNode` objects that must be updated/added/removed in "pg_dist_node".
+        """
+        self.failover = old.failover
+
+        new_primary = self.primary()
+        assert new_primary is not None
+        old_primary = old.primary()
+
+        gone_nodes = old - self - {old_primary}
+        added_nodes = self - old - {new_primary}
+
+        if not old_primary:
+            # We did not have any nodes in the group yet and we're adding one now
+            yield new_primary
+        elif old_primary == new_primary:
+            new_primary.nodeid = old_primary.nodeid
+            # Controlled switchover with pausing client connections.
+            # Achieved by updating the primary row and putting hostname = '${host}-demoted' in a transaction.
+            if old_primary.role != new_primary.role:
+                self.failover = True
+                yield new_primary
+        elif old_primary != new_primary:
+            self.failover = True
+
+            new_primary_old_node = old.get(new_primary)
+            old_primary_new_node = self.get(old_primary)
+
+            # The new primary was registered as a secondary before failover
+            if new_primary_old_node:
+                new_node = None
+                # Old primary is gone and some new secondaries were added.
+                # We can use the row of promoted secondary to add the new secondary.
+                if not old_primary_new_node and added_nodes:
+                    new_node = added_nodes.pop()
+                    new_node.nodeid = new_primary_old_node.nodeid
+                    yield new_node
+
+                    # notify _maybe_register_old_primary_as_secondary that the old primary should not be re-registered
+                    old_primary.role = 'secondary'
+                # In opposite case we need to change the primary record to '${host}-demoted:${port}'
+                # before we can put its host:port to the row of promoted secondary.
+                elif old_primary.role == 'primary':
+                    old_primary.role = 'demoted'
+                    yield old_primary
+
+                # The old primary is gone and the promoted secondary row wasn't yet used.
+                if not old_primary_new_node and not new_node:
+                    # We have to "add" the gone primary to the row of promoted secondary because
+                    # nodes could not be removed while the metadata isn't synced.
+                    old_primary_new_node = PgDistNode(old_primary.host, old_primary.port, new_primary_old_node.role)
+                    self.add(old_primary_new_node)
+
+                # put the old primary instead of promoted secondary
+                if old_primary_new_node:
+                    old_primary_new_node.nodeid = new_primary_old_node.nodeid
+                    yield old_primary_new_node
+
+            # update the primary record with the new information
+            new_primary.nodeid = old_primary.nodeid
+            yield new_primary
+
+            # The new primary was never registered as a standby and there are secondaries that have gone away. Since
+            # nodes can't be removed while metadata isn't synced we have to temporarily "add" the old primary back.
+            if not new_primary_old_node and gone_nodes:
+                # We were in the middle of controlled switchover while the primary disappeared.
+                # If there are any gone nodes that can't be reused for new secondaries we will
+                # use one of them to temporarily "add" the old primary back as a secondary.
+                if not old_primary_new_node and old_primary.role == 'demoted' and len(gone_nodes) > len(added_nodes):
+                    old_primary_new_node = PgDistNode(old_primary.host, old_primary.port, 'secondary')
+                    self.add(old_primary_new_node)
+
+                # Use one of the gone secondaries to put host:port of the old primary there.
+                if old_primary_new_node:
+                    old_primary_new_node.nodeid = gone_nodes.pop().nodeid
+                    yield old_primary_new_node
+
+        # Fill nodeid for standbys in the new topology from the old ones
+        old_replicas = {v: v for v in old if not v.is_primary()}
+        for n in self:
+            if not n.is_primary() and not n.nodeid and n in old_replicas:
+                n.nodeid = old_replicas[n].nodeid
+
+        # Reuse nodeid's of gone standbys to "add" new standbys
+        while gone_nodes and added_nodes:
+            a = added_nodes.pop()
+            a.nodeid = gone_nodes.pop().nodeid
+            yield a
+
+        # Adding or removing nodes operations are executed on primaries in all Citus groups in 2PC.
+        # If we know that the primary was updated (self.failover is True) that automatically means that
+        # adding/removing nodes calls will fail and the whole transaction will be aborted. Therefore
+        # we discard operations that add/remove secondaries if we know that the primary was just updated.
+        # The inconsistency will be automatically resolved on the next Patroni heartbeat loop.
+
+        # Remove remaining nodes that are gone, but only in case if metadata is in sync (self.failover is False).
+        for g in gone_nodes:
+            if not self.failover:
+                # Remove the node if we expect metadata to be in sync
+                yield PgDistNode(g.host, g.port, '')
+            else:
+                # Otherwise add these nodes to the new topology
+                self.add(g)
+
+        # Add new nodes to the metadata, but only in case if metadata is in sync (self.failover is False).
+        for a in added_nodes:
+            if not self.failover:
+                # Add the node if we expect metadata to be in sync
+                yield a
+            else:
+                # Otherwise remove them from the new topology
+                self.discard(a)
+
+
+class PgDistTask(PgDistGroup):
+    """A "task" that represents the current or desired state of "pg_dist_node" for a provided *groupid*.
+
+    :ivar group: the "groupid" in "pg_dist_node".
+    :ivar event: an "event" that resulted in creating this task.
+                 possible values: "before_demote", "before_promote", "after_promote".
+    :ivar timeout: a transaction timeout if the task resulted in starting a transaction.
+    :ivar cooldown: the cooldown value for ``citus_update_node()`` UDF call.
+    :ivar deadline: the time in unix seconds when the transaction is allowed to be rolled back.
+    """
+
+    def __init__(self, groupid: int, nodes: Optional[Collection[PgDistNode]], event: str,
+                 timeout: Optional[float] = None, cooldown: Optional[float] = None) -> None:
+        """Create a :class:`PgDistTask` object based on given arguments.
+
+        :param groupid: the groupid from "pg_dist_node".
+        :param nodes: a collection of :class:`PgDistNode` objects that belog to a *groupid*.
+        :param event: an "event" that resulted in creating this task.
+        :param timeout: a transaction timeout if the task resulted in starting a transaction.
+        :param cooldown: the cooldown value for ``citus_update_node()`` UDF call.
+        """
+        super(PgDistTask, self).__init__(groupid, nodes)
+
         # Event that is trying to change or changed the given row.
         # Possible values: before_demote, before_promote, after_promote.
         self.event = event
-        self.nodeid = nodeid
 
         # If transaction was started, we need to COMMIT/ROLLBACK before the deadline
         self.timeout = timeout
@@ -46,25 +339,20 @@ class PgDistNode(object):
         self._event = Event()
 
     def wait(self) -> None:
+        """Wait until this task is processed by a dedicated thread."""
         self._event.wait()
 
     def wakeup(self) -> None:
+        """Notify a thread that created a task that it was processed."""
         self._event.set()
 
     def __eq__(self, other: Any) -> bool:
-        return isinstance(other, PgDistNode) and self.event == other.event\
+        return isinstance(other, PgDistTask) and self.event == other.event\
-            and self.host == other.host and self.port == other.port
+            and super(PgDistTask, self).equals(other)
 
     def __ne__(self, other: Any) -> bool:
         return not self == other
 
-    def __str__(self) -> str:
-        return ('PgDistNode(nodeid={0},group={1},host={2},port={3},event={4})'
-                .format(self.nodeid, self.group, self.host, self.port, self.event))
-
-    def __repr__(self) -> str:
-        return str(self)
-
 
 class Citus(AbstractMPP):
 
@@ -109,11 +397,11 @@ class CitusHandler(Citus, AbstractMPPHandler, Thread):
             self._connection = postgresql.connection_pool.get(
                 'citus', {'dbname': config['database'],
                           'options': '-c statement_timeout=0 -c idle_in_transaction_session_timeout=0'})
-        self._pg_dist_node: Dict[int, PgDistNode] = {}  # Cache of pg_dist_node: {groupid: PgDistNode()}
+        self._pg_dist_group: Dict[int, PgDistTask] = {}  # Cache of pg_dist_node: {groupid: PgDistTask()}
-        self._tasks: List[PgDistNode] = []  # Requests to change pg_dist_node, every task is a `PgDistNode`
+        self._tasks: List[PgDistTask] = []  # Requests to change pg_dist_group, every task is a `PgDistTask`
-        self._in_flight: Optional[PgDistNode] = None  # Reference to the `PgDistNode` being changed in a transaction
+        self._in_flight: Optional[PgDistTask] = None  # Reference to the `PgDistTask` being changed in a transaction
-        self._schedule_load_pg_dist_node = True  # Flag that "pg_dist_node" should be queried from the database
+        self._schedule_load_pg_dist_group = True  # Flag that "pg_dist_group" should be queried from the database
-        self._condition = Condition()  # protects _pg_dist_node, _tasks, _in_flight, and _schedule_load_pg_dist_node
+        self._condition = Condition()  # protects _pg_dist_group, _tasks, _in_flight, and _schedule_load_pg_dist_group
         self.schedule_cache_rebuild()
 
     def schedule_cache_rebuild(self) -> None:
@@ -122,12 +410,12 @@ class CitusHandler(Citus, AbstractMPPHandler, Thread):
         Is called to notify handler that it has to refresh its metadata cache from the database.
         """
         with self._condition:
-            self._schedule_load_pg_dist_node = True
+            self._schedule_load_pg_dist_group = True
 
     def on_demote(self) -> None:
         with self._condition:
-            self._pg_dist_node.clear()
+            self._pg_dist_group.clear()
-            empty_tasks: List[PgDistNode] = []
+            empty_tasks: List[PgDistTask] = []
             self._tasks[:] = empty_tasks
             self._in_flight = None
 
@@ -143,22 +431,28 @@ class CitusHandler(Citus, AbstractMPPHandler, Thread):
             self.schedule_cache_rebuild()
             raise e
 
-    def load_pg_dist_node(self) -> bool:
+    def load_pg_dist_group(self) -> bool:
         """Read from the `pg_dist_node` table and put it into the local cache"""
 
         with self._condition:
-            if not self._schedule_load_pg_dist_node:
+            if not self._schedule_load_pg_dist_group:
                 return True
-            self._schedule_load_pg_dist_node = False
+            self._schedule_load_pg_dist_group = False
 
         try:
-            rows = self.query("SELECT nodeid, groupid, nodename, nodeport, noderole"
+            rows = self.query('SELECT groupid, nodename, nodeport, noderole, nodeid FROM pg_catalog.pg_dist_node')
-                              " FROM pg_catalog.pg_dist_node WHERE noderole = 'primary'")
         except Exception:
             return False
 
+        pg_dist_group: Dict[int, PgDistTask] = {}
+
+        for row in rows:
+            if row[0] not in pg_dist_group:
+                pg_dist_group[row[0]] = PgDistTask(row[0], nodes=set(), event='after_promote')
+            pg_dist_group[row[0]].add(PgDistNode(*row[1:]))
+
         with self._condition:
-            self._pg_dist_node = {r[1]: PgDistNode(r[1], r[2], r[3], 'after_promote', r[0]) for r in rows}
+            self._pg_dist_group = pg_dist_group
         return True
 
     def sync_meta_data(self, cluster: Cluster) -> None:
@@ -166,7 +460,7 @@ class CitusHandler(Citus, AbstractMPPHandler, Thread):
 
         We can't always rely on REST API calls from worker nodes in order
         to maintain `pg_dist_node`, therefore at least once per heartbeat
-        loop we make sure that workes registered in `self._pg_dist_node`
+        loop we make sure that workes registered in `self._pg_dist_group`
         cache are matching the cluster view from DCS by creating tasks
         the same way as it is done from the REST API."""
 
@@ -177,20 +471,21 @@ class CitusHandler(Citus, AbstractMPPHandler, Thread):
             if not self.is_alive():
                 self.start()
 
-        self.add_task('after_promote', CITUS_COORDINATOR_GROUP_ID, self._postgresql.connection_string)
+        self.add_task('after_promote', CITUS_COORDINATOR_GROUP_ID, cluster,
+                      self._postgresql.name, self._postgresql.connection_string)
 
-        for group, worker in cluster.workers.items():
+        for groupid, worker in cluster.workers.items():
             leader = worker.leader
             if leader and leader.conn_url\
                     and leader.data.get('role') in ('master', 'primary') and leader.data.get('state') == 'running':
-                self.add_task('after_promote', group, leader.conn_url)
+                self.add_task('after_promote', groupid, worker, leader.name, leader.conn_url)
 
-    def find_task_by_group(self, group: int) -> Optional[int]:
+    def find_task_by_groupid(self, groupid: int) -> Optional[int]:
         for i, task in enumerate(self._tasks):
-            if task.group == group:
+            if task.groupid == groupid:
                 return i
 
-    def pick_task(self) -> Tuple[Optional[int], Optional[PgDistNode]]:
+    def pick_task(self) -> Tuple[Optional[int], Optional[PgDistTask]]:
         """Returns the tuple(i, task), where `i` - is the task index in the self._tasks list
 
         Tasks are picked by following priorities:
@@ -198,44 +493,56 @@ class CitusHandler(Citus, AbstractMPPHandler, Thread):
         1. If there is already a transaction in progress, pick a task
            that that will change already affected worker primary.
         2. If the coordinator address should be changed - pick a task
-           with group=0 (coordinators are always in group 0).
+           with groupid=0 (coordinators are always in groupid 0).
         3. Pick a task that is the oldest (first from the self._tasks)
         """
 
         with self._condition:
             if self._in_flight:
-                i = self.find_task_by_group(self._in_flight.group)
+                i = self.find_task_by_groupid(self._in_flight.groupid)
             else:
                 while True:
-                    i = self.find_task_by_group(CITUS_COORDINATOR_GROUP_ID)  # set_coordinator
+                    i = self.find_task_by_groupid(CITUS_COORDINATOR_GROUP_ID)  # set_coordinator
                     if i is None and self._tasks:
                         i = 0
                     if i is None:
                         break
                     task = self._tasks[i]
-                    if task == self._pg_dist_node.get(task.group):
+                    if task == self._pg_dist_group.get(task.groupid):
-                        self._tasks.pop(i)  # nothing to do because cached version of pg_dist_node already matches
+                        self._tasks.pop(i)  # nothing to do because cached version of pg_dist_group already matches
                     else:
                         break
             task = self._tasks[i] if i is not None else None
 
-            # When tasks are added it could happen that self._pg_dist_node
-            # wasn't ready (self._schedule_load_pg_dist_node is False)
-            # and hence the nodeid wasn't filled.
-            if task and task.group in self._pg_dist_node:
-                task.nodeid = self._pg_dist_node[task.group].nodeid
             return i, task
 
-    def update_node(self, task: PgDistNode) -> None:
+    def update_node(self, groupid: int, node: PgDistNode, cooldown: float = 10000) -> None:
-        if task.nodeid is not None:
+        if node.role not in ('primary', 'secondary', 'demoted'):
+            self.query('SELECT pg_catalog.citus_remove_node(%s, %s)', node.host, node.port)
+        elif node.nodeid is not None:
+            host = node.host + ('-demoted' if node.role == 'demoted' else '')
             self.query('SELECT pg_catalog.citus_update_node(%s, %s, %s, true, %s)',
-                       task.nodeid, task.host, task.port, task.cooldown)
+                       node.nodeid, host, node.port, cooldown)
-        elif task.event != 'before_demote':
+        elif node.role != 'demoted':
-            task.nodeid = self.query("SELECT pg_catalog.citus_add_node(%s, %s, %s, 'primary', 'default')",
+            node.nodeid = self.query("SELECT pg_catalog.citus_add_node(%s, %s, %s, %s, 'default')",
-                                     task.host, task.port, task.group)[0][0]
+                                     node.host, node.port, groupid, node.role)[0][0]
 
-    def process_task(self, task: PgDistNode) -> bool:
+    def update_group(self, task: PgDistTask, transaction: bool) -> None:
-        """Updates a single row in `pg_dist_node` table, optionally in a transaction.
+        current_state = self._in_flight\
+            or self._pg_dist_group.get(task.groupid)\
+            or PgDistTask(task.groupid, set(), 'after_promote')
+        transitions = list(task.transition(current_state))
+        if transitions:
+            if not transaction and len(transitions) > 1:
+                self.query('BEGIN')
+            for node in transitions:
+                self.update_node(task.groupid, node, task.cooldown)
+            if not transaction and len(transitions) > 1:
+                task.failover = False
+                self.query('COMMIT')
+
+    def process_task(self, task: PgDistTask) -> bool:
+        """Updates a single row in `pg_dist_group` table, optionally in a transaction.
 
         The transaction is started if we do a demote of the worker node or before promoting the other worker if
         there is no transaction in progress. And, the transaction is committed when the switchover/failover completed.
@@ -246,34 +553,30 @@ class CitusHandler(Citus, AbstractMPPHandler, Thread):
         .. note:
             Read access to `self._in_flight` isn't protected because we know it can't be changed outside of our thread.
 
-        :param task: reference to a :class:`PgDistNode` object that represents a row to be updated/created.
+        :param task: reference to a :class:`PgDistTask` object that represents a row to be updated/created.
-        :returns: `True` if the row was succesfully created/updated or transaction in progress
+        :returns: ``True`` if the row was succesfully created/updated or transaction in progress
-            was committed as an indicator that the `self._pg_dist_node` cache should be updated,
+            was committed as an indicator that the `self._pg_dist_group` cache should be updated,
             or, if the new transaction was opened, this method returns `False`.
         """
 
         if task.event == 'after_promote':
-            # The after_promote may happen without previous before_demote and/or
+            self.update_group(task, self._in_flight is not None)
-            # before_promore.  In this case we just call self.update_node() method.
-            # If there is a transaction in progress, it could be that it already did
-            # required changes and we can simply COMMIT.
-            if not self._in_flight or self._in_flight.host != task.host or self._in_flight.port != task.port:
-                self.update_node(task)
             if self._in_flight:
                 self.query('COMMIT')
+            task.failover = False
             return True
         else:  # before_demote, before_promote
             if task.timeout:
                 task.deadline = time.time() + task.timeout
             if not self._in_flight:
                 self.query('BEGIN')
-            self.update_node(task)
+            self.update_group(task, True)
         return False
 
     def process_tasks(self) -> None:
         while True:
             # Read access to `_in_flight` isn't protected because we know it can't be changed outside of our thread.
-            if not self._in_flight and not self.load_pg_dist_node():
+            if not self._in_flight and not self.load_pg_dist_group():
                 break
 
             i, task = self.pick_task()
@@ -287,7 +590,7 @@ class CitusHandler(Citus, AbstractMPPHandler, Thread):
             with self._condition:
                 if self._tasks:
                     if update_cache:
-                        self._pg_dist_node[task.group] = task
+                        self._pg_dist_group[task.groupid] = task
 
                     if update_cache is False:  # an indicator that process_tasks has started a transaction
                         self._in_flight = task
@@ -302,7 +605,7 @@ class CitusHandler(Citus, AbstractMPPHandler, Thread):
         while True:
             try:
                 with self._condition:
-                    if self._schedule_load_pg_dist_node:
+                    if self._schedule_load_pg_dist_group:
                         timeout = -1
                     elif self._in_flight:
                         timeout = self._in_flight.deadline - time.time() if self._tasks else None
@@ -319,9 +622,9 @@ class CitusHandler(Citus, AbstractMPPHandler, Thread):
             except Exception:
                 logger.exception('run')
 
-    def _add_task(self, task: PgDistNode) -> bool:
+    def _add_task(self, task: PgDistTask) -> bool:
         with self._condition:
-            i = self.find_task_by_group(task.group)
+            i = self.find_task_by_groupid(task.groupid)
 
             # The `PgDistNode.timeout` == None is an indicator that it was scheduled from the sync_meta_data().
             if task.timeout is None:
@@ -333,37 +636,48 @@ class CitusHandler(Citus, AbstractMPPHandler, Thread):
                 # key is updated in DCS. Therefore it is possible that :func:`sync_meta_data` will try to create a task
                 # based on the outdated values of "state"/"role". To solve it we introduce an artificial timeout.
                 # Only when the timeout is reached new tasks could be scheduled from sync_meta_data()
-                if self._in_flight and self._in_flight.group == task.group and self._in_flight.timeout is not None\
+                if self._in_flight and self._in_flight.groupid == task.groupid and self._in_flight.timeout is not None\
                         and self._in_flight.deadline > time.time():
                     return False
 
-            # Override already existing task for the same worker group
+            # Override already existing task for the same worker groupid
             if i is not None:
                 if task != self._tasks[i]:
                     logger.debug('Overriding existing task: %s != %s', self._tasks[i], task)
                     self._tasks[i] = task
                     self._condition.notify()
                     return True
-            # Add the task to the list if Worker node state is different from the cached `pg_dist_node`
+            # Add the task to the list if Worker node state is different from the cached `pg_dist_group`
-            elif self._schedule_load_pg_dist_node or task != self._pg_dist_node.get(task.group)\
+            elif self._schedule_load_pg_dist_group or task != self._pg_dist_group.get(task.groupid)\
-                    or self._in_flight and task.group == self._in_flight.group:
+                    or self._in_flight and task.groupid == self._in_flight.groupid:
                 logger.debug('Adding the new task: %s', task)
                 self._tasks.append(task)
                 self._condition.notify()
                 return True
         return False
 
-    def add_task(self, event: str, group: int, conn_url: str,
+    @staticmethod
-                 timeout: Optional[float] = None, cooldown: Optional[float] = None) -> Optional[PgDistNode]:
+    def _pg_dist_node(role: str, conn_url: str) -> Optional[PgDistNode]:
         try:
             r = urlparse(conn_url)
+            if r.hostname:
+                return PgDistNode(r.hostname, r.port or 5432, role)
         except Exception as e:
-            return logger.error('Failed to parse connection url %s: %r', conn_url, e)
+            logger.error('Failed to parse connection url %s: %r', conn_url, e)
-        host = r.hostname
+
-        if host:
+    def add_task(self, event: str, groupid: int, cluster: Cluster, leader_name: str, leader_url: str,
-            port = r.port or 5432
+                 timeout: Optional[float] = None, cooldown: Optional[float] = None) -> Optional[PgDistTask]:
-            task = PgDistNode(group, host, port, event, timeout=timeout, cooldown=cooldown)
+        primary = self._pg_dist_node('demoted' if event == 'before_demote' else 'primary', leader_url)
-            return task if self._add_task(task) else None
+        if not primary:
+            return
+
+        task = PgDistTask(groupid, {primary}, event=event, timeout=timeout, cooldown=cooldown)
+        for member in cluster.members:
+            secondary = self._pg_dist_node('secondary', member.conn_url)\
+                if member.name != leader_name and member.is_running and member.conn_url else None
+            if secondary:
+                task.add(secondary)
+        return task if self._add_task(task) else None
 
     def handle_event(self, cluster: Cluster, event: Dict[str, Any]) -> None:
         if not self.is_alive():
@@ -371,10 +685,10 @@ class CitusHandler(Citus, AbstractMPPHandler, Thread):
 
         worker = cluster.workers.get(event['group'])
         if not (worker and worker.leader and worker.leader.name == event['leader'] and worker.leader.conn_url):
-            return
+            return logger.info('Discarding event %s', event)
 
-        task = self.add_task(event['type'], event['group'],
+        task = self.add_task(event['type'], event['group'], worker,
-                             worker.leader.conn_url,
+                             worker.leader.name, worker.leader.conn_url,
                              event['timeout'], event['cooldown'] * 1000)
         if task and event['type'] == 'before_demote':
             task.wait()

patroni/postgresql/rewind.py

@@ -13,6 +13,7 @@ from . import Postgresql
 from .connection import get_connection_cursor
 from .misc import format_lsn, fsync_dir, parse_history, parse_lsn
 from ..async_executor import CriticalTask
+from ..collections import EMPTY_DICT
 from ..dcs import Leader, RemoteMember
 
 logger = logging.getLogger(__name__)
@@ -418,7 +419,7 @@ class Rewind(object):
         dsn = self._postgresql.config.format_dsn(r, True)
         logger.info('running pg_rewind from %s', dsn)
 
-        restore_command = (self._postgresql.config.get('recovery_conf') or {}).get('restore_command') \
+        restore_command = (self._postgresql.config.get('recovery_conf') or EMPTY_DICT).get('restore_command') \
             if self._postgresql.major_version < 120000 else self._postgresql.get_guc_value('restore_command')
 
         # Until v15 pg_rewind expected postgresql.conf to be inside $PGDATA, which is not the case on e.g. Debian

patroni/psycopg.py

@@ -42,7 +42,8 @@ try:
             value.prepare(conn)
         return value.getquoted().decode('utf-8')
 except ImportError:
-    from psycopg import connect as __connect, sql, Error, DatabaseError, OperationalError, ProgrammingError
+    from psycopg import connect as __connect  # pyright: ignore [reportUnknownVariableType]
+    from psycopg import sql, Error, DatabaseError, OperationalError, ProgrammingError
 
     def _connect(dsn: Optional[str] = None, **kwargs: Any) -> 'Connection[Any]':
         """Call :func:`psycopg.connect` with *dsn* and ``**kwargs``.
@@ -56,7 +57,7 @@ except ImportError:
 
         :returns: a connection to the database.
         """
-        ret = __connect(dsn or "", **kwargs)
+        ret: 'Connection[Any]' = __connect(dsn or "", **kwargs)
         setattr(ret, 'server_version', ret.pgconn.server_version)  # compatibility with psycopg2
         return ret
 

patroni/validator.py

@@ -11,8 +11,7 @@ import socket
 
 from typing import Any, Dict, Union, Iterator, List, Optional as OptionalType, Tuple, TYPE_CHECKING
 
-from .collections import CaseInsensitiveSet
+from .collections import CaseInsensitiveSet, EMPTY_DICT
-
 from .dcs import dcs_modules
 from .exceptions import ConfigParseError
 from .utils import parse_int, split_host_port, data_directory_is_empty, get_major_version
@@ -245,7 +244,7 @@ def get_bin_name(bin_name: str) -> str:
     """
     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)
+    return (schema.data.get('postgresql', {}).get('bin_name', {}) or EMPTY_DICT).get(bin_name, bin_name)
 
 
 def validate_data_dir(data_dir: str) -> bool:

tests/__init__.py

@@ -179,8 +179,12 @@ class MockCursor(object):
                                  b'3\t0/403DD98\tno recovery target specified\n')]
         elif sql.startswith('SELECT pg_catalog.citus_add_node'):
             self.results = [(2,)]
-        elif sql.startswith('SELECT nodeid, groupid'):
+        elif sql.startswith('SELECT groupid, nodename'):
-            self.results = [(1, 0, 'host1', 5432, 'primary'), (2, 1, 'host2', 5432, 'primary')]
+            self.results = [(0, 'host1', 5432, 'primary', 1),
+                            (0, '127.0.0.1', 5436, 'secondary', 2),
+                            (1, 'host4', 5432, 'primary', 3),
+                            (1, '127.0.0.1', 5437, 'secondary', 4),
+                            (1, '127.0.0.1', 5438, 'secondary', 5)]
         else:
             self.results = [(None, None, None, None, None, None, None, None, None, None)]
         self.rowcount = len(self.results)

tests/test_citus.py

@@ -1,6 +1,10 @@
 import time
+import unittest
+
+from copy import deepcopy
 from mock import Mock, patch, PropertyMock
-from patroni.postgresql.mpp.citus import CitusHandler
+from typing import List
+from patroni.postgresql.mpp.citus import CitusHandler, PgDistGroup, PgDistNode
 from patroni.psycopg import ProgrammingError
 
 from . import BaseTestPostgresql, MockCursor, psycopg_connect, SleepException
@@ -20,7 +24,7 @@ class TestCitus(BaseTestPostgresql):
     @patch('time.time', Mock(side_effect=[100, 130, 160, 190, 220, 250, 280, 310, 340, 370]))
     @patch('patroni.postgresql.mpp.citus.logger.exception', Mock(side_effect=SleepException))
     @patch('patroni.postgresql.mpp.citus.logger.warning')
-    @patch('patroni.postgresql.mpp.citus.PgDistNode.wait', Mock())
+    @patch('patroni.postgresql.mpp.citus.PgDistTask.wait', Mock())
     @patch.object(CitusHandler, 'is_alive', Mock(return_value=True))
     def test_run(self, mock_logger_warning):
         # `before_demote` or `before_promote` REST API calls starting a
@@ -32,11 +36,11 @@ class TestCitus(BaseTestPostgresql):
 
         self.c.handle_event(self.cluster, {'type': 'before_demote', 'group': 1,
                                            'leader': 'leader', 'timeout': 30, 'cooldown': 10})
-        self.c.add_task('after_promote', 2, 'postgres://host3:5432/postgres')
+        self.c.add_task('after_promote', 2, self.cluster, self.cluster.leader_name, 'postgres://host3:5432/postgres')
         self.assertRaises(SleepException, self.c.run)
         mock_logger_warning.assert_called_once()
         self.assertTrue(mock_logger_warning.call_args[0][0].startswith('Rolling back transaction'))
-        self.assertTrue(repr(mock_logger_warning.call_args[0][1]).startswith('PgDistNode'))
+        self.assertTrue(repr(mock_logger_warning.call_args[0][1]).startswith('PgDistTask'))
 
     @patch.object(CitusHandler, 'is_alive', Mock(return_value=False))
     @patch.object(CitusHandler, 'start', Mock())
@@ -54,59 +58,68 @@ class TestCitus(BaseTestPostgresql):
     def test_add_task(self):
         with patch('patroni.postgresql.mpp.citus.logger.error') as mock_logger, \
                 patch('patroni.postgresql.mpp.citus.urlparse', Mock(side_effect=Exception)):
-            self.c.add_task('', 1, None)
+            self.c.add_task('', 1, self.cluster, '', None)
             mock_logger.assert_called_once()
 
         with patch('patroni.postgresql.mpp.citus.logger.debug') as mock_logger:
-            self.c.add_task('before_demote', 1, 'postgres://host:5432/postgres', 30)
+            self.c.add_task('before_demote', 1, self.cluster,
+                            self.cluster.leader_name, 'postgres://host:5432/postgres', 30)
             mock_logger.assert_called_once()
             self.assertTrue(mock_logger.call_args[0][0].startswith('Adding the new task:'))
 
         with patch('patroni.postgresql.mpp.citus.logger.debug') as mock_logger:
-            self.c.add_task('before_promote', 1, 'postgres://host:5432/postgres', 30)
+            self.c.add_task('before_promote', 1, self.cluster,
+                            self.cluster.leader_name, 'postgres://host:5432/postgres', 30)
             mock_logger.assert_called_once()
             self.assertTrue(mock_logger.call_args[0][0].startswith('Overriding existing task:'))
 
-        # add_task called from sync_meta_data should not override already scheduled or in flight task until deadline
+        # add_task called from sync_pg_dist_node should not override already scheduled or in flight task until deadline
-        self.assertIsNotNone(self.c.add_task('after_promote', 1, 'postgres://host:5432/postgres', 30))
+        self.assertIsNotNone(self.c.add_task('after_promote', 1, self.cluster,
-        self.assertIsNone(self.c.add_task('after_promote', 1, 'postgres://host:5432/postgres'))
+                                             self.cluster.leader_name, 'postgres://host:5432/postgres', 30))
+        self.assertIsNone(self.c.add_task('after_promote', 1, self.cluster,
+                                          self.cluster.leader_name, 'postgres://host:5432/postgres'))
         self.c._in_flight = self.c._tasks.pop()
         self.c._in_flight.deadline = self.c._in_flight.timeout + time.time()
-        self.assertIsNone(self.c.add_task('after_promote', 1, 'postgres://host:5432/postgres'))
+        self.assertIsNone(self.c.add_task('after_promote', 1, self.cluster,
+                                          self.cluster.leader_name, 'postgres://host:5432/postgres'))
         self.c._in_flight.deadline = 0
-        self.assertIsNotNone(self.c.add_task('after_promote', 1, 'postgres://host:5432/postgres'))
+        self.assertIsNotNone(self.c.add_task('after_promote', 1, self.cluster,
+                                             self.cluster.leader_name, 'postgres://host:5432/postgres'))
 
         # If there is no transaction in progress and cached pg_dist_node matching desired state task should not be added
         self.c._schedule_load_pg_dist_node = False
-        self.c._pg_dist_node[self.c._in_flight.group] = self.c._in_flight
+        self.c._pg_dist_group[self.c._in_flight.groupid] = self.c._in_flight
         self.c._in_flight = None
-        self.assertIsNone(self.c.add_task('after_promote', 1, 'postgres://host:5432/postgres'))
+        self.assertIsNone(self.c.add_task('after_promote', 1, self.cluster,
+                                          self.cluster.leader_name, 'postgres://host:5432/postgres'))
 
     def test_pick_task(self):
-        self.c.add_task('after_promote', 1, 'postgres://host2:5432/postgres')
+        self.c.add_task('after_promote', 0, self.cluster, self.cluster.leader_name, 'postgres://host1:5432/postgres')
-        with patch.object(CitusHandler, 'process_task') as mock_process_task:
+        with patch.object(CitusHandler, 'update_node') as mock_update_node:
             self.c.process_tasks()
-            # process_task() shouln't be called because pick_task double checks with _pg_dist_node
+            # process_task() shouln't be called because pick_task double checks with _pg_dist_group
-            mock_process_task.assert_not_called()
+            mock_update_node.assert_not_called()
 
     def test_process_task(self):
-        self.c.add_task('after_promote', 0, 'postgres://host2:5432/postgres')
+        self.c.add_task('after_promote', 1, self.cluster, self.cluster.leader_name, 'postgres://host2:5432/postgres')
-        task = self.c.add_task('before_promote', 1, 'postgres://host4:5432/postgres', 30)
+        task = self.c.add_task('before_promote', 1, self.cluster,
+                               self.cluster.leader_name, 'postgres://host4:5432/postgres', 30)
         self.c.process_tasks()
         self.assertTrue(task._event.is_set())
 
         # the after_promote should result only in COMMIT
-        task = self.c.add_task('after_promote', 1, 'postgres://host4:5432/postgres', 30)
+        task = self.c.add_task('after_promote', 1, self.cluster,
+                               self.cluster.leader_name, 'postgres://host4:5432/postgres', 30)
         with patch.object(CitusHandler, 'query') as mock_query:
             self.c.process_tasks()
             mock_query.assert_called_once()
             self.assertEqual(mock_query.call_args[0][0], 'COMMIT')
 
     def test_process_tasks(self):
-        self.c.add_task('after_promote', 0, 'postgres://host2:5432/postgres')
+        self.c.add_task('after_promote', 0, self.cluster, self.cluster.leader_name, 'postgres://host2:5432/postgres')
         self.c.process_tasks()
 
-        self.c.add_task('after_promote', 0, 'postgres://host3:5432/postgres')
+        self.c.add_task('after_promote', 0, self.cluster, self.cluster.leader_name, 'postgres://host3:5432/postgres')
         with patch('patroni.postgresql.mpp.citus.logger.error') as mock_logger, \
                 patch.object(CitusHandler, 'query', Mock(side_effect=Exception)):
             self.c.process_tasks()
@@ -118,16 +131,17 @@ class TestCitus(BaseTestPostgresql):
 
     @patch('patroni.postgresql.mpp.citus.logger.error')
     @patch.object(MockCursor, 'execute', Mock(side_effect=Exception))
-    def test_load_pg_dist_node(self, mock_logger):
+    def test_load_pg_dist_group(self, mock_logger):
-        # load_pg_dist_node() triggers, query fails and exception is property handled
+        # load_pg_dist_group) triggers, query fails and exception is property handled
         self.c.process_tasks()
-        self.assertTrue(self.c._schedule_load_pg_dist_node)
+        self.assertTrue(self.c._schedule_load_pg_dist_group)
         mock_logger.assert_called_once()
         self.assertTrue(mock_logger.call_args[0][0].startswith('Exception when executing query'))
-        self.assertTrue(mock_logger.call_args[0][1].startswith('SELECT nodeid, groupid, '))
+        self.assertTrue(mock_logger.call_args[0][1].startswith('SELECT groupid, nodename, '))
 
     def test_wait(self):
-        task = self.c.add_task('before_demote', 1, 'postgres://host:5432/postgres', 30)
+        task = self.c.add_task('before_demote', 1, self.cluster,
+                               self.cluster.leader_name, u'postgres://host:5432/postgres', 30)
         task._event.wait = Mock()
         task.wait()
 
@@ -171,3 +185,209 @@ class TestCitus(BaseTestPostgresql):
             self.c.bootstrap()
         mock_logger.assert_called_once()
         self.assertTrue(mock_logger.call_args[0][0].startswith('Exception when creating database'))
+
+
+class TestGroupTransition(unittest.TestCase):
+    nodeid = 100
+
+    def map_to_sql(self, group: int, transition: PgDistNode) -> str:
+        if transition.role not in ('primary', 'demoted', 'secondary'):
+            return "citus_remove_node('{0}', {1})".format(transition.host, transition.port)
+        elif transition.nodeid:
+            host = transition.host + ('-demoted' if transition.role == 'demoted' else '')
+            return "citus_update_node({0}, '{1}', {2})".format(transition.nodeid, host, transition.port)
+        else:
+            transition.nodeid = self.nodeid
+            self.nodeid += 1
+
+            return "citus_add_node('{0}', {1}, {2}, '{3}')".format(transition.host, transition.port,
+                                                                   group, transition.role)
+
+    def check_transitions(self, old_topology: PgDistGroup, new_topology: PgDistGroup,
+                          expected_transitions: List[str]) -> None:
+        check_topology = deepcopy(old_topology)
+
+        transitions: List[str] = []
+        for node in new_topology.transition(old_topology):
+            self.assertTrue(node not in check_topology or (check_topology.get(node) or node).role == 'demoted')
+            old_node = node.nodeid and next(iter(v for v in check_topology if v.nodeid == node.nodeid), None)
+            if old_node:
+                check_topology.discard(old_node)
+            transitions.append(self.map_to_sql(new_topology.groupid, node))
+            check_topology.add(node)
+        self.assertEqual(transitions, expected_transitions)
+
+    def test_new_topology(self):
+        old = PgDistGroup(0)
+        new = PgDistGroup(0, {PgDistNode('1', 5432, 'primary'),
+                              PgDistNode('2', 5432, 'secondary')})
+        expected = PgDistGroup(0, {PgDistNode('1', 5432, 'primary', nodeid=100),
+                                   PgDistNode('2', 5432, 'secondary', nodeid=101)})
+        self.check_transitions(old, new,
+                               ["citus_add_node('1', 5432, 0, 'primary')",
+                                "citus_add_node('2', 5432, 0, 'secondary')"])
+        self.assertTrue(new.equals(expected, True))
+
+    def test_switchover(self):
+        old = PgDistGroup(0, {PgDistNode('1', 5432, 'primary', nodeid=1),
+                              PgDistNode('2', 5432, 'secondary', nodeid=2)})
+        new = PgDistGroup(0, {PgDistNode('1', 5432, 'secondary'),
+                              PgDistNode('2', 5432, 'primary')})
+        expected = PgDistGroup(0, {PgDistNode('2', 5432, 'primary', nodeid=1),
+                                   PgDistNode('1', 5432, 'secondary', nodeid=2)})
+        self.check_transitions(old, new,
+                               ["citus_update_node(1, '1-demoted', 5432)",
+                                "citus_update_node(2, '1', 5432)",
+                                "citus_update_node(1, '2', 5432)"])
+        self.assertTrue(new.equals(expected, True))
+
+    def test_failover(self):
+        old = PgDistGroup(0, {PgDistNode('1', 5432, 'primary', nodeid=1),
+                              PgDistNode('2', 5432, 'secondary', nodeid=2)})
+        new = PgDistGroup(0, {PgDistNode('2', 5432, 'primary')})
+        expected = PgDistGroup(0, {PgDistNode('2', 5432, 'primary', nodeid=1),
+                                   PgDistNode('1', 5432, 'secondary', nodeid=2)})
+        self.check_transitions(old, new,
+                               ["citus_update_node(1, '1-demoted', 5432)",
+                                "citus_update_node(2, '1', 5432)",
+                                "citus_update_node(1, '2', 5432)"])
+        self.assertTrue(new.equals(expected, True))
+
+    def test_failover_and_new_secondary(self):
+        old = PgDistGroup(0, {PgDistNode('1', 5432, 'primary', nodeid=1),
+                              PgDistNode('2', 5432, 'secondary', nodeid=2)})
+        new = PgDistGroup(0, {PgDistNode('2', 5432, 'primary'),
+                              PgDistNode('3', 5432, 'secondary')})
+        expected = PgDistGroup(0, {PgDistNode('2', 5432, 'primary', nodeid=1),
+                                   PgDistNode('3', 5432, 'secondary', nodeid=2)})
+        # the secondary record is used to add the new standby and primary record is updated with the new hostname
+        self.check_transitions(old, new, ["citus_update_node(2, '3', 5432)", "citus_update_node(1, '2', 5432)"])
+        self.assertTrue(new.equals(expected, True))
+
+    def test_switchover_and_new_secondary_primary_gone(self):
+        old = PgDistGroup(0, {PgDistNode('1', 5432, 'demoted', nodeid=1),
+                              PgDistNode('2', 5432, 'secondary', nodeid=2)})
+        new = PgDistGroup(0, {PgDistNode('2', 5432, 'primary'),
+                              PgDistNode('3', 5432, 'secondary')})
+        expected = PgDistGroup(0, {PgDistNode('2', 5432, 'primary', nodeid=1),
+                                   PgDistNode('3', 5432, 'secondary', nodeid=2)})
+        # the secondary record is used to add the new standby and primary record is updated with the new hostname
+        self.check_transitions(old, new, ["citus_update_node(2, '3', 5432)", "citus_update_node(1, '2', 5432)"])
+        self.assertTrue(new.equals(expected, True))
+
+    def test_secondary_replaced(self):
+        old = PgDistGroup(0, {PgDistNode('1', 5432, 'primary', nodeid=1),
+                              PgDistNode('2', 5432, 'secondary', nodeid=2)})
+        new = PgDistGroup(0, {PgDistNode('1', 5432, 'primary'),
+                              PgDistNode('3', 5432, 'secondary')})
+        expected = PgDistGroup(0, {PgDistNode('1', 5432, 'primary', nodeid=1),
+                                   PgDistNode('3', 5432, 'secondary', nodeid=2)})
+        self.check_transitions(old, new, ["citus_update_node(2, '3', 5432)"])
+        self.assertTrue(new.equals(expected, True))
+
+    def test_secondary_repmoved(self):
+        old = PgDistGroup(0, {PgDistNode('1', 5432, 'primary', nodeid=1),
+                              PgDistNode('2', 5432, 'secondary', nodeid=2),
+                              PgDistNode('3', 5432, 'secondary', nodeid=3)})
+        new = PgDistGroup(0, {PgDistNode('1', 5432, 'primary'),
+                              PgDistNode('3', 5432, 'secondary')})
+        expected = PgDistGroup(0, {PgDistNode('1', 5432, 'primary', nodeid=1),
+                                   PgDistNode('3', 5432, 'secondary', nodeid=3)})
+        self.check_transitions(old, new, ["citus_remove_node('2', 5432)"])
+        self.assertTrue(new.equals(expected, True))
+
+    def test_switchover_and_secondary_removed(self):
+        old = PgDistGroup(0, {PgDistNode('1', 5432, 'primary', nodeid=1),
+                              PgDistNode('2', 5432, 'secondary', nodeid=2),
+                              PgDistNode('3', 5432, 'secondary', nodeid=3)})
+        new = PgDistGroup(0, {PgDistNode('1', 5432, 'secondary'),
+                              PgDistNode('2', 5432, 'primary')})
+        expected = PgDistGroup(0, {PgDistNode('1', 5432, 'secondary', nodeid=2),
+                                   PgDistNode('2', 5432, 'primary', nodeid=1),
+                                   PgDistNode('3', 5432, 'secondary', nodeid=3)})
+        self.check_transitions(old, new,
+                               ["citus_update_node(1, '1-demoted', 5432)",
+                                "citus_update_node(2, '1', 5432)",
+                                "citus_update_node(1, '2', 5432)"])
+        self.assertTrue(new.equals(expected, True))
+
+    def test_switchover_and_new_secondary(self):
+        old = PgDistGroup(0, {PgDistNode('1', 5432, 'primary', nodeid=1),
+                              PgDistNode('2', 5432, 'secondary', nodeid=2)})
+        new = PgDistGroup(0, {PgDistNode('1', 5432, 'secondary'),
+                              PgDistNode('2', 5432, 'primary'),
+                              PgDistNode('3', 5432, 'secondary')})
+        expected = PgDistGroup(0, {PgDistNode('1', 5432, 'secondary', nodeid=2),
+                                   PgDistNode('2', 5432, 'primary', nodeid=1)})
+        self.check_transitions(old, new,
+                               ["citus_update_node(1, '1-demoted', 5432)",
+                                "citus_update_node(2, '1', 5432)",
+                                "citus_update_node(1, '2', 5432)"])
+        self.assertTrue(new.equals(expected, True))
+
+    def test_failover_to_new_node_secondary_remains(self):
+        old = PgDistGroup(0, {PgDistNode('1', 5432, 'primary', nodeid=1),
+                              PgDistNode('2', 5432, 'secondary', nodeid=2)})
+        new = PgDistGroup(0, {PgDistNode('2', 5432, 'secondary'),
+                              PgDistNode('3', 5432, 'primary')})
+        expected = PgDistGroup(0, {PgDistNode('3', 5432, 'primary', nodeid=1),
+                                   PgDistNode('2', 5432, 'secondary', nodeid=2)})
+        self.check_transitions(old, new, ["citus_update_node(1, '3', 5432)"])
+        self.assertTrue(new.equals(expected, True))
+
+    def test_failover_to_new_node_secondary_removed(self):
+        old = PgDistGroup(0, {PgDistNode('1', 5432, 'primary', nodeid=1),
+                              PgDistNode('2', 5432, 'secondary', nodeid=2)})
+        new = PgDistGroup(0, {PgDistNode('3', 5432, 'primary')})
+        expected = PgDistGroup(0, {PgDistNode('3', 5432, 'primary', nodeid=1),
+                                   PgDistNode('2', 5432, 'secondary', nodeid=2)})
+        # the secondary record needs to be removed before we update the primary record
+        self.check_transitions(old, new, ["citus_update_node(1, '3', 5432)"])
+        self.assertTrue(new.equals(expected, True))
+
+    def test_switchover_to_new_node_and_secondary_removed(self):
+        old = PgDistGroup(0, {PgDistNode('1', 5432, 'primary', nodeid=1),
+                              PgDistNode('2', 5432, 'secondary', nodeid=2)})
+        new = PgDistGroup(0, {PgDistNode('1', 5432, 'secondary'),
+                              PgDistNode('3', 5432, 'primary')})
+        expected = PgDistGroup(0, {PgDistNode('3', 5432, 'primary', nodeid=1),
+                                   PgDistNode('1', 5432, 'secondary', nodeid=2)})
+        self.check_transitions(old, new, ["citus_update_node(1, '3', 5432)", "citus_update_node(2, '1', 5432)"])
+        self.assertTrue(new.equals(expected, True))
+
+    def test_switchover_with_pause(self):
+        old = PgDistGroup(0, {PgDistNode('1', 5432, 'primary', nodeid=1),
+                              PgDistNode('2', 5432, 'secondary', nodeid=2)})
+        new = PgDistGroup(0, {PgDistNode('1', 5432, 'demoted')})
+        expected = PgDistGroup(0, {PgDistNode('1', 5432, 'demoted', nodeid=1),
+                                   PgDistNode('2', 5432, 'secondary', nodeid=2)})
+        self.check_transitions(old, new, ["citus_update_node(1, '1-demoted', 5432)"])
+        self.assertTrue(new.equals(expected, True))
+
+    def test_switchover_after_paused_connections(self):
+        old = PgDistGroup(0, {PgDistNode('1', 5432, 'demoted', nodeid=1),
+                              PgDistNode('2', 5432, 'secondary', nodeid=2)})
+        new = PgDistGroup(0, {PgDistNode('2', 5432, 'primary')})
+        expected = PgDistGroup(0, {PgDistNode('1', 5432, 'secondary', nodeid=2),
+                                   PgDistNode('2', 5432, 'primary', nodeid=1)})
+        self.check_transitions(old, new, ["citus_update_node(2, '1', 5432)", "citus_update_node(1, '2', 5432)"])
+        self.assertTrue(new.equals(expected, True))
+
+    def test_switchover_to_new_node_after_paused_connections(self):
+        old = PgDistGroup(0, {PgDistNode('1', 5432, 'demoted', nodeid=1),
+                              PgDistNode('2', 5432, 'secondary', nodeid=2)})
+        new = PgDistGroup(0, {PgDistNode('3', 5432, 'primary')})
+        expected = PgDistGroup(0, {PgDistNode('1', 5432, 'secondary', nodeid=2),
+                                   PgDistNode('3', 5432, 'primary', nodeid=1)})
+        self.check_transitions(old, new, ["citus_update_node(1, '3', 5432)", "citus_update_node(2, '1', 5432)"])
+        self.assertTrue(new.equals(expected, True))
+
+    def test_switchover_to_new_node_after_paused_connections_secondary_added(self):
+        old = PgDistGroup(0, {PgDistNode('1', 5432, 'demoted', nodeid=1),
+                              PgDistNode('2', 5432, 'secondary', nodeid=2)})
+        new = PgDistGroup(0, {PgDistNode('4', 5432, 'secondary'),
+                              PgDistNode('3', 5432, 'primary')})
+        expected = PgDistGroup(0, {PgDistNode('4', 5432, 'secondary', nodeid=2),
+                                   PgDistNode('3', 5432, 'primary', nodeid=1)})
+        self.check_transitions(old, new, ["citus_update_node(1, '3', 5432)", "citus_update_node(2, '4', 5432)"])
+        self.assertTrue(new.equals(expected, True))

tests/test_config.py

@@ -160,12 +160,10 @@ 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"""
-        config = Config("postgres0.yml")
-
         # Providing one of `nofailover` or `failover_priority` is fine
         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())
+            self.assertIsNone(self.config._validate_failover_tags())
             mock_logger.warning.assert_not_called()
 
         # Providing both `nofailover` and `failover_priority` is fine if consistent
@@ -175,7 +173,7 @@ class TestConfig(unittest.TestCase):
             {"nofailover": "False", "failover_priority": 0}
         ):
             mock_get.side_effect = [consistent_state] * 2
-            self.assertIsNone(config._validate_failover_tags())
+            self.assertIsNone(self.config._validate_failover_tags())
             mock_logger.warning.assert_not_called()
 
         # Providing both inconsistently should log a warning
@@ -186,7 +184,7 @@ class TestConfig(unittest.TestCase):
             {"nofailover": "", "failover_priority": 0}
         ):
             mock_get.side_effect = [inconsistent_state] * 2
-            self.assertIsNone(config._validate_failover_tags())
+            self.assertIsNone(self.config._validate_failover_tags())
             mock_logger.warning.assert_called_once_with(
                 'Conflicting configuration between nofailover: %s and failover_priority: %s.'
                 + ' Defaulting to nofailover: %s',

tests/test_kubernetes.py

@@ -80,7 +80,7 @@ def mock_namespaced_kind(*args, **kwargs):
 
 
 def mock_load_k8s_config(self, *args, **kwargs):
-    self._server = ''
+    self._server = 'http://localhost'
 
 
 class TestK8sConfig(unittest.TestCase):
@@ -242,6 +242,7 @@ class BaseTestKubernetes(unittest.TestCase):
         self.k.get_cluster()
 
 
+@patch('urllib3.PoolManager.request', Mock())
 @patch.object(k8s_client.CoreV1Api, 'patch_namespaced_config_map', mock_namespaced_kind, create=True)
 class TestKubernetesConfigMaps(BaseTestKubernetes):
 
@@ -374,6 +375,7 @@ class TestKubernetesConfigMaps(BaseTestKubernetes):
         mock_warning.assert_called_once()
 
 
+@patch('urllib3.PoolManager.request', Mock())
 class TestKubernetesEndpointsNoPodIP(BaseTestKubernetes):
     @patch.object(k8s_client.CoreV1Api, 'list_namespaced_endpoints', mock_list_namespaced_endpoints, create=True)
     def setUp(self, config=None):
@@ -388,6 +390,7 @@ class TestKubernetesEndpointsNoPodIP(BaseTestKubernetes):
         self.assertEqual(args[2].subsets[0].addresses[0].ip, '10.0.0.1')
 
 
+@patch('urllib3.PoolManager.request', Mock())
 class TestKubernetesEndpoints(BaseTestKubernetes):
 
     @patch.object(k8s_client.CoreV1Api, 'list_namespaced_endpoints', mock_list_namespaced_endpoints, create=True)
@@ -478,6 +481,7 @@ def mock_watch(*args):
     return urllib3.HTTPResponse()
 
 
+@patch('urllib3.PoolManager.request', Mock())
 class TestCacheBuilder(BaseTestKubernetes):
 
     @patch.object(k8s_client.CoreV1Api, 'list_namespaced_config_map', mock_list_namespaced_config_map, create=True)