Repository: zalando/patroni
Pull request #2726 (2023-07-04 16:06:21)

Start postgres not in recovery in some cases
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.

coveralls: ## Pull Request Test Coverage Report for [Build 5456677403](https://coveralls.io/builds/61131006)

* **10** of **10**   **(100.0%)**  changed or added relevant lines in **1** file are covered.
* No unchanged relevant lines lost coverage.
* Overall coverage remained the same at **99.858%**

---



|  Totals | [![Coverage Status](https://coveralls.io/builds/61131006/badge)](https://coveralls.io/builds/61131006) |
| :-- | --: |
| Change from base [Build 5456584464](https://coveralls.io/builds/61130749): |  0.0% |
| Covered Lines: | 12640 |
| Relevant Lines: | 12658 |

---
##### 💛  - [Coveralls](https://coveralls.io)


Patch:
diff --git a/.gitignore b/.gitignore
index 38c06f07..c3af6eb0 100644
--- a/.gitignore
+++ b/.gitignore
@@ -57,3 +57,9 @@ docs/source/_templates/
 
 #VSCode IDE
 .vscode/
+
+# Virtual environment
+venv*/
+
+# Default test data directory
+data/
diff --git a/README.rst b/README.rst
index 18be2b49..84e67379 100644
--- a/README.rst
+++ b/README.rst
@@ -8,7 +8,7 @@ You can find a version of this documentation that is searchable and also easier
 
 There are many ways to run high availability with PostgreSQL; for a list, see the `PostgreSQL Documentation <https://wiki.postgresql.org/wiki/Replication,_Clustering,_and_Connection_Pooling>`__.
 
-Patroni is a template for high availability (HA) PostgreSQL solutions using Python. For maximum accessibility, Patroni supports a variety of distributed configuration stores like `ZooKeeper <https://zookeeper.apache.org/>`__, `etcd <https://github.com/coreos/etcd>`__, `Consul <https://github.com/hashicorp/consul>`__ or `Kubernetes <https://kubernetes.io>`__. Database engineers, DBAs, DevOps engineers, and SREs who are looking to quickly deploy HA PostgreSQL in datacenters — or anywhere else — will hopefully find it useful.
+Patroni is a template for high availability (HA) PostgreSQL solutions using Python. For maximum accessibility, Patroni supports a variety of distributed configuration stores like `ZooKeeper <https://zookeeper.apache.org/>`__, `etcd <https://github.com/coreos/etcd>`__, `Consul <https://github.com/hashicorp/consul>`__ or `Kubernetes <https://kubernetes.io>`__. Database engineers, DBAs, DevOps engineers, and SREs who are looking to quickly deploy HA PostgreSQL in datacenters - or anywhere else - will hopefully find it useful.
 
 We call Patroni a "template" because it is far from being a one-size-fits-all or plug-and-play replication system. It will have its own caveats. Use wisely.
 
diff --git a/features/basic_replication.feature b/features/basic_replication.feature
index 4b8b2686..0e2e8c4b 100644
--- a/features/basic_replication.feature
+++ b/features/basic_replication.feature
@@ -72,14 +72,13 @@ 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
+    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/citus.feature b/features/citus.feature
index 2cf295eb..35ccebbe 100644
--- a/features/citus.feature
+++ b/features/citus.feature
@@ -10,20 +10,21 @@ Feature: citus
     And I start postgres3 in citus group 1
     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 worker in group 0
-    And postgres2 is registered in the postgres0 as the worker in group 1
+    And postgres0 is registered in the postgres0 as the primary in group 0 after 5 seconds
+    And postgres2 is registered in the postgres0 as the primary 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
     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 postgres1 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=postgres0 after 15 seconds
-    And postgres1 is registered in the postgres2 as the worker in group 0
-    When I run patronictl.py failover batman --group 0 --candidate postgres0 --force
+    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 "sync" key in a group 0 in DCS has sync_standby=postgres1 after 15 seconds
-    And postgres0 is registered in the postgres2 as the worker in group 0
 
   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 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 postgres3 is registered in the postgres0 as the worker in group 1
     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 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 postgres2 is registered in the postgres0 as the worker in group 1
     And a thread is still alive
     When I stop a thread
     Then a distributed table on postgres0 has expected rows
@@ -52,7 +54,7 @@ Feature: citus
     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 postgres2 is registered in the postgres0 as the worker in group 1
+    And postgres2 is registered in the postgres0 as the primary in group 1 after 5 seconds
     And a thread is still alive
     When I stop a thread
     Then a distributed table on postgres0 has expected rows
@@ -64,8 +66,7 @@ Feature: citus
     When I run patronictl.py edit-config batman --group 2 -s ttl=20 --force
     Then I receive a response returncode 0
     And I receive a response output "+ttl: 20"
-    When I sleep for 2 seconds
-    Then postgres4 is registered in the postgres2 as the worker in group 2
+    Then postgres4 is registered in the postgres2 as the primary in group 2 after 5 seconds
     When I shut down postgres4
     Then There is a transaction in progress on postgres0 changing pg_dist_node
     When I run patronictl.py restart batman postgres2 --group 1 --force
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/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/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/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/features/steps/citus.py b/features/steps/citus.py
index d645a504..1af70a30 100644
--- a/features/steps/citus.py
+++ b/features/steps/citus.py
@@ -44,12 +44,24 @@ def start_citus(context, name, group):
     return context.pctl.start(name, custom_config={"citus": {"database": "postgres", "group": int(group)}})
 
 
-@step('{name1:w} is registered in the {name2:w} as the worker in group {group:d}')
-def check_registration(context, name1, name2, group):
+@step('{name1:w} is registered in the {name2:w} as the {role:w} in group {group:d} after {time_limit:d} seconds')
+def check_registration(context, name1, name2, role, group, time_limit):
+    time_limit *= context.timeout_multiplier
+    max_time = time.time() + int(time_limit)
+
     worker_port = int(context.pctl.query(name1, "SHOW port").fetchone()[0])
-    r = context.pctl.query(name2, "SELECT nodeport FROM pg_catalog.pg_dist_node WHERE groupid = {0}".format(group))
-    assert worker_port == r.fetchone()[0],\
-        "Worker {0} is not registered in pg_dist_node on the coordinator {1}".format(name1, name2)
+
+    while time.time() < max_time:
+        try:
+            cur = context.pctl.query(name2, "SELECT nodeport, noderole"
+                                            " FROM pg_catalog.pg_dist_node WHERE groupid = {0}".format(group))
+            mapping = {r[0]: r[1] for r in cur}
+            if mapping.get(worker_port) == role:
+                return
+        except Exception:
+            pass
+        time.sleep(1)
+    assert False, "Node {0} is not registered in pg_dist_node on the node {1}".format(name1, name2)
 
 
 @step('I create a distributed table on {name:w}')
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/patroni/ctl.py b/patroni/ctl.py
index 8e231466..12463aad 100644
--- a/patroni/ctl.py
+++ b/patroni/ctl.py
@@ -258,7 +258,7 @@ option_insecure = click.option('-k', '--insecure', is_flag=True, help='Allow con
 role_choice = click.Choice(['leader', 'primary', 'standby-leader', 'replica', 'standby', 'any', 'master'])
 
 
-@click.group()
+@click.group(cls=click.Group)
 @click.option('--config-file', '-c', help='Configuration file',
               envvar='PATRONICTL_CONFIG_FILE', default=CONFIG_FILE_PATH)
 @click.option('--dcs-url', '--dcs', '-d', 'dcs_url', help='The DCS connect url', envvar='DCS_URL')
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 3e7f0681..94e546ac 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()
@@ -722,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 b89e36a5..32d00359 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()
@@ -327,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:
@@ -595,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)
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/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)
diff --git a/setup.py b/setup.py
index bff90cd4..d61eab6e 100644
--- a/setup.py
+++ b/setup.py
@@ -88,7 +88,7 @@ class Flake8(_Command):
                 yield package_directory
 
     def targets(self):
-        return [package for package in self.package_files()] + ['tests', 'setup.py']
+        return [package for package in self.package_files()] + ['tests', 'features', 'setup.py']
 
     def run(self):
         from flake8.main.cli import main
@@ -116,7 +116,7 @@ class PyTest(_Command):
 
 
 def read(fname):
-    with open(os.path.join(__location__, fname)) as fd:
+    with open(os.path.join(__location__, fname), encoding='utf-8') as fd:
         return fd.read()
 
 


Test patch:
diff --git a/.github/workflows/tests.yaml b/.github/workflows/tests.yaml
index 78cce057..59a53c6e 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.315
+        version: 1.1.316
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())
 
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)