From 35c97fa402ad2ff135493dfcb1c58e7169041794 Mon Sep 17 00:00:00 2001
From: Alexander Kukushkin <cyberdemn@gmail.com>
Date: Mon, 10 Jul 2023 09:19:10 +0200
Subject: [PATCH 01/11] Make sure the version_prefix for etcd3 is set to
 /v3beta (#2729)

Setting it before calling the parent constructor didn't really work because it is being overwritten in the `etcd.Client.__init__()`.

The only viable way of doing it is passing a custom value to the parent class.

Close https://github.com/zalando/patroni/issues/2142
---
 patroni/dcs/etcd.py  | 5 ++++-
 patroni/dcs/etcd3.py | 3 +--
 2 files changed, 5 insertions(+), 3 deletions(-)

diff --git a/patroni/dcs/etcd.py b/patroni/dcs/etcd.py
index 3e7f0681..5f667771 100644
--- a/patroni/dcs/etcd.py
+++ b/patroni/dcs/etcd.py
@@ -99,7 +99,7 @@ class AbstractEtcdClientWithFailover(abc.ABC, etcd.Client):
         self._dns_resolver = dns_resolver
         self.set_machines_cache_ttl(cache_ttl)
         self._machines_cache_updated = 0
-        kwargs = {p: config.get(p) for p in ('host', 'port', 'protocol', 'use_proxies',
+        kwargs = {p: config.get(p) for p in ('host', 'port', 'protocol', 'use_proxies', 'version_prefix',
                                              'username', 'password', 'cert', 'ca_cert') if config.get(p)}
         super(AbstractEtcdClientWithFailover, self).__init__(read_timeout=config['retry_timeout'], **kwargs)
         # For some reason python3-etcd on debian and ubuntu are not based on the latest version
@@ -443,6 +443,9 @@ class EtcdClient(AbstractEtcdClientWithFailover):
 
     ERROR_CLS = EtcdError
 
+    def __init__(self, config: Dict[str, Any], dns_resolver: DnsCachingResolver, cache_ttl: int = 300) -> None:
+        super(EtcdClient, self).__init__({**config, 'version_prefix': None}, dns_resolver, cache_ttl)
+
     def __del__(self) -> None:
         try:
             self.http.clear()
diff --git a/patroni/dcs/etcd3.py b/patroni/dcs/etcd3.py
index b89e36a5..efb955e0 100644
--- a/patroni/dcs/etcd3.py
+++ b/patroni/dcs/etcd3.py
@@ -206,8 +206,7 @@ class Etcd3Client(AbstractEtcdClientWithFailover):
     def __init__(self, config: Dict[str, Any], dns_resolver: DnsCachingResolver, cache_ttl: int = 300) -> None:
         self._token = None
         self._cluster_version: Tuple[int] = tuple()
-        self.version_prefix = '/v3beta'
-        super(Etcd3Client, self).__init__(config, dns_resolver, cache_ttl)
+        super(Etcd3Client, self).__init__({**config, 'version_prefix': '/v3beta'}, dns_resolver, cache_ttl)
 
         try:
             self.authenticate()

From 3c1b274ab72f705faca9c2d47f641bc343eb734f Mon Sep 17 00:00:00 2001
From: Alexander Kukushkin <cyberdemn@gmail.com>
Date: Mon, 10 Jul 2023 09:19:43 +0200
Subject: [PATCH 02/11] Use quorum read in patronictl if it is possible (#2730)

implementations and terminologis are DCS specific:
- Etcd v2 calls is `quorum` read
- Etcd v3 calls it `linearizable` (vs `serializable`)
- Consul calls it `consistent`

Following DCS don't offer this feature:
- ZooKeeper calls it linearizable, but reads are sequentially consistent
- Raft - no quorum reads are possible ATM
- Kubernetes - uses Etcd under the hood, but provides no API to choose read consistency level

Close https://github.com/zalando/patroni/issues/1199
---
 patroni/dcs/consul.py     |  8 ++++++--
 patroni/dcs/etcd.py       |  4 ++--
 patroni/dcs/etcd3.py      | 11 ++++++-----
 patroni/dcs/kubernetes.py |  4 ++--
 4 files changed, 16 insertions(+), 11 deletions(-)

diff --git a/patroni/dcs/consul.py b/patroni/dcs/consul.py
index a62e7747..73755a41 100644
--- a/patroni/dcs/consul.py
+++ b/patroni/dcs/consul.py
@@ -400,8 +400,12 @@ class Consul(AbstractDCS):
 
         return Cluster(initialize, config, leader, last_lsn, members, failover, sync, history, slots, failsafe)
 
+    @property
+    def _consistency(self) -> str:
+        return 'consistent' if self._ctl else self._client.consistency
+
     def _cluster_loader(self, path: str) -> Cluster:
-        _, results = self.retry(self._client.kv.get, path, recurse=True)
+        _, results = self.retry(self._client.kv.get, path, recurse=True, consistency=self._consistency)
         if results is None:
             raise NotFound
         nodes = {}
@@ -412,7 +416,7 @@ class Consul(AbstractDCS):
         return self._cluster_from_nodes(nodes)
 
     def _citus_cluster_loader(self, path: str) -> Dict[int, Cluster]:
-        _, results = self.retry(self._client.kv.get, path, recurse=True)
+        _, results = self.retry(self._client.kv.get, path, recurse=True, consistency=self._consistency)
         clusters: Dict[int, Dict[str, Cluster]] = defaultdict(dict)
         for node in results or []:
             key = node['Key'][len(path):].split('/', 1)
diff --git a/patroni/dcs/etcd.py b/patroni/dcs/etcd.py
index 5f667771..94e546ac 100644
--- a/patroni/dcs/etcd.py
+++ b/patroni/dcs/etcd.py
@@ -725,13 +725,13 @@ class Etcd(AbstractEtcd):
         return Cluster(initialize, config, leader, last_lsn, members, failover, sync, history, slots, failsafe)
 
     def _cluster_loader(self, path: str) -> Cluster:
-        result = self.retry(self._client.read, path, recursive=True)
+        result = self.retry(self._client.read, path, recursive=True, quorum=self._ctl)
         nodes = {node.key[len(result.key):].lstrip('/'): node for node in result.leaves}
         return self._cluster_from_nodes(result.etcd_index, nodes)
 
     def _citus_cluster_loader(self, path: str) -> Dict[int, Cluster]:
         clusters: Dict[int, Dict[str, etcd.EtcdResult]] = defaultdict(dict)
-        result = self.retry(self._client.read, path, recursive=True)
+        result = self.retry(self._client.read, path, recursive=True, quorum=self._ctl)
         for node in result.leaves:
             key = node.key[len(result.key):].lstrip('/').split('/', 1)
             if len(key) == 2 and citus_group_re.match(key[0]):
diff --git a/patroni/dcs/etcd3.py b/patroni/dcs/etcd3.py
index efb955e0..32d00359 100644
--- a/patroni/dcs/etcd3.py
+++ b/patroni/dcs/etcd3.py
@@ -326,14 +326,14 @@ class Etcd3Client(AbstractEtcdClientWithFailover):
             return retry(e)
 
     @_handle_auth_errors
-    def range(self, key: str, range_end: Union[bytes, str, None] = None,
+    def range(self, key: str, range_end: Union[bytes, str, None] = None, serializable: bool = True,
               retry: Optional[Retry] = None) -> Dict[str, Any]:
         params = build_range_request(key, range_end)
-        params['serializable'] = True  # For better performance. We can tolerate stale reads.
+        params['serializable'] = serializable  # For better performance. We can tolerate stale reads
         return self.call_rpc('/kv/range', params, retry)
 
-    def prefix(self, key: str, retry: Optional[Retry] = None) -> Dict[str, Any]:
-        return self.range(key, prefix_range_end(key), retry)
+    def prefix(self, key: str, serializable: bool = True, retry: Optional[Retry] = None) -> Dict[str, Any]:
+        return self.range(key, prefix_range_end(key), serializable, retry)
 
     @_handle_auth_errors
     def lease_grant(self, ttl: int, retry: Optional[Retry] = None) -> str:
@@ -594,7 +594,8 @@ class PatroniEtcd3Client(Etcd3Client):
                 self._wait_cache(self.read_timeout)
                 ret = self._kv_cache.copy()
         else:
-            ret = self._etcd3.retry(self.prefix, path).get('kvs', [])
+            serializable = not getattr(self._etcd3, '_ctl')  # use linearizable for patronictl
+            ret = self._etcd3.retry(self.prefix, path, serializable).get('kvs', [])
         for node in ret:
             node.update({'key': base64_decode(node['key']),
                          'value': base64_decode(node.get('value', '')),
diff --git a/patroni/dcs/kubernetes.py b/patroni/dcs/kubernetes.py
index 498cfd0f..f770ac64 100644
--- a/patroni/dcs/kubernetes.py
+++ b/patroni/dcs/kubernetes.py
@@ -766,7 +766,7 @@ class Kubernetes(AbstractDCS):
             k8s_config.load_kube_config(context=config.get('context', 'kind-kind'))
 
         pod_ip = config.get('pod_ip')
-        self.__ips: List[str] = [] if config.get('patronictl') or not isinstance(pod_ip, str) else [pod_ip]
+        self.__ips: List[str] = [] if self._ctl or not isinstance(pod_ip, str) else [pod_ip]
         self.__ports: List[K8sObject] = []
         ports: List[Dict[str, Any]] = config.get('ports', [{}])
         for p in ports:
@@ -774,7 +774,7 @@ class Kubernetes(AbstractDCS):
             port.update({n: p[n] for n in ('name', 'protocol') if p.get(n)})
             self.__ports.append(k8s_client.V1EndpointPort(**port))
 
-        bypass_api_service = not config.get('patronictl') and config.get('bypass_api_service')
+        bypass_api_service = not self._ctl and config.get('bypass_api_service')
         self._api = CoreV1ApiProxy(config.get('use_endpoints'), bypass_api_service)
         self._should_create_config_service = self._api.use_endpoints
         self.reload_config(config)

From 4725f12f9a24ee5ad6b3ee0c1792de61f7a3a7cc Mon Sep 17 00:00:00 2001
From: Feike Steenbergen <feikesteenbergen@users.noreply.github.com>
Date: Mon, 10 Jul 2023 13:44:54 +0200
Subject: [PATCH 03/11] Allow integer gucs without units in validation (#2734)

Previously, integer gucs, for example `max_connections` would not pass the validation, as these settings have no unit, if and only if they were specified as a string.

This causes problems if the `max_connections` is configured in `patroni.yaml` as a string, for example, the following configuration would not result in the right `max_connections` settings, as `max_connections` is configured as a string:

    bootstrap:
      dcs:
        postgresql:
          parameters:
            log_checkpoints: "on"
            log_connections: "off"
            max_connections: "57"

Allowing a user to specify *all* parameters as a string was accepted before in Patroni and also seems very useful, as many of us will be using Ansible/Helm/Golang to build a Patroni configuration, in which creating a `map[string]string` is easier than having to deal with data types.

Attemps to address issue #2735

Regression was introduced in https://github.com/zalando/patroni/commit/76b3b99de2f2bfaa8ab2df9e47dbfc3749d14e84
---
 features/patroni_api.feature | 12 ++++++------
 patroni/utils.py             | 15 +++++++++++++++
 patroni/validator.py         |  3 +--
 3 files changed, 22 insertions(+), 8 deletions(-)

diff --git a/features/patroni_api.feature b/features/patroni_api.feature
index 0c47fe6e..624d3271 100644
--- a/features/patroni_api.feature
+++ b/features/patroni_api.feature
@@ -35,21 +35,21 @@ Scenario: check local configuration reload
 	Then I receive a response code 202
 
 Scenario: check dynamic configuration change via DCS
-	Given I run patronictl.py edit-config -s 'ttl=10' -p 'max_connections=101' --force batman
-	Then I receive a response returncode 0
-	And I receive a response output "+ttl: 10"
+	Given I issue a PATCH request to http://127.0.0.1:8008/config with {"ttl": 20, "postgresql": {"parameters": {"max_connections": "101"}}}
+	Then I receive a response code 200
 	And Response on GET http://127.0.0.1:8008/patroni contains pending_restart after 11 seconds
 	When I issue a GET request to http://127.0.0.1:8008/config
 	Then I receive a response code 200
-	And I receive a response ttl 10
+	And I receive a response ttl 20
 	When I issue a GET request to http://127.0.0.1:8008/patroni
 	Then I receive a response code 200
 	And I receive a response tags {'new_tag': 'new_value'}
 	And I sleep for 4 seconds
 
 Scenario: check the scheduled restart
-	Given I issue a PATCH request to http://127.0.0.1:8008/config with {"postgresql": {"parameters": {"superuser_reserved_connections": "6"}}}
-	Then I receive a response code 200
+	Given I run patronictl.py edit-config -p 'superuser_reserved_connections=6' --force batman
+	Then I receive a response returncode 0
+	And I receive a response output "+    superuser_reserved_connections: 6"
 	And Response on GET http://127.0.0.1:8008/patroni contains pending_restart after 5 seconds
 	Given I issue a scheduled restart at http://127.0.0.1:8008 in 5 seconds with {"role": "replica"}
 	Then I receive a response code 202
diff --git a/patroni/utils.py b/patroni/utils.py
index eb02c561..ee9a3d21 100644
--- a/patroni/utils.py
+++ b/patroni/utils.py
@@ -326,6 +326,21 @@ def parse_int(value: Any, base_unit: Optional[str] = None) -> Optional[int]:
         >>> parse_int('1TB', 'GB') is None
         True
 
+        >>> parse_int(50, None) == 50
+        True
+
+        >>> parse_int("51", None) == 51
+        True
+
+        >>> parse_int("nonsense", None) == None
+        True
+
+        >>> parse_int("nonsense", "kB") == None
+        True
+
+        >>> parse_int("nonsense") == None
+        True
+
         >>> parse_int(0) == 0
         True
 
diff --git a/patroni/validator.py b/patroni/validator.py
index f780f9e0..fc6cdde5 100644
--- a/patroni/validator.py
+++ b/patroni/validator.py
@@ -792,8 +792,7 @@ class IntValidator(object):
         :param value: value to be checked against the rules defined for this :class:`IntValidator` instance.
         :returns: ``True`` if *value* is valid and within the expected range.
         """
-        if self.base_unit:
-            value = parse_int(value, self.base_unit) or ""
+        value = parse_int(value, self.base_unit) or ""
         ret = isinstance(value, int)\
             and (self.min is None or value >= self.min)\
             and (self.max is None or value <= self.max)

From 412c51ddf1cca682a49cf1c4d11a46fbb446d644 Mon Sep 17 00:00:00 2001
From: Mark Pekala <mark.pekala@nutanix.com>
Date: Mon, 10 Jul 2023 22:43:57 -0700
Subject: [PATCH 04/11] Prevent splitbrain from duplicate names in
 configuration (#2724)

When starting check if node with the same is registered in DCS and try to query it's REST API.
If REST API is accessible exit with the error.

Close #1804
---
 features/basic_replication.feature  |  4 ++++
 features/environment.py             | 12 +++++++---
 features/steps/basic_replication.py | 23 +++++++++++++++++++
 patroni/__main__.py                 | 22 ++++++++++++++++++-
 tests/test_patroni.py               | 34 +++++++++++++++++++++++++++++
 5 files changed, 91 insertions(+), 4 deletions(-)

diff --git a/features/basic_replication.feature b/features/basic_replication.feature
index 4b8b2686..4a0eda68 100644
--- a/features/basic_replication.feature
+++ b/features/basic_replication.feature
@@ -83,3 +83,7 @@ Feature: basic replication
     Then postgres0 role is the secondary after 20 seconds
     When I add the table buz to postgres1
     Then table buz is present on postgres0 after 20 seconds
+
+  Scenario: check graceful rejection when two nodes have the same name
+    Given I start duplicate postgres0 on port 8011
+    Then there is a "Can't start; there is already a node named 'postgres0' running" CRITICAL in the dup-postgres0 patroni log
diff --git a/features/environment.py b/features/environment.py
index a4d22015..f7690ad2 100644
--- a/features/environment.py
+++ b/features/environment.py
@@ -52,10 +52,9 @@ class AbstractController(abc.ABC):
         self._log = open(os.path.join(self._output_dir, self._name + '.log'), 'a')
         self._handle = self._start()
 
-        assert self._has_started(), "Process {0} is not running after being started".format(self._name)
-
         max_wait_limit *= self._context.timeout_multiplier
         for _ in range(max_wait_limit):
+            assert self._has_started(), "Process {0} is not running after being started".format(self._name)
             if self._is_accessible():
                 break
             time.sleep(1)
@@ -344,6 +343,13 @@ class PatroniController(AbstractController):
                         '--datadir=' + os.path.join(self._work_directory, dest),
                         '--dbname=' + self.backup_source])
 
+    def read_patroni_log(self, level):
+        try:
+            with open(str(os.path.join(self._output_dir or '', self._name + ".log"))) as f:
+                return [line for line in f.readlines() if line[24:24 + len(level)] == level]
+        except IOError:
+            return []
+
 
 class ProcessHang(object):
 
@@ -827,7 +833,7 @@ class PatroniPoolController(object):
 
     def __getattr__(self, func):
         if func not in ['stop', 'query', 'write_label', 'read_label', 'check_role_has_changed_to',
-                        'add_tag_to_config', 'get_watchdog', 'patroni_hang', 'backup']:
+                        'add_tag_to_config', 'get_watchdog', 'patroni_hang', 'backup', 'read_patroni_log']:
             raise AttributeError("PatroniPoolController instance has no attribute '{0}'".format(func))
 
         def wrapper(name, *args, **kwargs):
diff --git a/features/steps/basic_replication.py b/features/steps/basic_replication.py
index 9b51a859..6499a03f 100644
--- a/features/steps/basic_replication.py
+++ b/features/steps/basic_replication.py
@@ -9,6 +9,22 @@ def start_patroni(context, name):
     return context.pctl.start(name)
 
 
+@step('I start duplicate {name:w} on port {port:d}')
+def start_duplicate_patroni(context, name, port):
+    config = {
+        "name": name,
+        "restapi": {
+            "listen": "127.0.0.1:{0}".format(port)
+        }
+    }
+    try:
+        context.pctl.start('dup-' + name, custom_config=config)
+        assert False, "Process was expected to fail"
+    except AssertionError as e:
+        assert 'is not running after being started' in str(e),\
+            "No error was raised by duplicate start of {0} ".format(name)
+
+
 @step('I shut down {name:w}')
 def stop_patroni(context, name):
     return context.pctl.stop(name, timeout=60)
@@ -90,3 +106,10 @@ def replication_works(context, primary, replica, time_limit):
         When I add the table test_{0} to {1}
         Then table test_{0} is present on {2} after {3} seconds
     """.format(int(time()), primary, replica, time_limit))
+
+
+@then('there is a "{message}" {level:w} in the {node} patroni log')
+def check_patroni_log(context, message, level, node):
+    messsages_of_level = context.pctl.read_patroni_log(node, level)
+    assert any(message in line for line in messsages_of_level),\
+        "There was no {0} {1} in the {2} patroni log".format(message, level, node)
diff --git a/patroni/__main__.py b/patroni/__main__.py
index 2a669d3f..b365bfe2 100644
--- a/patroni/__main__.py
+++ b/patroni/__main__.py
@@ -30,12 +30,15 @@ class Patroni(AbstractPatroniDaemon):
 
         self.version = __version__
         self.dcs = get_dcs(self.config)
+        self.request = PatroniRequest(self.config, True)
+
+        self.ensure_unique_name()
+
         self.watchdog = Watchdog(self.config)
         self.load_dynamic_configuration()
 
         self.postgresql = Postgresql(self.config['postgresql'])
         self.api = RestApiServer(self, self.config['restapi'])
-        self.request = PatroniRequest(self.config, True)
         self.ha = Ha(self)
 
         self.tags = self.get_tags()
@@ -60,6 +63,23 @@ class Patroni(AbstractPatroniDaemon):
                 logger.warning('Can not get cluster from dcs')
                 time.sleep(5)
 
+    def ensure_unique_name(self) -> None:
+        """A helper method to prevent splitbrain from operator naming error."""
+        from patroni.dcs import Member
+
+        cluster = self.dcs.get_cluster()
+        if not cluster:
+            return
+        member = cluster.get_member(self.config['name'], False)
+        if not isinstance(member, Member):
+            return
+        try:
+            _ = self.request(member, endpoint="/liveness")
+            logger.fatal("Can't start; there is already a node named '%s' running", self.config['name'])
+            sys.exit(1)
+        except Exception:
+            return
+
     def get_tags(self) -> Dict[str, Any]:
         return {tag: value for tag, value in self.config.get('tags', {}).items()
                 if tag not in ('clonefrom', 'nofailover', 'noloadbalance', 'nosync') or value}
diff --git a/tests/test_patroni.py b/tests/test_patroni.py
index d5c76c8b..e5785231 100644
--- a/tests/test_patroni.py
+++ b/tests/test_patroni.py
@@ -10,6 +10,7 @@ from http.server import HTTPServer
 from mock import Mock, PropertyMock, patch
 from patroni.api import RestApiServer
 from patroni.async_executor import AsyncExecutor
+from patroni.dcs import Cluster, Member
 from patroni.dcs.etcd import AbstractEtcdClientWithFailover
 from patroni.exceptions import DCSError
 from patroni.postgresql import Postgresql
@@ -202,3 +203,36 @@ class TestPatroni(unittest.TestCase):
             self.assertRaises(SystemExit, check_psycopg)
         with patch('builtins.__import__', mock_import):
             self.assertRaises(SystemExit, check_psycopg)
+
+    def test_ensure_unique_name(self):
+        # None/empty cluster implies unique name
+        with patch('patroni.dcs.AbstractDCS.get_cluster', Mock(return_value=None)):
+            self.assertIsNone(self.p.ensure_unique_name())
+        empty_cluster = Cluster.empty()
+        with patch('patroni.dcs.AbstractDCS.get_cluster', Mock(return_value=empty_cluster)):
+            self.assertIsNone(self.p.ensure_unique_name())
+        without_members = empty_cluster._asdict()
+        del without_members['members']
+
+        # Cluster with members with different names implies unique name
+        okay_cluster = Cluster(
+            members=[Member(version=1, name="distinct", session=1, data={})],
+            **without_members
+        )
+        with patch('patroni.dcs.AbstractDCS.get_cluster', Mock(return_value=okay_cluster)):
+            self.assertIsNone(self.p.ensure_unique_name())
+
+        # Cluster with a member with the same name that is running
+        bad_cluster = Cluster(
+            members=[Member(version=1, name="postgresql0", session=1, data={
+                "api_url": "https://127.0.0.1:8008",
+            })],
+            **without_members
+        )
+        with patch('patroni.dcs.AbstractDCS.get_cluster', Mock(return_value=bad_cluster)):
+            # If the api of the running node cannot be reached, this implies unique name
+            with patch.object(self.p, 'request', Mock(side_effect=ConnectionError)):
+                self.assertIsNone(self.p.ensure_unique_name())
+            # Only if the api of the running node is reachable do we throw an error
+            with patch.object(self.p, 'request', Mock()):
+                self.assertRaises(SystemExit, self.p.ensure_unique_name)

From b8cff3515a2733b6ad6334ef78ae54eac4b14a8a Mon Sep 17 00:00:00 2001
From: Alexander Kukushkin <cyberdemn@gmail.com>
Date: Tue, 11 Jul 2023 15:04:10 +0200
Subject: [PATCH 05/11] Reduce flakiness of citus behave tests, take 2 (#2742)

Reorder some checks and verify that the old primary is already in the `running` state before checking replication. This check elliminates the race condition when replication started to work but node name is removed from the `synchronous_standby_names` because state isn't `running`.
---
 features/citus.feature | 12 +++++++-----
 1 file changed, 7 insertions(+), 5 deletions(-)

diff --git a/features/citus.feature b/features/citus.feature
index 0ee08fdf..35ccebbe 100644
--- a/features/citus.feature
+++ b/features/citus.feature
@@ -16,14 +16,15 @@ Feature: citus
   Scenario: coordinator failover updates pg_dist_node
     Given I run patronictl.py failover batman --group 0 --candidate postgres1 --force
     Then postgres1 role is the primary after 10 seconds
+    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 "sync" key in a group 0 in DCS has sync_standby=postgres0 after 15 seconds
     And postgres1 is registered in the postgres2 as the primary in group 0 after 5 seconds
-    When I run patronictl.py failover batman --group 0 --candidate postgres0 --force
+    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 "sync" key in a group 0 in DCS has sync_standby=postgres1 after 15 seconds
     And postgres0 is registered in the postgres2 as the primary in group 0 after 5 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
     Given I create a distributed table on postgres0
@@ -31,16 +32,17 @@ Feature: citus
     When I run patronictl.py switchover batman --group 1 --force
     Then I receive a response returncode 0
     And postgres3 role is the primary after 10 seconds
+    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 "sync" key in a group 1 in DCS has sync_standby=postgres2 after 15 seconds
     And postgres3 is registered in the postgres0 as the primary in group 1 after 5 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
     Then I receive a response returncode 0
     And postgres2 role is the primary after 10 seconds
     And replication works from postgres2 to postgres3 after 15 seconds
-    And "sync" key in a group 1 in DCS has sync_standby=postgres3 after 15 seconds
     And postgres2 is registered in the postgres0 as the primary in group 1 after 5 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
     Then a distributed table on postgres0 has expected rows

From 6e96db173fb6ed06101b9344b4ead6f08a0985c6 Mon Sep 17 00:00:00 2001
From: Alexander Kukushkin <cyberdemn@gmail.com>
Date: Wed, 12 Jul 2023 09:42:34 +0200
Subject: [PATCH 06/11] Start postgres not in recovery in some cases (#2726)

If we know for sure that a few moments ago postgres was still running as a primary and we still have the leader lock and can successfully update it, in this case we can safely start postgres back not in recovery. That will allow to avoid bumping timeline without a reason and hopefully improve reliability because it will address issues similar to #2720.

In addition to that remove `if self.state_handler.is_starting()` check from the `recover()` method. This branch could never be reached because the `starting` state is handled earlier in the `_run_cycle()`. Besides that remove redundant `self._crash_recovery_executed`.

P.S. now we do not cover cases when Patroni was killed along with Postgres.
Lets consider that we just started Patroni, there is no leader, and `pg_controldata` reports `Database cluster state` as `shut down`. It feels logical to use `Latest checkpoint location` and `Latest checkpoint's TimeLineID` to do a usual leader race and start directly as a primary, but it could be totally wrong. The thing is that we run `postgres --single` if standby wasn't shut down cleanly before executing `pg_rewind`. As a result `Database cluster state` transition from `in archive recovery` to `shut down`, but if such a node becomes a leader the timeline must be increased.
---
 features/basic_replication.feature |  7 +----
 features/recovery.feature          | 24 ++++++++++++++++
 patroni/ha.py                      | 46 +++++++++++++++++++++++-------
 tests/test_ha.py                   | 13 +++++++--
 4 files changed, 71 insertions(+), 19 deletions(-)
 create mode 100644 features/recovery.feature

diff --git a/features/basic_replication.feature b/features/basic_replication.feature
index 4a0eda68..0e2e8c4b 100644
--- a/features/basic_replication.feature
+++ b/features/basic_replication.feature
@@ -72,16 +72,11 @@ Feature: basic replication
     Then table bar is present on postgres1 after 20 seconds
     And Response on GET http://127.0.0.1:8010/config contains master_start_timeout after 10 seconds
 
-  Scenario: check immediate failover when master_start_timeout=0
-    Given I kill postmaster on postgres2
-    Then postgres1 is a leader after 10 seconds
-    And postgres1 role is the primary after 10 seconds
-
   Scenario: check rejoin of the former primary with pg_rewind
     Given I add the table splitbrain to postgres0
     And I start postgres0
     Then postgres0 role is the secondary after 20 seconds
-    When I add the table buz to postgres1
+    When I add the table buz to postgres2
     Then table buz is present on postgres0 after 20 seconds
 
   Scenario: check graceful rejection when two nodes have the same name
diff --git a/features/recovery.feature b/features/recovery.feature
new file mode 100644
index 00000000..809f7fb9
--- /dev/null
+++ b/features/recovery.feature
@@ -0,0 +1,24 @@
+Feature: recovery
+  We want to check that crashed postgres is started back
+
+  Scenario: check that timeline is not incremented when primary is started after crash
+    Given I start postgres0
+    Then postgres0 is a leader after 10 seconds
+    And there is a non empty initialize key in DCS after 15 seconds
+    When I start postgres1
+    And I add the table foo to postgres0
+    Then table foo is present on postgres1 after 20 seconds
+    When I kill postmaster on postgres0
+    Then postgres0 role is the primary after 10 seconds
+    When I issue a GET request to http://127.0.0.1:8008/
+    Then I receive a response code 200
+    And I receive a response role master
+    And I receive a response timeline 1
+
+  Scenario: check immediate failover when master_start_timeout=0
+    Given I issue a PATCH request to http://127.0.0.1:8008/config with {"master_start_timeout": 0}
+    Then I receive a response code 200
+    And Response on GET http://127.0.0.1:8008/config contains master_start_timeout after 10 seconds
+    When I kill postmaster on postgres0
+    Then postgres1 is a leader after 10 seconds
+    And postgres1 role is the primary after 10 seconds
diff --git a/patroni/ha.py b/patroni/ha.py
index 3108f787..a85dd690 100644
--- a/patroni/ha.py
+++ b/patroni/ha.py
@@ -149,7 +149,6 @@ class Ha(object):
         self._leader_timeline = None
         self.recovering = False
         self._async_response = CriticalTask()
-        self._crash_recovery_executed = False
         self._crash_recovery_started = 0
         self._start_timeout = None
         self._async_executor = AsyncExecutor(self.state_handler.cancellable, self.wakeup)
@@ -411,8 +410,7 @@ class Ha(object):
         return result
 
     def _handle_crash_recovery(self) -> Optional[str]:
-        if not self._crash_recovery_executed and (self.cluster.is_unlocked() or self._rewind.can_rewind):
-            self._crash_recovery_executed = True
+        if self._crash_recovery_started == 0 and (self.cluster.is_unlocked() or self._rewind.can_rewind):
             self._crash_recovery_started = time.time()
             msg = 'doing crash recovery in a single user mode'
             return self._async_executor.try_run_async(msg, self._rewind.ensure_clean_shutdown) or msg
@@ -438,15 +436,29 @@ class Ha(object):
             return self._async_executor.try_run_async(msg, self._do_reinitialize, args=(self.cluster,)) or msg
 
     def recover(self) -> str:
-        # Postgres is not running and we will restart in standby mode. Watchdog is not needed until we promote.
-        self.watchdog.disable()
+        """Handle the case when postgres isn't running.
 
+        Depending on the state of Patroni, DCS cluster view, and pg_controldata the following could happen:
+        - if ``primary_start_timeout`` is 0 and this node owns the leader lock, the lock
+          will be voluntarily released if there are healthy replicas to take it over.
+        - if postgres was running as a ``primary`` and this node owns the leader lock, postgres is started as primary.
+        - crash recover in a single-user mode is executed in the following cases:
+          - postgres was running as ``primary`` wasn't ``shut down`` cleanly and there is no leader in DCS
+          - postgres was running as ``replica`` wasn't ``shut down in recovery`` (cleanly)
+            and we need to run ``pg_rewind`` to join back to the cluster.
+        - ``pg_rewind`` is executed if it is necessary, or optinally, the data directory could
+           be removed if it is allowed by configuration.
+        - after ``crash recovery`` and/or ``pg_rewind`` are executed, postgres is started in recovery.
+
+        :returns: action message, describing what was performed.
+        """
         if self.has_lock() and self.update_lock():
             timeout = self.global_config.primary_start_timeout
             if timeout == 0:
                 # We are requested to prefer failing over to restarting primary. But see first if there
                 # is anyone to fail over to.
                 if self.is_failover_possible(self.cluster.members):
+                    self.watchdog.disable()
                     logger.info("Primary crashed. Failing over.")
                     self.demote('immediate')
                     return 'stopped PostgreSQL to fail over after a crash'
@@ -455,6 +467,23 @@ class Ha(object):
 
         data = self.state_handler.controldata()
         logger.info('pg_controldata:\n%s\n', '\n'.join('  {0}: {1}'.format(k, v) for k, v in data.items()))
+
+        # timeout > 0 indicates that we still have the leader lock, and it was just updated
+        if timeout\
+                and data.get('Database cluster state') in ('in production', 'shutting down', 'shut down')\
+                and self.state_handler.state == 'crashed'\
+                and self.state_handler.role in ('primary', 'master')\
+                and not self.state_handler.config.recovery_conf_exists():
+            # We know 100% that we were running as a primary a few moments ago, therefore could just start postgres
+            msg = 'starting primary after failure'
+            if self._async_executor.try_run_async(msg, self.state_handler.start,
+                                                  args=(timeout, self._async_executor.critical_task)) is None:
+                self.recovering = True
+                return msg
+
+        # Postgres is not running, and we will restart in standby mode. Watchdog is not needed until we promote.
+        self.watchdog.disable()
+
         if data.get('Database cluster state') in ('in production', 'shutting down', 'in crash recovery'):
             msg = self._handle_crash_recovery()
             if msg:
@@ -965,9 +994,6 @@ class Ha(object):
             if ret is not None:  # continue if we just deleted the stale failover key as a leader
                 return ret
 
-        if self.state_handler.is_starting():  # postgresql still starting up is unhealthy
-            return False
-
         if self.state_handler.is_leader():
             # in pause leader is the healthiest only when no initialize or sysid matches with initialize!
             return not self.is_paused() or not self.cluster.initialize\
@@ -1451,7 +1477,7 @@ class Ha(object):
 
             if self.state_handler.role in ('master', 'primary'):
                 logger.info('Demoting primary during %s', self._async_executor.scheduled_action)
-                if self._async_executor.scheduled_action == 'restart':
+                if self._async_executor.scheduled_action in ('restart', 'starting primary after failure'):
                     # Restart needs a special interlocking cancel because postmaster may be just started in a
                     # background thread and has not even written a pid file yet.
                     with self._async_executor.critical_task as task:
@@ -1616,7 +1642,7 @@ class Ha(object):
                         return msg
 
                 # Reset some states after postgres successfully started up
-                self._crash_recovery_executed = False
+                self._crash_recovery_started = 0
                 if self._rewind.executed and not self._rewind.failed:
                     self._rewind.reset_state()
 
diff --git a/tests/test_ha.py b/tests/test_ha.py
index 8495a1f0..cfcb51c4 100644
--- a/tests/test_ha.py
+++ b/tests/test_ha.py
@@ -282,11 +282,20 @@ class TestHa(PostgresInit):
         self.p.follow = false
         self.p.is_running = false
         self.p.name = 'leader'
-        self.p.set_role('primary')
+        self.p.set_role('demoted')
         self.p.controldata = lambda: {'Database cluster state': 'shut down', 'Database system identifier': SYSID}
         self.ha.cluster = get_cluster_initialized_with_leader()
         self.assertEqual(self.ha.run_cycle(), 'starting as readonly because i had the session lock')
 
+    def test_start_primary_after_failure(self):
+        self.p.start = false
+        self.p.is_running = false
+        self.p.name = 'leader'
+        self.p.set_role('primary')
+        self.p.controldata = lambda: {'Database cluster state': 'in production', 'Database system identifier': SYSID}
+        self.ha.cluster = get_cluster_initialized_with_leader()
+        self.assertEqual(self.ha.run_cycle(), 'starting primary after failure')
+
     @patch.object(Rewind, 'ensure_clean_shutdown', Mock())
     def test_crash_recovery(self):
         self.ha.has_lock = true
@@ -837,8 +846,6 @@ class TestHa(PostgresInit):
         self.ha.dcs._last_failsafe = None
         with patch.object(Watchdog, 'is_healthy', PropertyMock(return_value=False)):
             self.assertFalse(self.ha.is_healthiest_node())
-        with patch('patroni.postgresql.Postgresql.is_starting', return_value=True):
-            self.assertFalse(self.ha.is_healthiest_node())
         self.ha.is_paused = true
         self.assertFalse(self.ha.is_healthiest_node())
 

From e4fe239a9dcef39325d52510bf90935d3ec32dd5 Mon Sep 17 00:00:00 2001
From: Alexander Kukushkin <cyberdemn@gmail.com>
Date: Wed, 12 Jul 2023 09:43:40 +0200
Subject: [PATCH 07/11] A few fixes in synchronous_mode (#2741)

- make sure that physical replication slots are created even before the promote happened (when async executor is busy with promote).
- execute `txid_current()` with `synchronous_commit=off` so it doesn't accidentally wait for absent synchronous standbys when `synchronous_mode_strict` is enable and `synchronous_standby_names=*`. These standbys can't connect because replication slots weren't there.
- `synchronous_standby_names` wasn't set to `*` after bootstrap with `synchronous_mode` and `synchronous_mode_strict`.
- add `-c statement_timeout=0` to `PGOPTIONS` when executing `post_bootstrap` script.

Close https://github.com/zalando/patroni/issues/2738
---
 patroni/ha.py                   | 20 ++++++++++++++------
 patroni/postgresql/bootstrap.py |  2 +-
 patroni/postgresql/sync.py      |  9 +++++++--
 tests/test_ha.py                |  1 +
 4 files changed, 23 insertions(+), 9 deletions(-)

diff --git a/patroni/ha.py b/patroni/ha.py
index a85dd690..c55b327b 100644
--- a/patroni/ha.py
+++ b/patroni/ha.py
@@ -1541,6 +1541,9 @@ class Ha(object):
         self.dcs.set_config_value(json.dumps(self.patroni.config.dynamic_configuration, separators=(',', ':')))
         self.dcs.take_leader()
         self.set_is_leader(True)
+        if self.is_synchronous_mode():
+            self.state_handler.sync_handler.set_synchronous_standby_names(
+                CaseInsensitiveSet('*') if self.global_config.is_synchronous_mode_strict else CaseInsensitiveSet())
         self.state_handler.call_nowait(CallbackAction.ON_START)
         self.load_cluster_from_dcs()
 
@@ -1734,16 +1737,21 @@ class Ha(object):
                 msg = self.process_healthy_cluster()
                 ret = self.evaluate_scheduled_restart() or msg
 
-            # we might not have a valid PostgreSQL connection here if another thread
-            # stops PostgreSQL, therefore, we only reload replication slots if no
-            # asynchronous processes are running (should be always the case for the primary)
-            if not self._async_executor.busy and not self.state_handler.is_starting():
+            # We might not have a valid PostgreSQL connection here if AsyncExecutor is doing
+            # something with PostgreSQL. Therefore we will sync replication slots only if no
+            # asynchronous processes are running or we know that this is a standby being promoted.
+            # But, we don't want to run pg_rewind checks or copy logical slots from itself,
+            # therefore we have a couple additional `not is_promoting` checks.
+            is_promoting = self._async_executor.scheduled_action == 'promote'
+            if (not self._async_executor.busy or is_promoting) and not self.state_handler.is_starting():
                 create_slots = self._sync_replication_slots(False)
+
                 if not self.state_handler.cb_called:
-                    if not self.state_handler.is_leader():
+                    if not is_promoting and not self.state_handler.is_leader():
                         self._rewind.trigger_check_diverged_lsn()
                     self.state_handler.call_nowait(CallbackAction.ON_START)
-                if create_slots and self.cluster.leader:
+
+                if not is_promoting and create_slots and self.cluster.leader:
                     err = self._async_executor.try_run_async('copy_logical_slots',
                                                              self.state_handler.slots_handler.copy_logical_slots,
                                                              args=(self.cluster, create_slots))
diff --git a/patroni/postgresql/bootstrap.py b/patroni/postgresql/bootstrap.py
index a76f9f5f..6e25012b 100644
--- a/patroni/postgresql/bootstrap.py
+++ b/patroni/postgresql/bootstrap.py
@@ -185,7 +185,7 @@ class Bootstrap(object):
                 r['host'] = 'localhost'  # set it to localhost to write into pgpass
 
             env = self._postgresql.config.write_pgpass(r)
-            env['PGOPTIONS'] = '-c synchronous_commit=local'
+            env['PGOPTIONS'] = '-c synchronous_commit=local -c statement_timeout=0'
 
             try:
                 ret = self._postgresql.cancellable.call(shlex.split(cmd) + [connstring], env=env)
diff --git a/patroni/postgresql/sync.py b/patroni/postgresql/sync.py
index c56bdbcd..d0f28586 100644
--- a/patroni/postgresql/sync.py
+++ b/patroni/postgresql/sync.py
@@ -193,7 +193,12 @@ class SyncHandler(object):
 
         # Newly connected replicas will be counted as sync only when reached self._primary_flush_lsn
         self._primary_flush_lsn = self._postgresql.last_operation()
-        self._postgresql.query('SELECT pg_catalog.txid_current()')  # Ensure some WAL traffic to move replication
+        # Ensure some WAL traffic to move replication
+        self._postgresql.query("""DO $$
+BEGIN
+    SET local synchronous_commit = 'off';
+    PERFORM * FROM pg_catalog.txid_current();
+END;$$""")
         self._postgresql.reset_cluster_info_state(None)  # Reset internal cache to query fresh values
 
     def current_state(self, cluster: Cluster) -> Tuple[CaseInsensitiveSet, CaseInsensitiveSet]:
@@ -289,6 +294,6 @@ class SyncHandler(object):
         # Reset internal cache to query fresh values
         self._postgresql.reset_cluster_info_state(None)
 
-        # timeline == 0 -- indicates that this is the replica, shoudn't ever happen
+        # timeline == 0 -- indicates that this is the replica
         if self._postgresql.get_primary_timeline() > 0:
             self._handle_synchronous_standby_names_change()
diff --git a/tests/test_ha.py b/tests/test_ha.py
index cfcb51c4..a621b03b 100644
--- a/tests/test_ha.py
+++ b/tests/test_ha.py
@@ -583,6 +583,7 @@ class TestHa(PostgresInit):
         self.p.is_leader = false
         self.assertEqual(self.ha.run_cycle(), 'waiting for end of recovery after bootstrap')
         self.p.is_leader = true
+        self.ha.is_synchronous_mode = true
         self.assertEqual(self.ha.run_cycle(), 'running post_bootstrap')
         self.assertEqual(self.ha.run_cycle(), 'initialized a new cluster')
 

From 47854d77e840e68d0b84aa653a3ff4eea6433d01 Mon Sep 17 00:00:00 2001
From: Matt Baker <93600443+matthbakeredb@users.noreply.github.com>
Date: Wed, 12 Jul 2023 08:55:33 +0100
Subject: [PATCH 08/11] Refactor allowed_keys (#2745)

Refactor allowed_keys method as a class variable

Method does not perform any computation or modify data as it is a static
tuple, therefore it is better expressed as a class variable.
---
 patroni/dcs/__init__.py | 25 ++++++++++++++-----------
 patroni/ha.py           |  2 +-
 2 files changed, 15 insertions(+), 12 deletions(-)

diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py
index a61c6b59..2f270dc0 100644
--- a/patroni/dcs/__init__.py
+++ b/patroni/dcs/__init__.py
@@ -239,23 +239,26 @@ class Member(NamedTuple):
 
 
 class RemoteMember(Member):
-    """Represents a remote member (typically a primary) for a standby cluster"""
+    """Represents a remote member (typically a primary) for a standby cluster.
+
+    :cvar ALLOWED_KEYS: Controls access to relevant key names that could be in stored :attr:`~RemoteMember.data`.
+    """
+
+    ALLOWED_KEYS: Tuple[str, ...] = (
+        'primary_slot_name',
+        'create_replica_methods',
+        'restore_command',
+        'archive_cleanup_command',
+        'recovery_min_apply_delay',
+        'no_replication_slot'
+    )
 
     @classmethod
     def from_name_and_data(cls, name: str, data: Dict[str, Any]) -> 'RemoteMember':
         return super(RemoteMember, cls).__new__(cls, -1, name, None, data)
 
-    @staticmethod
-    def allowed_keys() -> Tuple[str, ...]:
-        return ('primary_slot_name',
-                'create_replica_methods',
-                'restore_command',
-                'archive_cleanup_command',
-                'recovery_min_apply_delay',
-                'no_replication_slot')
-
     def __getattr__(self, name: str) -> Any:
-        if name in RemoteMember.allowed_keys():
+        if name in RemoteMember.ALLOWED_KEYS:
             return self.data.get(name)
 
 
diff --git a/patroni/ha.py b/patroni/ha.py
index c55b327b..5f5de4fa 100644
--- a/patroni/ha.py
+++ b/patroni/ha.py
@@ -1896,7 +1896,7 @@ class Ha(object):
         cluster_params = self.global_config.get_standby_cluster_config()
 
         if cluster_params:
-            data.update({k: v for k, v in cluster_params.items() if k in RemoteMember.allowed_keys()})
+            data.update({k: v for k, v in cluster_params.items() if k in RemoteMember.ALLOWED_KEYS})
             data['no_replication_slot'] = 'primary_slot_name' not in cluster_params
             conn_kwargs = member.conn_kwargs() if member else \
                 {k: cluster_params[k] for k in ('host', 'port') if k in cluster_params}

From 665f49b3209f4c3f1e0fd95219cb6dce999db9d3 Mon Sep 17 00:00:00 2001
From: Matt Baker <93600443+matthbakeredb@users.noreply.github.com>
Date: Wed, 12 Jul 2023 09:15:19 +0100
Subject: [PATCH 09/11] Refactor _copy_items (#2748)

Just a reformat to aid readability.
---
 patroni/postgresql/slots.py | 20 ++++++++++++++++++--
 1 file changed, 18 insertions(+), 2 deletions(-)

diff --git a/patroni/postgresql/slots.py b/patroni/postgresql/slots.py
index b0a53561..d08a09c3 100644
--- a/patroni/postgresql/slots.py
+++ b/patroni/postgresql/slots.py
@@ -21,8 +21,24 @@ logger = logging.getLogger(__name__)
 
 
 def compare_slots(s1: Dict[str, Any], s2: Dict[str, Any], dbid: str = 'database') -> bool:
-    return s1['type'] == s2['type'] and (s1['type'] == 'physical'
-                                         or s1.get(dbid) == s2.get(dbid) and s1['plugin'] == s2['plugin'])
+    """Compare 2 replication slot objects for equality.
+
+    ..note ::
+        If the first argument is a ``physical`` replication slot then only the `type` of the second slot is compared.
+        If the first argument is another ``type`` (e.g. ``logical``) then *dbid* and ``plugin`` are compared.
+
+    :param s1: First slot dictionary to be compared.
+    :param s2: Second slot dictionary to be compared.
+    :param dbid: Optional attribute to be compared when comparing ``logical`` replication slots.
+
+    :return: ``True`` if the slot ``type`` of *s1* and *s2* is matches, and the ``type`` of *s1* is ``physical``,
+             OR the ``types`` match AND the *dbid* and ``plugin`` attributes are equal.
+
+    """
+    return (s1['type'] == s2['type']
+            and (s1['type'] == 'physical'
+                 or s1.get(dbid) == s2.get(dbid)
+                 and s1['plugin'] == s2['plugin']))
 
 
 class SlotsAdvanceThread(Thread):

From d46ca88e6bb4b4788ec86689026c2d2e4da68c79 Mon Sep 17 00:00:00 2001
From: Alexander Kukushkin <cyberdemn@gmail.com>
Date: Thu, 13 Jul 2023 09:24:20 +0200
Subject: [PATCH 10/11] Make it visible replication state on standbys (#2733)

To do that we use `pg_stat_get_wal_receiver()` function, which is available since 9.6. For older versions the `patronictl list` output and REST API responses remain as before.

In case if there is no wal receiver process we check if `restore_command` is set and show the state as `in archive recovery`.

Example of `patronictl list` output:
```bash
$ patronictl list
+ Cluster: batman -------------+---------+---------------------+----+-----------+
| Member      | Host           | Role    | State               | TL | Lag in MB |
+-------------+----------------+---------+---------------------+----+-----------+
| postgresql0 | 127.0.0.1:5432 | Leader  | running             | 12 |           |
| postgresql1 | 127.0.0.1:5433 | Replica | in archive recovery | 12 |         0 |
+-------------+----------------+---------+---------------------+----+-----------+

$ patronictl list
+ Cluster: batman -------------+---------+-----------+----+-----------+
| Member      | Host           | Role    | State     | TL | Lag in MB |
+-------------+----------------+---------+-----------+----+-----------+
| postgresql0 | 127.0.0.1:5432 | Leader  | running   | 12 |           |
| postgresql1 | 127.0.0.1:5433 | Replica | streaming | 12 |         0 |
+-------------+----------------+---------+-----------+----+-----------+
```

Example of REST API response:
```bash
$ curl -s localhost:8009 | jq .
{
  "state": "running",
  "postmaster_start_time": "2023-07-06 13:12:00.595118+02:00",
  "role": "replica",
  "server_version": 150003,
  "xlog": {
    "received_location": 335544480,
    "replayed_location": 335544480,
    "replayed_timestamp": null,
    "paused": false
  },
  "timeline": 12,
  "replication_state": "in archive recovery",
  "dcs_last_seen": 1688642069,
  "database_system_identifier": "7252327498286490579",
  "patroni": {
    "version": "3.0.3",
    "scope": "batman"
  }
}

$ curl -s localhost:8009 | jq .
{
  "state": "running",
  "postmaster_start_time": "2023-07-06 13:12:00.595118+02:00",
  "role": "replica",
  "server_version": 150003,
  "xlog": {
    "received_location": 335544816,
    "replayed_location": 335544816,
    "replayed_timestamp": null,
    "paused": false
  },
  "timeline": 12,
  "replication_state": "streaming",
  "dcs_last_seen": 1688642089,
  "database_system_identifier": "7252327498286490579",
  "patroni": {
    "version": "3.0.3",
    "scope": "batman"
  }
}
```
---
 docs/rest_api.rst                | 10 +++++--
 features/standby_cluster.feature | 10 +++++++
 patroni/api.py                   | 25 ++++++++++++++--
 patroni/ctl.py                   |  3 +-
 patroni/ha.py                    |  7 +++--
 patroni/postgresql/__init__.py   | 51 +++++++++++++++++++++++++++-----
 patroni/postgresql/config.py     |  3 ++
 patroni/utils.py                 |  3 +-
 tests/__init__.py                |  4 +--
 tests/test_api.py                |  9 ++++--
 tests/test_ha.py                 |  5 +++-
 tests/test_rewind.py             |  7 +++--
 tests/test_slots.py              |  4 +--
 13 files changed, 115 insertions(+), 26 deletions(-)

diff --git a/docs/rest_api.rst b/docs/rest_api.rst
index 7bde651f..9deb3f5e 100644
--- a/docs/rest_api.rst
+++ b/docs/rest_api.rst
@@ -141,8 +141,8 @@ Retrieve the Patroni metrics in Prometheus format through the ``GET /metrics`` e
 	# TYPE patroni_replica gauge
 	patroni_replica{scope="batman"} 0
 	# HELP patroni_sync_standby Value is 1 if this node is a sync standby replica, 0 otherwise.
-        # TYPE patroni_sync_standby gauge
-        patroni_sync_standby{scope="batman"} 0
+	# TYPE patroni_sync_standby gauge
+	patroni_sync_standby{scope="batman"} 0
 	# HELP patroni_xlog_received_location Current location of the received Postgres transaction log, 0 if this node is not a replica.
 	# TYPE patroni_xlog_received_location counter
 	patroni_xlog_received_location{scope="batman"} 0
@@ -155,6 +155,12 @@ Retrieve the Patroni metrics in Prometheus format through the ``GET /metrics`` e
 	# HELP patroni_xlog_paused Value is 1 if the Postgres xlog is paused, 0 otherwise.
 	# TYPE patroni_xlog_paused gauge
 	patroni_xlog_paused{scope="batman"} 0
+	# HELP patroni_postgres_streaming Value is 1 if Postgres is streaming, 0 otherwise.
+	# TYPE patroni_postgres_streaming gauge
+	patroni_postgres_streaming{scope="batman"} 1
+	# HELP patroni_postgres_in_archive_recovery Value is 1 if Postgres is replicating from archive, 0 otherwise.
+	# TYPE patroni_postgres_in_archive_recovery gauge
+	patroni_postgres_in_archive_recovery{scope="batman"} 0
 	# HELP patroni_postgres_server_version Version of Postgres (if running), 0 otherwise.
 	# TYPE patroni_postgres_server_version gauge
 	patroni_postgres_server_version {scope="batman"} 140004
diff --git a/features/standby_cluster.feature b/features/standby_cluster.feature
index 4e3bd5f0..850c7970 100644
--- a/features/standby_cluster.feature
+++ b/features/standby_cluster.feature
@@ -13,6 +13,10 @@ Feature: standby cluster
     When I start postgres0
     Then "members/postgres0" key in DCS has state=running after 10 seconds
     And replication works from postgres1 to postgres0 after 15 seconds
+    When I issue a GET request to http://127.0.0.1:8008/patroni
+    Then I receive a response code 200
+    And I receive a response replication_state streaming
+    And "members/postgres0" key in DCS has replication_state=streaming after 10 seconds
 
   @slot-advance
   Scenario: check permanent logical slots are synced to the replica
@@ -34,6 +38,9 @@ Feature: standby cluster
     Then postgres1 is a leader of batman1 after 10 seconds
     When I add the table foo to postgres0
     Then table foo is present on postgres1 after 20 seconds
+    When I issue a GET request to http://127.0.0.1:8009/patroni
+    Then I receive a response code 200
+    And I receive a response replication_state streaming
     And I sleep for 3 seconds
     When I issue a GET request to http://127.0.0.1:8009/primary
     Then I receive a response code 503
@@ -44,6 +51,9 @@ Feature: standby cluster
     When I start postgres2 in a cluster batman1
     Then postgres2 role is the replica after 24 seconds
     And table foo is present on postgres2 after 20 seconds
+    When I issue a GET request to http://127.0.0.1:8010/patroni
+    Then I receive a response code 200
+    And I receive a response replication_state streaming
     And postgres1 does not have a logical replication slot named test_logical
 
   Scenario: check failover
diff --git a/patroni/api.py b/patroni/api.py
index 7be2a049..836a8673 100644
--- a/patroni/api.py
+++ b/patroni/api.py
@@ -535,6 +535,18 @@ class RestApiHandler(BaseHTTPRequestHandler):
         metrics.append("patroni_xlog_paused{0} {1}"
                        .format(scope_label, int(postgres.get('xlog', {}).get('paused', False) is True)))
 
+        if postgres.get('server_version', 0) >= 90600:
+            metrics.append("# HELP patroni_postgres_streaming Value is 1 if Postgres is streaming, 0 otherwise.")
+            metrics.append("# TYPE patroni_postgres_streaming gauge")
+            metrics.append("patroni_postgres_streaming{0} {1}"
+                           .format(scope_label, int(postgres.get('replication_state') == 'streaming')))
+
+            metrics.append("# HELP patroni_postgres_in_archive_recovery Value is 1"
+                           " if Postgres is replicating from archive, 0 otherwise.")
+            metrics.append("# TYPE patroni_postgres_in_archive_recovery gauge")
+            metrics.append("patroni_postgres_in_archive_recovery{0} {1}"
+                           .format(scope_label, int(postgres.get('replication_state') == 'in archive recovery')))
+
         metrics.append("# HELP patroni_postgres_server_version Version of Postgres (if running), 0 otherwise.")
         metrics.append("# TYPE patroni_postgres_server_version gauge")
         metrics.append("patroni_postgres_server_version {0} {1}".format(scope_label, postgres.get('server_version', 0)))
@@ -1151,8 +1163,11 @@ class RestApiHandler(BaseHTTPRequestHandler):
 
             if postgresql.state not in ('running', 'restarting', 'starting'):
                 raise RetryFailedError('')
+            replication_state = ('(pg_catalog.pg_stat_get_wal_receiver()).status'
+                                 if postgresql.major_version >= 90600 else 'NULL') + ", " +\
+                ("pg_catalog.current_setting('restore_command')" if postgresql.major_version >= 120000 else "NULL")
             stmt = ("SELECT " + postgresql.POSTMASTER_START_TIME + ", " + postgresql.TL_LSN + ","
-                    " pg_catalog.pg_last_xact_replay_timestamp(),"
+                    " pg_catalog.pg_last_xact_replay_timestamp(), " + replication_state + ","
                     " pg_catalog.array_to_json(pg_catalog.array_agg(pg_catalog.row_to_json(ri))) "
                     "FROM (SELECT (SELECT rolname FROM pg_catalog.pg_authid WHERE oid = usesysid) AS usename,"
                     " application_name, client_addr, w.state, sync_state, sync_priority"
@@ -1188,8 +1203,12 @@ class RestApiHandler(BaseHTTPRequestHandler):
                     if not cluster or cluster.is_unlocked() or not cluster.leader else cluster.leader.timeline
                 result['timeline'] = postgresql.replica_cached_timeline(leader_timeline)
 
-            if row[7]:
-                result['replication'] = row[7]
+            replication_state = postgresql.replication_state_from_parameters(row[1] > 0, row[7], row[8])
+            if replication_state:
+                result['replication_state'] = replication_state
+
+            if row[9]:
+                result['replication'] = row[9]
 
         except (psycopg.Error, RetryFailedError, PostgresConnectionException):
             state = postgresql.state
diff --git a/patroni/ctl.py b/patroni/ctl.py
index 12463aad..55a054c1 100644
--- a/patroni/ctl.py
+++ b/patroni/ctl.py
@@ -1490,7 +1490,8 @@ def output_members(obj: Dict[str, Any], cluster: Cluster, name: str,
         * ``Role``: ``Leader``, ``Standby Leader``, ``Sync Standby`` or ``Replica``;
         * ``State``: ``stopping``, ``stopped``, ``stop failed``, ``crashed``, ``running``, ``starting``,
           ``start failed``, ``restarting``, ``restart failed``, ``initializing new cluster``, ``initdb failed``,
-          ``running custom bootstrap script``, ``custom bootstrap failed``, or ``creating replica``, and so on;
+          ``running custom bootstrap script``, ``custom bootstrap failed``, ``creating replica``, ``streaming``,
+          ``in archive recovery``, and so on;
         * ``TL``: current timeline in Postgres;
           ``Lag in MB``: replication lag.
 
diff --git a/patroni/ha.py b/patroni/ha.py
index 5f5de4fa..1148802a 100644
--- a/patroni/ha.py
+++ b/patroni/ha.py
@@ -306,10 +306,13 @@ class Ha(object):
             if self._async_executor.scheduled_action in (None, 'promote') \
                     and data['state'] in ['running', 'restarting', 'starting']:
                 try:
-                    timeline: Optional[int]
                     timeline, wal_position, pg_control_timeline = self.state_handler.timeline_wal_position()
                     data['xlog_location'] = wal_position
-                    if not timeline:  # try pg_stat_wal_receiver to get the timeline
+                    if not timeline:  # running as a standby
+                        replication_state = self.state_handler.replication_state()
+                        if replication_state:
+                            data['replication_state'] = replication_state
+                        # try pg_stat_wal_receiver to get the timeline
                         timeline = self.state_handler.received_timeline()
                     if not timeline:
                         # So far the only way to get the current timeline on the standby is from
diff --git a/patroni/postgresql/__init__.py b/patroni/postgresql/__init__.py
index b2423c89..632cbee7 100644
--- a/patroni/postgresql/__init__.py
+++ b/patroni/postgresql/__init__.py
@@ -197,18 +197,19 @@ class Postgresql(object):
                         and self.role in ('master', 'primary', 'promoted') else "'on', '', NULL")
 
         if self._major_version >= 90600:
-            extra = ("(SELECT pg_catalog.json_agg(s.*) FROM (SELECT slot_name, slot_type as type, datoid::bigint, "
-                     "plugin, catalog_xmin, pg_catalog.pg_wal_lsn_diff(confirmed_flush_lsn, '0/0')::bigint"
-                     " AS confirmed_flush_lsn FROM pg_catalog.pg_get_replication_slots()) AS s)"
-                     if self._has_permanent_logical_slots and self._major_version >= 110000 else "NULL") + extra
+            extra = ("pg_catalog.current_setting('restore_command')" if self._major_version >= 120000 else "NULL") +\
+                ", " + ("(SELECT pg_catalog.json_agg(s.*) FROM (SELECT slot_name, slot_type as type, datoid::bigint, "
+                        "plugin, catalog_xmin, pg_catalog.pg_wal_lsn_diff(confirmed_flush_lsn, '0/0')::bigint"
+                        " AS confirmed_flush_lsn FROM pg_catalog.pg_get_replication_slots()) AS s)"
+                        if self._has_permanent_logical_slots and self._major_version >= 110000 else "NULL") + extra
             extra = (", CASE WHEN latest_end_lsn IS NULL THEN NULL ELSE received_tli END,"
-                     " slot_name, conninfo, {0} FROM pg_catalog.pg_stat_get_wal_receiver()").format(extra)
+                     " slot_name, conninfo, status, {0} FROM pg_catalog.pg_stat_get_wal_receiver()").format(extra)
             if self.role == 'standby_leader':
                 extra = "timeline_id" + extra + ", pg_catalog.pg_control_checkpoint()"
             else:
                 extra = "0" + extra
         else:
-            extra = "0, NULL, NULL, NULL, NULL" + extra
+            extra = "0, NULL, NULL, NULL, NULL, NULL, NULL" + extra
 
         return ("SELECT " + self.TL_LSN + ", {2}").format(self.wal_name, self.lsn_name, extra)
 
@@ -426,7 +427,8 @@ class Postgresql(object):
                 result = self._is_leader_retry(self._query, self.cluster_info_query).fetchone()
                 cluster_info_state = dict(zip(['timeline', 'wal_position', 'replayed_location',
                                                'received_location', 'replay_paused', 'pg_control_timeline',
-                                               'received_tli', 'slot_name', 'conninfo', 'slots', 'synchronous_commit',
+                                               'received_tli', 'slot_name', 'conninfo', 'receiver_state',
+                                               'restore_command', 'slots', 'synchronous_commit',
                                                'synchronous_standby_names', 'pg_stat_replication'], result))
                 if self._has_permanent_logical_slots:
                     cluster_info_state['slots'] =\
@@ -472,6 +474,41 @@ class Postgresql(object):
         """:returns: a result set of 'SELECT * FROM pg_stat_replication'."""
         return self._cluster_info_state_get('pg_stat_replication') or []
 
+    def replication_state_from_parameters(self, is_leader: bool, receiver_state: Optional[str],
+                                          restore_command: Optional[str]) -> Optional[str]:
+        """Figure out the replication state from input parameters.
+
+        .. note::
+            This method could be only called when Postgres is up, running and queries are successfuly executed.
+
+        :is_leader: `True` is postgres is not running in recovery
+        :receiver_state: value from `pg_stat_get_wal_receiver.state` or None if Postgres is older than 9.6
+        :restore_command: value of ``restore_command`` GUC for PostgreSQL 12+ or
+                          `postgresql.recovery_conf.restore_command` if it is set in Patroni configuration
+
+        :returns: - `None` for the primary and for Postgres older than 9.6;
+                  - 'streaming' if replica is streaming according to the `pg_stat_wal_receiver` view;
+                  - 'in archive recovery' if replica isn't streaming and there is a `restore_command`
+        """
+        if self._major_version >= 90600 and not is_leader:
+            if receiver_state == 'streaming':
+                return 'streaming'
+            # For Postgres older than 12 we get `restore_command` from Patroni config, otherwise we check GUC
+            if self._major_version < 120000 and self.config.restore_command() or restore_command:
+                return 'in archive recovery'
+
+    def replication_state(self) -> Optional[str]:
+        """Checks replication state from `pg_stat_get_wal_receiver()`.
+
+        .. note::
+            Available only since 9.6
+
+        :returns: ``streaming``, ``in archive recovery``, or ``None``
+        """
+        return self.replication_state_from_parameters(self.is_leader(),
+                                                      self._cluster_info_state_get('receiver_state'),
+                                                      self._cluster_info_state_get('restore_command'))
+
     def is_leader(self) -> bool:
         try:
             return bool(self._cluster_info_state_get('timeline'))
diff --git a/patroni/postgresql/config.py b/patroni/postgresql/config.py
index ef3f5c36..9461714f 100644
--- a/patroni/postgresql/config.py
+++ b/patroni/postgresql/config.py
@@ -1177,3 +1177,6 @@ class ConfigHandler(object):
 
     def get(self, key: str, default: Optional[Any] = None) -> Optional[Any]:
         return self._config.get(key, default)
+
+    def restore_command(self) -> Optional[str]:
+        return (self.get('recovery_conf') or {}).get('restore_command')
diff --git a/patroni/utils.py b/patroni/utils.py
index ee9a3d21..b7694d17 100644
--- a/patroni/utils.py
+++ b/patroni/utils.py
@@ -773,7 +773,8 @@ def cluster_as_json(cluster: 'Cluster', global_config: Optional['GlobalConfig']
         else:
             role = 'replica'
 
-        member = {'name': m.name, 'role': role, 'state': m.data.get('state', ''), 'api_url': m.api_url}
+        state = (m.data.get('replication_state', '') if role != 'leader' else '') or m.data.get('state', '')
+        member = {'name': m.name, 'role': role, 'state': state, 'api_url': m.api_url}
         conn_kwargs = m.conn_kwargs()
         if conn_kwargs.get('host'):
             member['host'] = conn_kwargs['host']
diff --git a/tests/__init__.py b/tests/__init__.py
index 03598ae1..dcf9ed9e 100644
--- a/tests/__init__.py
+++ b/tests/__init__.py
@@ -108,7 +108,7 @@ class MockCursor(object):
         elif sql.startswith('WITH slots AS (SELECT slot_name, active'):
             self.results = [(False, True)] if self.rowcount == 1 else [None]
         elif sql.startswith('SELECT CASE WHEN pg_catalog.pg_is_in_recovery()'):
-            self.results = [(1, 2, 1, 0, False, 1, 1, None, None,
+            self.results = [(1, 2, 1, 0, False, 1, 1, None, None, 'streaming', '',
                              [{"slot_name": "ls", "confirmed_flush_lsn": 12345}],
                              'on', 'n1', None)]
         elif sql.startswith('SELECT pg_catalog.pg_is_in_recovery()'):
@@ -117,7 +117,7 @@ class MockCursor(object):
             replication_info = '[{"application_name":"walreceiver","client_addr":"1.2.3.4",' +\
                                '"state":"streaming","sync_state":"async","sync_priority":0}]'
             now = datetime.datetime.now(tzutc)
-            self.results = [(now, 0, '', 0, '', False, now, replication_info)]
+            self.results = [(now, 0, '', 0, '', False, now, 'streaming', None, replication_info)]
         elif sql.startswith('SELECT name, setting'):
             self.results = [('wal_segment_size', '2048', '8kB', 'integer', 'internal'),
                             ('wal_block_size', '8192', None, 'integer', 'internal'),
diff --git a/tests/test_api.py b/tests/test_api.py
index 25342dca..f9d0b1c1 100644
--- a/tests/test_api.py
+++ b/tests/test_api.py
@@ -29,7 +29,8 @@ class MockPostgresql(object):
     name = 'test'
     state = 'running'
     role = 'primary'
-    server_version = '999999'
+    server_version = 90625
+    major_version = 90600
     sysid = 'dummysysid'
     scope = 'dummy'
     pending_restart = True
@@ -55,6 +56,10 @@ class MockPostgresql(object):
     def is_running():
         return True
 
+    @staticmethod
+    def replication_state_from_parameters(*args):
+        return 'streaming'
+
 
 class MockWatchdog(object):
     is_healthy = False
@@ -219,7 +224,7 @@ class TestRestApiHandler(unittest.TestCase):
         with patch.object(MockHa, 'restart_scheduled', Mock(return_value=True)):
             MockRestApiServer(RestApiHandler, 'GET /primary')
         self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'GET /primary'))
-        with patch.object(RestApiServer, 'query', Mock(return_value=[('', 1, '', '', '', '', False, '')])):
+        with patch.object(RestApiServer, 'query', Mock(return_value=[('', 1, '', '', '', '', False, None, None, '')])):
             self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'GET /patroni'))
         with patch.object(GlobalConfig, 'is_standby_cluster', Mock(return_value=True)),\
                 patch.object(GlobalConfig, 'is_paused', Mock(return_value=True)):
diff --git a/tests/test_ha.py b/tests/test_ha.py
index a621b03b..dad9562f 100644
--- a/tests/test_ha.py
+++ b/tests/test_ha.py
@@ -223,9 +223,12 @@ class TestHa(PostgresInit):
 
     @patch.object(Postgresql, 'received_timeline', Mock(return_value=None))
     def test_touch_member(self):
+        self.p._major_version = 110000
+        self.p.is_leader = false
         self.p.timeline_wal_position = Mock(return_value=(0, 1, 0))
         self.p.replica_cached_timeline = Mock(side_effect=Exception)
-        self.ha.touch_member()
+        with patch.object(Postgresql, '_cluster_info_state_get', Mock(return_value='streaming')):
+            self.ha.touch_member()
         self.p.timeline_wal_position = Mock(return_value=(0, 1, 1))
         self.p.set_role('standby_leader')
         self.ha.touch_member()
diff --git a/tests/test_rewind.py b/tests/test_rewind.py
index 3d921621..b5858d8e 100644
--- a/tests/test_rewind.py
+++ b/tests/test_rewind.py
@@ -91,9 +91,10 @@ class TestRewind(BaseTestPostgresql):
                                              'Latest checkpoint location': '0/'})):
             self.r.rewind_or_reinitialize_needed_and_possible(self.leader)
 
-        with patch.object(Postgresql, 'is_running', Mock(return_value=True)):
-            with patch.object(MockCursor, 'fetchone', Mock(side_effect=[(0, 0, 1, 1, 0, 0, 0, 0, 0, None), Exception])):
-                self.r.rewind_or_reinitialize_needed_and_possible(self.leader)
+        with patch.object(Postgresql, 'is_running', Mock(return_value=True)),\
+                patch.object(MockCursor, 'fetchone',
+                             Mock(side_effect=[(0, 0, 1, 1, 0, 0, 0, 0, 0, None, None, None), Exception])):
+            self.r.rewind_or_reinitialize_needed_and_possible(self.leader)
 
     @patch.object(CancellableSubprocess, 'call', mock_cancellable_call)
     @patch.object(Postgresql, 'checkpoint', side_effect=['', '1'],)
diff --git a/tests/test_slots.py b/tests/test_slots.py
index bdb4bc5d..a962dbe3 100644
--- a/tests/test_slots.py
+++ b/tests/test_slots.py
@@ -77,14 +77,14 @@ class TestSlotsHandler(BaseTestPostgresql):
         with patch.object(Postgresql, '_query') as mock_query:
             self.p.reset_cluster_info_state(None)
             mock_query.return_value.fetchone.return_value = (
-                1, 0, 0, 0, 0, 0, 0, 0, 0,
+                1, 0, 0, 0, 0, 0, 0, 0, 0, None, None,
                 [{"slot_name": "ls", "type": "logical", "datoid": 5, "plugin": "b",
                   "confirmed_flush_lsn": 12345, "catalog_xmin": 105}])
             self.assertEqual(self.p.slots(), {'ls': 12345})
 
             self.p.reset_cluster_info_state(None)
             mock_query.return_value.fetchone.return_value = (
-                1, 0, 0, 0, 0, 0, 0, 0, 0,
+                1, 0, 0, 0, 0, 0, 0, 0, 0, None, None,
                 [{"slot_name": "ls", "type": "logical", "datoid": 6, "plugin": "b",
                   "confirmed_flush_lsn": 12345, "catalog_xmin": 105}])
             self.assertEqual(self.p.slots(), {})

From a4d29eb99ea4e943f9d4fdae48ba9d14b46c567c Mon Sep 17 00:00:00 2001
From: Alexander Kukushkin <cyberdemn@gmail.com>
Date: Thu, 13 Jul 2023 11:51:38 +0200
Subject: [PATCH 11/11] Release v3.0.4 (#2754)

- update release notes
- bump version
- bump pyright version
---
 .github/workflows/tests.yaml |  2 +-
 docs/releases.rst            | 44 ++++++++++++++++++++++++++++++++++++
 patroni/version.py           |  2 +-
 3 files changed, 46 insertions(+), 2 deletions(-)

diff --git a/.github/workflows/tests.yaml b/.github/workflows/tests.yaml
index 59a53c6e..355ea927 100644
--- a/.github/workflows/tests.yaml
+++ b/.github/workflows/tests.yaml
@@ -173,4 +173,4 @@ jobs:
 
     - uses: jakebailey/pyright-action@v1
       with:
-        version: 1.1.316
+        version: 1.1.317
diff --git a/docs/releases.rst b/docs/releases.rst
index b9a3eae7..48cad683 100644
--- a/docs/releases.rst
+++ b/docs/releases.rst
@@ -3,6 +3,50 @@
 Release notes
 =============
 
+Version 3.0.4
+-------------
+
+**New features**
+
+- Make the replication status of standby nodes visible (Alexander Kukushkin)
+
+  For PostgreSQL 9.6+ Patroni will report the replication state as ``streaming`` when the standby is streaming from the other node or ``in archive recovery`` when there is no replication connection and ``restore_command`` is set. The state is visible in ``member`` keys in DCS, in the REST API, and in ``patronictl list`` output.
+
+
+**Improvements**
+
+- Improved error messages with Etcd v3 (Alexander Kukushkin)
+
+  When Etcd v3 cluster isn't accessible Patroni was reporting that it can't access ``/v2`` endpoints.
+
+- Use quorum read in ``patronictl`` if it is possible (Alexander Kukushkin)
+
+  Etcd or Consul clusters could be degraded to read-only, but from the ``patronictl`` view everything was fine. Now it will fail with the error.
+
+- Prevent splitbrain from duplicate names in configuration (Mark Pekala)
+
+  When starting Patroni will check if node with the same name is registered in DCS, and try to query its REST API. If REST API is accessible Patroni exits with an error. It will help to protect from the human error.
+
+- Start Postgres not in recovery if it crashed while Patroni is running (Alexander Kukushkin)
+
+  It may reduce recovery time and will help from unnecessary timeline increments.
+
+
+**Bugfixes**
+
+- REST API SSL certificate were not reloaded upon receiving a SIGHUP (Israel Barth Rubio)
+
+  Regression was introduced in 3.0.3.
+
+- Fixed integer GUCs validation for parameters like ``max_connections`` (Feike Steenbergen)
+
+  Patroni didn't like quoted numeric values. Regression was introduced in 3.0.3.
+
+- Fix issue with ``synchronous_mode`` (Alexander Kukushkin)
+
+  Execute ``txid_current()`` with ``synchronous_commit=off`` so it doesn't accidentally wait for absent synchronous standbys when ``synchronous_mode_strict`` is enabled.
+
+
 Version 3.0.3
 -------------
 
diff --git a/patroni/version.py b/patroni/version.py
index 96c68e77..8653a3bc 100644
--- a/patroni/version.py
+++ b/patroni/version.py
@@ -2,4 +2,4 @@
 
 :var __version__: the current Patroni version.
 """
-__version__ = '3.0.3'
+__version__ = '3.0.4'