Repository: zalando/patroni
Pull request #1820 (2021-01-20 15:39:39)

Failover logical slots
Replication slots and logical replication were introduced in PostgreSQL 9.4 and their main purpose not only retaining WAL files required for logical decoding by tracking `restart_lsn` and `confirmed_flush_lsn`,  but is also tracking `catalog_xmin` in order to prevent autovacuum removing dead catalog tuples that might still be needed for logical decoding.

The biggest problem of logical replication slots is, that vanilla PostgreSQL doesn't allow creating them on standby and therefore in case of failover if you create logical slots on a freshly promoted node there are chances that some transactions could be lost.

Unfortunately, all attempts of implementing the logical decoding on the standby and failover logical slots in the postgres core didn't have much success, but at least now we are in a state when it is possible to do it by external tooling. You can read more about this topic [here](https://wiki.postgresql.org/wiki/Failover_slots) and [here](https://wiki.postgresql.org/wiki/Logical_replication_and_physical_standby_failover).

One of the competitive advantages of Patroni is that it doesn't rely on any third-party extensions. Therefore the main goal of this feature is to implement it in a way that logical slots are safe to use after the failover and there is no transaction loss.
1. We create the logical slot on a standby. It is possible to achieve it by performing `pg_ctl stop`, copying the `pg_replslot/$slot_name/state` file from the primary to standby, and later when you start the postgres up, the slot becomes visible.
2. We "create" logical slots only on members that are eligible to become the leader (i.e. they don't have `tags.nofailover` set)
3. Starting from PostgreSQL 11 there is the [pg_replication_slot_advance()](https://www.postgresql.org/docs/current/functions-admin.html#FUNCTIONS-REPLICATION) function available, that actually works not only with physical slots, but as well with logical. Therefore we can move the logical slots forward on the replica without third-party extensions.
4. The hot_standby_feedback must be enabled not only on standbys with logical slots, but also on cascading nodes. Example: `Primary <-- Node1(tags.nofailover=true) <-- Node2`.
5. Since the primary tracks the `catalog_xmin`of the physical replica on the corresponding replication slot, the `postgresql.use_slots` must be enabled (default).
6. Last, but not least, we track whether logical slots on standby are safe to use after promotion and in case if it is not - log the ERROR. Maybe in the future we will start removing unsafe logical slots.

Implementing this feature required some refactoring of the code that interacts with DCS. So far Patroni was using the `/optime/leader` key in order to persist the last known LSN of the primary. In case of the logical replication slots configured we also have to persist information about the `confirmed_flush_lsn` for all permanent logical slots that exist on the primary. This information is used for calling `pg_replication_slot_advance()` function on standbys. Since we don't want to increase the number of requests to the DCS, the best thing we could to do - saving information about the leader optime and logical slots position into the new key, `/status`. Another challenge is backward compatibility. Old Patroni version doesn't know anything about the new key and the format of the value of `/status` field therefore we need to keep updating the `/optime/leader` if there are cluster members with Patroni `2.0.1` and older. Introducing the new key in the JSON format makes it easier implementing new features in the future. For example we could keep there information not only about permanent logical slots, but also about physical.

The old behavior, i.e. creating logical slots immediately after the promotion is disabled. Moreover, in the case of PostgreSQL 10 and older,  Patroni will refuse to create permanent logical slots on the primary and report the error.

erthalion: :+1: 
CyberDem0n: :+1:

Patch:
diff --git a/.github/workflows/install_deps.py b/.github/workflows/install_deps.py
index e4281692..1a4477ea 100644
--- a/.github/workflows/install_deps.py
+++ b/.github/workflows/install_deps.py
@@ -39,7 +39,7 @@ def install_packages(what):
     }
     packages['exhibitor'] = packages['zookeeper']
     packages = packages.get(what, [])
-    ver = str({'etcd': '9.6', 'etcd3': '9.6', 'consul': 10, 'exhibitor': 11, 'kubernetes': 12, 'raft': 13}.get(what))
+    ver = str({'etcd': '9.6', 'etcd3': '13', 'consul': 12, 'exhibitor': 11, 'kubernetes': 13, 'raft': 12}.get(what))
     subprocess.call(['sudo', 'apt-get', 'update', '-y'])
     return subprocess.call(['sudo', 'apt-get', 'install', '-y', 'postgresql-' + ver, 'expect-dev', 'wget'] + packages)
 
diff --git a/docs/SETTINGS.rst b/docs/SETTINGS.rst
index 499623b9..cd951e68 100644
--- a/docs/SETTINGS.rst
+++ b/docs/SETTINGS.rst
@@ -34,7 +34,7 @@ Dynamic configuration is stored in the DCS (Distributed Configuration Store) and
     -  **restore\_command**: command to restore WAL records from the remote master to standby leader, can be different from the list defined in :ref:`postgresql_settings`
     -  **archive\_cleanup\_command**: cleanup command for standby leader
     -  **recovery\_min\_apply\_delay**: how long to wait before actually apply WAL records on a standby leader
--  **slots**: define permanent replication slots. These slots will be preserved during switchover/failover. Patroni will try to create slots before opening connections to the cluster.
+-  **slots**: define permanent replication slots. These slots will be preserved during switchover/failover. The logical slots are copied from the primary to a standby with restart, and after that their position advanced every **loop_wait** seconds (if necessary). Copying logical slot files performed via ``libpq`` connection and using either rewind or superuser credentials (see **postgresql.authentication** section). There is always a chance that the logical slot position on the replica is a bit behind the former primary, therefore application should be prepared that some messages could be received the second time after the failover. The easiest way of doing so - tracking ``confirmed_flush_lsn``. Enabling permanent logical replication slots requires **postgresql.use_slots** to be set and will also automatically enable the ``hot_standby_feedback``. Since the failover of logical replication slots is unsafe on PostgreSQL 9.6 and older and PostgreSQL version 10 is missing some important functions, the feature only works with PostgreSQL 11+.
     -  **my_slot_name**: the name of replication slot. If the permanent slot name matches with the name of the current primary it will not be created. Everything else is the responsibility of the operator to make sure that there are no clashes in names between replication slots automatically created by Patroni for members and permanent replication slots.
         -  **type**: slot type. Could be ``physical`` or ``logical``. If the slot is logical, you have to additionally define ``database`` and ``plugin``.
         -  **database**: the database name where logical slots should be created.
diff --git a/features/callback.py b/features/callback.py
deleted file mode 100755
index 767be49f..00000000
--- a/features/callback.py
+++ /dev/null
@@ -1,17 +0,0 @@
-#!/usr/bin/env python
-import os
-import psycopg2
-import sys
-
-
-if __name__ == '__main__':
-    if not (len(sys.argv) >= 3 and sys.argv[3] == "master"):
-        sys.exit(1)
-
-    os.environ['PGPASSWORD'] = 'zalando'
-    connection = psycopg2.connect(host='127.0.0.1', port=sys.argv[1], user='postgres')
-    cursor = connection.cursor()
-    cursor.execute("SELECT slot_name FROM pg_replication_slots WHERE slot_type = 'logical'")
-
-    with open("data/postgres0/label", "w") as label:
-        label.write(next(iter(cursor.fetchone()), ""))
diff --git a/features/standby_cluster.feature b/features/standby_cluster.feature
index 2e24c299..bad8c51a 100644
--- a/features/standby_cluster.feature
+++ b/features/standby_cluster.feature
@@ -1,5 +1,5 @@
 Feature: standby cluster
-  Scenario: check permanent logical slots are preserved on failover/switchover
+  Scenario: prepare the cluster with logical slots
     Given I start postgres1
     Then postgres1 is a leader after 10 seconds
     And there is a non empty initialize key in DCS after 15 seconds
@@ -10,15 +10,24 @@ Feature: standby cluster
     When I issue a PATCH request to http://127.0.0.1:8009/config with {"slots": {"test_logical": {"type": "logical", "database": "postgres", "plugin": "test_decoding"}}}
     Then I receive a response code 200
     And I do a backup of postgres1
-    When I start postgres0 with callback configured
+    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
+
+  @skip
+  Scenario: check permanent logical slots are synced to the replica
+    Given I run patronictl.py restart batman postgres1 --force
+    Then Logical slot test_logical is in sync between postgres0 and postgres1 after 10 seconds
+    When I add the table replicate_me to postgres1
+    And I get all changes from logical slot test_logical on postgres1
+    Then Logical slot test_logical is in sync between postgres0 and postgres1 after 10 seconds
+
+  Scenario: Detach exiting node from the cluster
     When I shut down postgres1
     Then postgres0 is a leader after 10 seconds
     And "members/postgres0" key in DCS has role=master after 3 seconds
     When I issue a GET request to http://127.0.0.1:8008/
     Then I receive a response code 200
-    And there is a label with "test_logical" in postgres0 data directory
 
   Scenario: check replication of a single table in a standby cluster
     Given I start postgres1 in a standby cluster batman1 as a clone of postgres0
@@ -35,6 +44,7 @@ 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
+    And postgres1 does not have a logical replication slot named test_logical
 
   Scenario: check failover
     When I kill postgres1
diff --git a/features/steps/slots.py b/features/steps/slots.py
index 8e87b064..f761cceb 100644
--- a/features/steps/slots.py
+++ b/features/steps/slots.py
@@ -1,5 +1,7 @@
+import time
+import psycopg2
+
 from behave import step, then
-import psycopg2 as pg
 
 
 @step('I create a logical replication slot {slot_name} on {pg_name:w} with the {plugin:w} plugin')
@@ -8,7 +10,7 @@ def create_logical_replication_slot(context, slot_name, pg_name, plugin):
         output = context.pctl.query(pg_name, ("SELECT pg_create_logical_replication_slot('{0}', '{1}'),"
                                               " current_database()").format(slot_name, plugin))
         print(output.fetchone())
-    except pg.Error as e:
+    except psycopg2.Error as e:
         print(e)
         assert False, "Error creating slot {0} on {1} with plugin {2}".format(slot_name, pg_name, plugin)
 
@@ -22,7 +24,7 @@ def has_logical_replication_slot(context, pg_name, slot_name, plugin):
         assert row[0] == "logical", "Found replication slot named {0} but wasn't a logical slot".format(slot_name)
         assert row[1] == plugin, ("Found replication slot named {0} but was using plugin "
                                   "{1} rather than {2}").format(slot_name, row[1], plugin)
-    except pg.Error:
+    except psycopg2.Error:
         assert False, "Error looking for slot {0} on {1} with plugin {2}".format(slot_name, pg_name, plugin)
 
 
@@ -32,5 +34,27 @@ def does_not_have_logical_replication_slot(context, pg_name, slot_name):
         row = context.pctl.query(pg_name, ("SELECT 1 FROM pg_replication_slots"
                                            " WHERE slot_name = '{0}'").format(slot_name)).fetchone()
         assert not row, "Found unexpected replication slot named {0}".format(slot_name)
-    except pg.Error:
+    except psycopg2.Error:
         assert False, "Error looking for slot {0} on {1}".format(slot_name, pg_name)
+
+
+@step('Logical slot {slot_name:w} is in sync between {pg_name1:w} and {pg_name2:w} after {time_limit:d} seconds')
+def logical_slots_in_sync(context, slot_name, pg_name1, pg_name2, time_limit):
+    time_limit *= context.timeout_multiplier
+    max_time = time.time() + int(time_limit)
+    while time.time() < max_time:
+        try:
+            query = "SELECT confirmed_flush_lsn FROM pg_replication_slots WHERE slot_name = '{0}'".format(slot_name)
+            slot1 = context.pctl.query(pg_name1, query).fetchone()
+            slot2 = context.pctl.query(pg_name2, query).fetchone()
+            if slot1[0] == slot2[0]:
+                return
+        except Exception:
+            pass
+        time.sleep(1)
+    assert False, "Logical slot {0} is not in sync between {1} and {2}".format(slot_name, pg_name1, pg_name2)
+
+
+@step('I get all changes from logical slot {slot_name:w} on {pg_name:w}')
+def logical_slot_get_changes(context, slot_name, pg_name):
+    context.pctl.query(pg_name, "SELECT * FROM pg_logical_slot_get_changes('{0}', NULL, NULL)".format(slot_name))
diff --git a/features/steps/standby_cluster.py b/features/steps/standby_cluster.py
index c04fd6f3..eb011dae 100644
--- a/features/steps/standby_cluster.py
+++ b/features/steps/standby_cluster.py
@@ -14,17 +14,6 @@ executable = sys.executable if os.name != 'nt' else sys.executable.replace('\\',
 callback = executable + " features/callback2.py "
 
 
-@step('I start {name:w} with callback configured')
-def start_patroni_with_callbacks(context, name):
-    return context.pctl.start(name, custom_config={
-        "postgresql": {
-            "callbacks": {
-                "on_role_change": executable + " features/callback.py"
-            }
-        }
-    })
-
-
 @step('I start {name:w} in a cluster {cluster_name:w}')
 def start_patroni(context, name, cluster_name):
     return context.pctl.start(name, custom_config={
@@ -55,7 +44,8 @@ def start_patroni_standby_cluster(context, name, cluster_name, name2):
                     "port": port,
                     "primary_slot_name": "pm_1",
                     "create_replica_methods": ["backup_restore", "basebackup"]
-                }
+                },
+                "postgresql": {"parameters": {"wal_level": "logical"}}
             }
         },
         "postgresql": {
diff --git a/patroni/api.py b/patroni/api.py
index e02d5acf..7bc9fc63 100644
--- a/patroni/api.py
+++ b/patroni/api.py
@@ -97,7 +97,7 @@ class RestApiHandler(BaseHTTPRequestHandler):
         patroni = self.server.patroni
         cluster = patroni.dcs.cluster
 
-        leader_optime = cluster and cluster.last_leader_operation or 0
+        leader_optime = cluster and cluster.last_lsn or 0
         replayed_location = response.get('xlog', {}).get('replayed_location', 0)
         max_replica_lag = parse_int(self.path_query.get('lag', [sys.maxsize])[0], 'B')
         if max_replica_lag is None:
diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py
index e5503f3e..10c2d5f8 100644
--- a/patroni/dcs/__init__.py
+++ b/patroni/dcs/__init__.py
@@ -13,12 +13,13 @@ import time
 
 from collections import defaultdict, namedtuple
 from copy import deepcopy
-from patroni.exceptions import PatroniFatalException
-from patroni.utils import parse_bool, uri
 from random import randint
 from six.moves.urllib_parse import urlparse, urlunparse, parse_qsl
 from threading import Event, Lock
 
+from ..exceptions import PatroniFatalException
+from ..utils import deep_compare, parse_bool, uri
+
 slot_name_re = re.compile('^[a-z0-9_]{1,63}$')
 logger = logging.getLogger(__name__)
 
@@ -133,6 +134,8 @@ class Member(namedtuple('Member', 'index,name,session,data')):
         else:
             try:
                 data = json.loads(data)
+                if not isinstance(data, dict):
+                    data = {}
             except (TypeError, ValueError):
                 data = {}
         return Member(index, name, session, data)
@@ -206,6 +209,15 @@ class Member(namedtuple('Member', 'index,name,session,data')):
     def is_running(self):
         return self.state == 'running'
 
+    @property
+    def version(self):
+        version = self.data.get('version')
+        if version:
+            try:
+                return tuple(map(int, version.split('.')))
+            except Exception:
+                logger.debug('Failed to parse Patroni version %s', version)
+
 
 class RemoteMember(Member):
     """ Represents a remote master for a standby cluster
@@ -259,14 +271,10 @@ class Leader(namedtuple('Leader', 'index,session,member')):
         """
         >>> Leader(1, '', Member.from_node(1, '', '', '{"version":"z"}')).checkpoint_after_promote
         """
-        version = self.data.get('version')
-        if version:
-            try:
-                # 1.5.6 is the last version which doesn't expose checkpoint_after_promote: false
-                if tuple(map(int, version.split('.'))) > (1, 5, 6):
-                    return self.data['role'] == 'master' and 'checkpoint_after_promote' not in self.data
-            except Exception:
-                logger.debug('Failed to parse Patroni version %s', version)
+        version = self.member.version
+        # 1.5.6 is the last version which doesn't expose checkpoint_after_promote: false
+        if version and version > (1, 5, 6):
+            return self.data.get('role') == 'master' and 'checkpoint_after_promote' not in self.data
 
 
 class Failover(namedtuple('Failover', 'index,leader,candidate,scheduled_at')):
@@ -432,19 +440,20 @@ class TimelineHistory(namedtuple('TimelineHistory', 'index,value,lines')):
         return TimelineHistory(index, value, lines)
 
 
-class Cluster(namedtuple('Cluster', 'initialize,config,leader,last_leader_operation,members,failover,sync,history')):
+class Cluster(namedtuple('Cluster', 'initialize,config,leader,last_lsn,members,failover,sync,history,slots')):
 
     """Immutable object (namedtuple) which represents PostgreSQL cluster.
     Consists of the following fields:
     :param initialize: shows whether this cluster has initialization key stored in DC or not.
     :param config: global dynamic configuration, reference to `ClusterConfig` object
     :param leader: `Leader` object which represents current leader of the cluster
-    :param last_leader_operation: int or long object containing position of last known leader operation.
-        This value is stored in `/optime/leader` key
+    :param last_lsn: int or long object containing position of last known leader LSN.
+        This value is stored in the `/status` key or `/optime/leader` (legacy) key
     :param members: list of Member object, all PostgreSQL cluster members including leader
     :param failover: reference to `Failover` object
     :param sync: reference to `SyncState` object, last observed synchronous replication state.
     :param history: reference to `TimelineHistory` object
+    :param slots: state of permanent logical replication slots on the primary in the format: {"slot_name": int}
     """
 
     def is_unlocked(self):
@@ -470,22 +479,41 @@ class Cluster(namedtuple('Cluster', 'initialize,config,leader,last_leader_operat
     def is_synchronous_mode(self):
         return self.check_mode('synchronous_mode')
 
-    def get_replication_slots(self, my_name, role):
+    @property
+    def __permanent_slots(self):
+        return self.config and self.config.permanent_slots or {}
+
+    @property
+    def __permanent_physical_slots(self):
+        return {name: value for name, value in self.__permanent_slots.items()
+                if not value or isinstance(value, dict) and value.get('type', 'physical') == 'physical'}
+
+    @property
+    def __permanent_logical_slots(self):
+        return {name: value for name, value in self.__permanent_slots.items() if isinstance(value, dict)
+                and value.get('type', 'logical') == 'logical' and value.get('database') and value.get('plugin')}
+
+    @property
+    def use_slots(self):
+        return self.config and self.config.data.get('postgresql', {}).get('use_slots', True)
+
+    def get_replication_slots(self, my_name, role, nofailover, major_version, show_error=False):
         # if the replicatefrom tag is set on the member - we should not create the replication slot for it on
         # the current master, because that member would replicate from elsewhere. We still create the slot if
         # the replicatefrom destination member is currently not a member of the cluster (fallback to the
         # master), or if replicatefrom destination member happens to be the current master
-        use_slots = self.config and self.config.data.get('postgresql', {}).get('use_slots', True)
+        use_slots = self.use_slots
         if role in ('master', 'standby_leader'):
             slot_members = [m.name for m in self.members if use_slots and m.name != my_name and
                             (m.replicatefrom is None or m.replicatefrom == my_name or
                              not self.has_member(m.replicatefrom))]
-            permanent_slots = (self.config and self.config.permanent_slots or {}).copy()
+            permanent_slots = self.__permanent_slots if use_slots and \
+                role == 'master' else self.__permanent_physical_slots
         else:
             # only manage slots for replicas that replicate from this one, except for the leader among them
             slot_members = [m.name for m in self.members if use_slots and
                             m.replicatefrom == my_name and m.name != self.leader.name]
-            permanent_slots = {}
+            permanent_slots = self.__permanent_logical_slots if use_slots and not nofailover else {}
 
         slots = {slot_name_from_member_name(name): {'type': 'physical'} for name in slot_members}
 
@@ -499,6 +527,7 @@ class Cluster(namedtuple('Cluster', 'initialize,config,leader,last_leader_operat
                                    for k, v in slot_conflicts.items() if len(v) > 1))
 
         # "merge" replication slots for members with permanent_replication_slots
+        disabled_permanent_logical_slots = []
         for name, value in permanent_slots.items():
             if not slot_name_re.match(name):
                 logger.error("Invalid permanent replication slot name '%s'", name)
@@ -516,7 +545,9 @@ class Cluster(namedtuple('Cluster', 'initialize,config,leader,last_leader_operat
                         slots[name] = value
                     continue
                 elif value['type'] == 'logical' and value.get('database') and value.get('plugin'):
-                    if name in slots:
+                    if major_version < 110000:
+                        disabled_permanent_logical_slots.append(name)
+                    elif name in slots:
                         logger.error("Permanent logical replication slot {'%s': %s} is conflicting with" +
                                      " physical replication slot for cluster member", name, value)
                     else:
@@ -525,20 +556,53 @@ class Cluster(namedtuple('Cluster', 'initialize,config,leader,last_leader_operat
 
             logger.error("Bad value for slot '%s' in permanent_slots: %s", name, permanent_slots[name])
 
+        if disabled_permanent_logical_slots and show_error:
+            logger.error("Permanent logical replication slots supported by Patroni only starting from PostgreSQL 11. "
+                         "Following slots will not be created: %s.", disabled_permanent_logical_slots)
+
         return slots
 
-    def has_permanent_logical_slots(self, name):
-        slots = self.get_replication_slots(name, 'master').values()
+    def has_permanent_logical_slots(self, my_name, nofailover, major_version=110000):
+        if major_version < 110000:
+            return False
+        slots = self.get_replication_slots(my_name, 'replica', nofailover, major_version).values()
         return any(v for v in slots if v.get("type") == "logical")
 
+    def should_enforce_hot_standby_feedback(self, my_name, nofailover, major_version):
+        """
+        The hot_standby_feedback must be enabled if the current replica has logical slots
+        or it is working as a cascading replica for the other node that has logical slots.
+        """
+
+        if major_version < 110000:
+            return False
+
+        if self.has_permanent_logical_slots(my_name, nofailover, major_version):
+            return True
+
+        if self.use_slots:
+            members = [m for m in self.members if m.replicatefrom == my_name and m.name != self.leader.name]
+            return any(self.should_enforce_hot_standby_feedback(m.name, m.nofailover, major_version) for m in members)
+        return False
+
+    def get_my_slot_name_on_primary(self, my_name, replicatefrom):
+        """
+        P <-- I <-- L
+        In case of cascading replication we have to check not our physical slot,
+        but slot of the replica that connects us to the primary.
+        """
+
+        m = self.get_member(replicatefrom, False) if replicatefrom else None
+        return self.get_my_slot_name_on_primary(m.name, m.replicatefrom) if m else slot_name_from_member_name(my_name)
+
     @property
     def timeline(self):
         """
-        >>> Cluster(0, 0, 0, 0, 0, 0, 0, 0).timeline
+        >>> Cluster(0, 0, 0, 0, 0, 0, 0, 0, 0).timeline
         0
-        >>> Cluster(0, 0, 0, 0, 0, 0, 0, TimelineHistory.from_node(1, '[]')).timeline
+        >>> Cluster(0, 0, 0, 0, 0, 0, 0, TimelineHistory.from_node(1, '[]'), 0).timeline
         1
-        >>> Cluster(0, 0, 0, 0, 0, 0, 0, TimelineHistory.from_node(1, '[["a"]]')).timeline
+        >>> Cluster(0, 0, 0, 0, 0, 0, 0, TimelineHistory.from_node(1, '[["a"]]'), 0).timeline
         0
         """
         if self.history:
@@ -551,6 +615,10 @@ class Cluster(namedtuple('Cluster', 'initialize,config,leader,last_leader_operat
                 return 1
         return 0
 
+    @property
+    def min_version(self):
+        return next(iter(sorted(filter(lambda v: v, [m.version for m in self.members])) + [None]))
+
 
 @six.add_metaclass(abc.ABCMeta)
 class AbstractDCS(object):
@@ -562,7 +630,8 @@ class AbstractDCS(object):
     _HISTORY = 'history'
     _MEMBERS = 'members/'
     _OPTIME = 'optime'
-    _LEADER_OPTIME = _OPTIME + '/' + _LEADER
+    _STATUS = 'status'  # JSON, containts "leader_lsn" and confirmed_flush_lsn of logical "slots" on the leader
+    _LEADER_OPTIME = _OPTIME + '/' + _LEADER  # legacy
     _SYNC = 'sync'
 
     def __init__(self, config):
@@ -578,7 +647,8 @@ class AbstractDCS(object):
         self._cluster = None
         self._cluster_valid_till = 0
         self._cluster_thread_lock = Lock()
-        self._last_leader_operation = ''
+        self._last_lsn = ''
+        self._last_status = {}
         self.event = Event()
 
     def client_path(self, path):
@@ -612,6 +682,10 @@ class AbstractDCS(object):
     def history_path(self):
         return self.client_path(self._HISTORY)
 
+    @property
+    def status_path(self):
+        return self.client_path(self._STATUS)
+
     @property
     def leader_optime_path(self):
         return self.client_path(self._LEADER_OPTIME)
@@ -682,14 +756,30 @@ class AbstractDCS(object):
             self._cluster_valid_till = 0
 
     @abc.abstractmethod
-    def _write_leader_optime(self, last_operation):
-        """write current xlog location into `/optime/leader` key in DCS
-        :param last_operation: absolute xlog location in bytes
+    def _write_leader_optime(self, last_lsn):
+        """write current WAL LSN into `/optime/leader` key in DCS
+
+        :param last_lsn: absolute WAL LSN in bytes
+        :returns: `!True` on success."""
+
+    def write_leader_optime(self, last_lsn):
+        if self._last_lsn != last_lsn and self._write_leader_optime(last_lsn):
+            self._last_lsn = last_lsn
+
+    @abc.abstractmethod
+    def _write_status(self, value):
+        """write current WAL LSN and confirmed_flush_lsn of permanent slots into the `/status` key in DCS
+
+        :param value: status serialized in JSON forman
         :returns: `!True` on success."""
 
-    def write_leader_optime(self, last_operation):
-        if self._last_leader_operation != last_operation and self._write_leader_optime(last_operation):
-            self._last_leader_operation = last_operation
+    def write_status(self, value):
+        if not deep_compare(self._last_status, value) and self._write_status(json.dumps(value, separators=(',', ':'))):
+            self._last_status = value
+        cluster = self.cluster
+        min_version = cluster and cluster.min_version
+        if min_version and min_version < (2, 0, 3):
+            self._write_leader_optime(str(value[self._OPTIME]))
 
     @abc.abstractmethod
     def _update_leader(self):
@@ -701,16 +791,20 @@ class AbstractDCS(object):
         You have to use CAS (Compare And Swap) operation in order to update leader key,
         for example for etcd `prevValue` parameter must be used."""
 
-    def update_leader(self, last_operation, access_is_restricted=False):
+    def update_leader(self, last_lsn, slots=None):
         """Update leader key (or session) ttl and optime/leader
 
-        :param last_operation: absolute xlog location in bytes
+        :param last_lsn: absolute WAL LSN in bytes
+        :param slots: dict with permanent slots confirmed_flush_lsn
         :returns: `!True` if leader key (or session) has been updated successfully.
             If not, `!False` must be returned and current instance would be demoted."""
 
         ret = self._update_leader()
-        if ret and last_operation:
-            self.write_leader_optime(last_operation)
+        if ret and last_lsn:
+            status = {self._OPTIME: last_lsn}
+            if slots:
+                status['slots'] = slots
+            self.write_status(status)
         return ret
 
     @abc.abstractmethod
@@ -779,13 +873,13 @@ class AbstractDCS(object):
         """Remove leader key from DCS.
         This method should remove leader key if current instance is the leader"""
 
-    def delete_leader(self, last_operation=None):
+    def delete_leader(self, last_lsn=None):
         """Update optime/leader and voluntarily remove leader key from DCS.
         This method should remove leader key if current instance is the leader.
-        :param last_operation: latest checkpoint location in bytes"""
+        :param last_lsn: latest checkpoint location in bytes"""
 
-        if last_operation:
-            self.write_leader_optime(last_operation)
+        if last_lsn:
+            self.write_leader_optime(last_lsn)
         return self._delete_leader()
 
     @abc.abstractmethod
diff --git a/patroni/dcs/consul.py b/patroni/dcs/consul.py
index a9acb08b..6c395252 100644
--- a/patroni/dcs/consul.py
+++ b/patroni/dcs/consul.py
@@ -337,9 +337,24 @@ class Consul(AbstractDCS):
             history = nodes.get(self._HISTORY)
             history = history and TimelineHistory.from_node(history['ModifyIndex'], history['Value'])
 
-            # get last leader operation
-            last_leader_operation = nodes.get(self._LEADER_OPTIME)
-            last_leader_operation = 0 if last_leader_operation is None else int(last_leader_operation['Value'])
+            # get last known leader lsn and slots
+            status = nodes.get(self._STATUS)
+            if status:
+                try:
+                    status = json.loads(status['Value'])
+                    last_lsn = status.get(self._OPTIME)
+                    slots = status.get('slots')
+                except Exception:
+                    slots = last_lsn = None
+            else:
+                last_lsn = nodes.get(self._LEADER_OPTIME)
+                last_lsn = last_lsn and last_lsn['Value']
+                slots = None
+
+            try:
+                last_lsn = int(last_lsn)
+            except Exception:
+                last_lsn = 0
 
             # get list of members
             members = [self.member(n) for k, n in nodes.items() if k.startswith(self._MEMBERS) and k.count('/') == 1]
@@ -366,9 +381,9 @@ class Consul(AbstractDCS):
             sync = nodes.get(self._SYNC)
             sync = SyncState.from_node(sync and sync['ModifyIndex'], sync and sync['Value'])
 
-            return Cluster(initialize, config, leader, last_leader_operation, members, failover, sync, history)
+            return Cluster(initialize, config, leader, last_lsn, members, failover, sync, history, slots)
         except NotFound:
-            return Cluster(None, None, None, None, [], None, None, None)
+            return Cluster(None, None, None, None, [], None, None, None, None)
         except Exception:
             logger.exception('get_cluster')
             raise ConsulError('Consul is not responding properly')
@@ -491,8 +506,12 @@ class Consul(AbstractDCS):
         return self._client.kv.put(self.config_path, value, cas=index)
 
     @catch_consul_errors
-    def _write_leader_optime(self, last_operation):
-        return self._client.kv.put(self.leader_optime_path, last_operation)
+    def _write_leader_optime(self, last_lsn):
+        return self._client.kv.put(self.leader_optime_path, last_lsn)
+
+    @catch_consul_errors
+    def _write_status(self, value):
+        return self._client.kv.put(self.status_path, value)
 
     @catch_consul_errors
     def _update_leader(self):
diff --git a/patroni/dcs/etcd.py b/patroni/dcs/etcd.py
index b6e20ba0..5dfffad7 100644
--- a/patroni/dcs/etcd.py
+++ b/patroni/dcs/etcd.py
@@ -602,9 +602,24 @@ class Etcd(AbstractEtcd):
             history = nodes.get(self._HISTORY)
             history = history and TimelineHistory.from_node(history.modifiedIndex, history.value)
 
-            # get last leader operation
-            last_leader_operation = nodes.get(self._LEADER_OPTIME)
-            last_leader_operation = 0 if last_leader_operation is None else int(last_leader_operation.value)
+            # get last know leader lsn and slots
+            status = nodes.get(self._STATUS)
+            if status:
+                try:
+                    status = json.loads(status.value)
+                    last_lsn = status.get(self._OPTIME)
+                    slots = status.get('slots')
+                except Exception:
+                    slots = last_lsn = None
+            else:
+                last_lsn = nodes.get(self._LEADER_OPTIME)
+                last_lsn = last_lsn and last_lsn.value
+                slots = None
+
+            try:
+                last_lsn = int(last_lsn)
+            except Exception:
+                last_lsn = 0
 
             # get list of members
             members = [self.member(n) for k, n in nodes.items() if k.startswith(self._MEMBERS) and k.count('/') == 1]
@@ -626,9 +641,9 @@ class Etcd(AbstractEtcd):
             sync = nodes.get(self._SYNC)
             sync = SyncState.from_node(sync and sync.modifiedIndex, sync and sync.value)
 
-            cluster = Cluster(initialize, config, leader, last_leader_operation, members, failover, sync, history)
+            cluster = Cluster(initialize, config, leader, last_lsn, members, failover, sync, history, slots)
         except etcd.EtcdKeyNotFound:
-            cluster = Cluster(None, None, None, None, [], None, None, None)
+            cluster = Cluster(None, None, None, None, [], None, None, None, None)
         except Exception as e:
             self._handle_exception(e, 'get_cluster', raise_ex=EtcdError('Etcd is not responding properly'))
         self._has_failed = False
@@ -665,8 +680,12 @@ class Etcd(AbstractEtcd):
         return self._client.write(self.config_path, value, prevIndex=index or 0)
 
     @catch_etcd_errors
-    def _write_leader_optime(self, last_operation):
-        return self._client.set(self.leader_optime_path, last_operation)
+    def _write_leader_optime(self, last_lsn):
+        return self._client.set(self.leader_optime_path, last_lsn)
+
+    @catch_etcd_errors
+    def _write_status(self, value):
+        return self._client.set(self.status_path, value)
 
     @catch_etcd_errors
     def _update_leader(self):
diff --git a/patroni/dcs/etcd3.py b/patroni/dcs/etcd3.py
index 2d9c0656..51786676 100644
--- a/patroni/dcs/etcd3.py
+++ b/patroni/dcs/etcd3.py
@@ -654,9 +654,24 @@ class Etcd3(AbstractEtcd):
             history = nodes.get(self._HISTORY)
             history = history and TimelineHistory.from_node(history['mod_revision'], history['value'])
 
-            # get last leader operation
-            last_leader_operation = nodes.get(self._LEADER_OPTIME)
-            last_leader_operation = 0 if last_leader_operation is None else int(last_leader_operation['value'])
+            # get last know leader lsn and slots
+            status = nodes.get(self._STATUS)
+            if status:
+                try:
+                    status = json.loads(status['value'])
+                    last_lsn = status.get(self._OPTIME)
+                    slots = status.get('slots')
+                except Exception:
+                    slots = last_lsn = None
+            else:
+                last_lsn = nodes.get(self._LEADER_OPTIME)
+                last_lsn = last_lsn and last_lsn['value']
+                slots = None
+
+            try:
+                last_lsn = int(last_lsn)
+            except Exception:
+                last_lsn = 0
 
             # get list of members
             members = [self.member(n) for k, n in nodes.items() if k.startswith(self._MEMBERS) and k.count('/') == 1]
@@ -680,7 +695,7 @@ class Etcd3(AbstractEtcd):
             sync = nodes.get(self._SYNC)
             sync = SyncState.from_node(sync and sync['mod_revision'], sync and sync['value'])
 
-            cluster = Cluster(initialize, config, leader, last_leader_operation, members, failover, sync, history)
+            cluster = Cluster(initialize, config, leader, last_lsn, members, failover, sync, history, slots)
         except UnsupportedEtcdVersion:
             raise
         except Exception as e:
@@ -741,8 +756,12 @@ class Etcd3(AbstractEtcd):
         return self._client.put(self.config_path, value, mod_revision=index)
 
     @catch_etcd_errors
-    def _write_leader_optime(self, last_operation):
-        return self._client.put(self.leader_optime_path, last_operation)
+    def _write_leader_optime(self, last_lsn):
+        return self._client.put(self.leader_optime_path, last_lsn)
+
+    @catch_etcd_errors
+    def _write_status(self, value):
+        return self._client.put(self.status_path, value)
 
     @catch_etcd_errors
     def _update_leader(self):
diff --git a/patroni/dcs/kubernetes.py b/patroni/dcs/kubernetes.py
index abd98a54..1a659036 100644
--- a/patroni/dcs/kubernetes.py
+++ b/patroni/dcs/kubernetes.py
@@ -724,9 +724,19 @@ class Kubernetes(AbstractDCS):
             self._leader_resource_version = metadata.resource_version if metadata else None
             annotations = metadata and metadata.annotations or {}
 
-            # get last leader operation
-            last_leader_operation = annotations.get(self._OPTIME)
-            last_leader_operation = 0 if last_leader_operation is None else int(last_leader_operation)
+            # get last known leader lsn
+            last_lsn = annotations.get(self._OPTIME)
+            try:
+                last_lsn = 0 if last_lsn is None else int(last_lsn)
+            except Exception:
+                last_lsn = 0
+
+            # get permanent slots state (confirmed_flush_lsn)
+            slots = annotations.get('slots')
+            try:
+                slots = slots and json.loads(slots)
+            except Exception:
+                slots = None
 
             # get leader
             leader_record = {n: annotations.get(n) for n in (self._LEADER, 'acquireTime',
@@ -760,7 +770,7 @@ class Kubernetes(AbstractDCS):
             metadata = sync and sync.metadata
             sync = SyncState.from_node(metadata and metadata.resource_version,  metadata and metadata.annotations)
 
-            return Cluster(initialize, config, leader, last_leader_operation, members, failover, sync, history)
+            return Cluster(initialize, config, leader, last_lsn, members, failover, sync, history, slots)
         except Exception:
             logger.exception('get_cluster')
             raise KubernetesError('Kubernetes API is not responding properly')
@@ -881,7 +891,10 @@ class Kubernetes(AbstractDCS):
                 return logger.exception('create_config_service failed')
         self._should_create_config_service = False
 
-    def _write_leader_optime(self, last_operation):
+    def _write_leader_optime(self, last_lsn):
+        """Unused"""
+
+    def _write_status(self, value):
         """Unused"""
 
     def _update_leader(self):
@@ -931,7 +944,7 @@ class Kubernetes(AbstractDCS):
 
         return self.patch_or_create(self.leader_path, annotations, kind_resource_version, ips=ips, retry=_retry)
 
-    def update_leader(self, last_operation, access_is_restricted=False):
+    def update_leader(self, last_lsn, slots=None):
         kind = self._kinds.get(self.leader_path)
         kind_annotations = kind and kind.metadata.annotations or {}
 
@@ -943,12 +956,12 @@ class Kubernetes(AbstractDCS):
         annotations = {self._LEADER: self._name, 'ttl': str(self._ttl), 'renewTime': now,
                        'acquireTime': leader_observed_record.get('acquireTime') or now,
                        'transitions': leader_observed_record.get('transitions') or '0'}
-        if last_operation:
-            annotations[self._OPTIME] = last_operation
+        if last_lsn:
+            annotations[self._OPTIME] = str(last_lsn)
+        annotations['slots'] = json.dumps(slots) if slots else None
 
         resource_version = kind and kind.metadata.resource_version
-        ips = [] if access_is_restricted else self.__ips
-        return self._update_leader_with_retry(annotations, resource_version, ips)
+        return self._update_leader_with_retry(annotations, resource_version, self.__ips)
 
     def attempt_to_acquire_leader(self, permanent=False):
         now = datetime.datetime.now(tzutc).isoformat()
@@ -1024,12 +1037,12 @@ class Kubernetes(AbstractDCS):
     def _delete_leader(self):
         """Unused"""
 
-    def delete_leader(self, last_operation=None):
+    def delete_leader(self, last_lsn=None):
         kind = self._kinds.get(self.leader_path)
         if kind and (kind.metadata.annotations or {}).get(self._LEADER) == self._name:
             annotations = {self._LEADER: None}
-            if last_operation:
-                annotations[self._OPTIME] = last_operation
+            if last_lsn:
+                annotations[self._OPTIME] = last_lsn
             self.patch_or_create(self.leader_path, annotations, kind.metadata.resource_version, True, False, [])
             self.reset_cluster()
 
diff --git a/patroni/dcs/raft.py b/patroni/dcs/raft.py
index 811e3ff6..f94e6e64 100644
--- a/patroni/dcs/raft.py
+++ b/patroni/dcs/raft.py
@@ -4,11 +4,12 @@ import os
 import threading
 import time
 
-from patroni.dcs import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, SyncState, TimelineHistory
-from ..utils import validate_directory
 from pysyncobj import SyncObj, SyncObjConf, replicated, FAIL_REASON
 from pysyncobj.transport import Node, TCPTransport, CONNECTION_STATE
 
+from . import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, SyncState, TimelineHistory
+from ..utils import validate_directory
+
 logger = logging.getLogger(__name__)
 
 
@@ -328,7 +329,7 @@ class Raft(AbstractDCS):
         prefix = self.client_path('')
         response = self._sync_obj.get(prefix, recursive=True)
         if not response:
-            return Cluster(None, None, None, None, [], None, None, None)
+            return Cluster(None, None, None, None, [], None, None, None, None)
         nodes = {os.path.relpath(key, prefix).replace('\\', '/'): value for key, value in response.items()}
 
         # get initialize flag
@@ -343,9 +344,24 @@ class Raft(AbstractDCS):
         history = nodes.get(self._HISTORY)
         history = history and TimelineHistory.from_node(history['index'], history['value'])
 
-        # get last leader operation
-        last_leader_operation = nodes.get(self._LEADER_OPTIME)
-        last_leader_operation = 0 if last_leader_operation is None else int(last_leader_operation['value'])
+        # get last know leader lsn and slots
+        status = nodes.get(self._STATUS)
+        if status:
+            try:
+                status = json.loads(status['value'])
+                last_lsn = status.get(self._OPTIME)
+                slots = status.get('slots')
+            except Exception:
+                slots = last_lsn = None
+        else:
+            last_lsn = nodes.get(self._LEADER_OPTIME)
+            last_lsn = last_lsn and last_lsn['value']
+            slots = None
+
+        try:
+            last_lsn = int(last_lsn)
+        except Exception:
+            last_lsn = 0
 
         # get list of members
         members = [self.member(k, n) for k, n in nodes.items() if k.startswith(self._MEMBERS) and k.count('/') == 1]
@@ -366,10 +382,13 @@ class Raft(AbstractDCS):
         sync = nodes.get(self._SYNC)
         sync = SyncState.from_node(sync and sync['index'], sync and sync['value'])
 
-        return Cluster(initialize, config, leader, last_leader_operation, members, failover, sync, history)
+        return Cluster(initialize, config, leader, last_lsn, members, failover, sync, history, slots)
+
+    def _write_leader_optime(self, last_lsn):
+        return self._sync_obj.set(self.leader_optime_path, last_lsn, timeout=1)
 
-    def _write_leader_optime(self, last_operation):
-        return self._sync_obj.set(self.leader_optime_path, last_operation, timeout=1)
+    def _write_status(self, value):
+        return self._sync_obj.set(self.status_path, value, timeout=1)
 
     def _update_leader(self):
         ret = self._sync_obj.set(self.leader_path, self._name, ttl=self._ttl, prevValue=self._name)
diff --git a/patroni/dcs/zookeeper.py b/patroni/dcs/zookeeper.py
index a7732cd4..3be8adb4 100644
--- a/patroni/dcs/zookeeper.py
+++ b/patroni/dcs/zookeeper.py
@@ -6,9 +6,10 @@ import time
 from kazoo.client import KazooClient, KazooState, KazooRetry
 from kazoo.exceptions import NoNodeError, NodeExistsError
 from kazoo.handlers.threading import SequentialThreadingHandler
-from patroni.dcs import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, SyncState, TimelineHistory
-from patroni.exceptions import DCSError
-from patroni.utils import deep_compare
+
+from . import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, SyncState, TimelineHistory
+from ..exceptions import DCSError
+from ..utils import deep_compare
 
 logger = logging.getLogger(__name__)
 
@@ -74,7 +75,7 @@ class ZooKeeper(AbstractDCS):
         self._client.add_listener(self.session_listener)
 
         self._fetch_cluster = True
-        self._fetch_optime = True
+        self._fetch_status = True
 
         self._orig_kazoo_connect = self._client._connection._connect
         self._client._connection._connect = self._kazoo_connect
@@ -100,13 +101,13 @@ class ZooKeeper(AbstractDCS):
         if state in [KazooState.SUSPENDED, KazooState.LOST]:
             self.cluster_watcher(None)
 
-    def optime_watcher(self, event):
-        self._fetch_optime = True
+    def status_watcher(self, event):
+        self._fetch_status = True
         self.event.set()
 
     def cluster_watcher(self, event):
         self._fetch_cluster = True
-        self.optime_watcher(event)
+        self.status_watcher(event)
 
     def reload_config(self, config):
         self.set_retry_timeout(config['retry_timeout'])
@@ -151,11 +152,29 @@ class ZooKeeper(AbstractDCS):
         except NoNodeError:
             return None
 
-    def get_leader_optime(self, leader):
-        watch = self.optime_watcher if not leader or leader.name != self._name else None
-        optime = self.get_node(self.leader_optime_path, watch)
-        self._fetch_optime = False
-        return optime and int(optime[0]) or 0
+    def get_status(self, leader):
+        watch = self.status_watcher if not leader or leader.name != self._name else None
+
+        status = self.get_node(self.status_path, watch)
+        if status:
+            try:
+                status = json.loads(status[0])
+                last_lsn = status.get(self._OPTIME)
+                slots = status.get('slots')
+            except Exception:
+                slots = last_lsn = None
+        else:
+            last_lsn = self.get_node(self.leader_optime_path, watch)
+            last_lsn = last_lsn and last_lsn[0]
+            slots = None
+
+        try:
+            last_lsn = int(last_lsn)
+        except Exception:
+            last_lsn = 0
+
+        self._fetch_status = False
+        return last_lsn, slots
 
     @staticmethod
     def member(name, value, znode):
@@ -218,14 +237,14 @@ class ZooKeeper(AbstractDCS):
                 leader = Leader(leader[1].version, leader[1].ephemeralOwner, member)
                 self._fetch_cluster = member.index == -1
 
-        # get last leader operation
-        last_leader_operation = self._OPTIME in nodes and self.get_leader_optime(leader)
+        # get last known leader lsn and slots
+        last_lsn, slots = self.get_status(leader)
 
         # failover key
         failover = self.get_node(self.failover_path, watch=self.cluster_watcher) if self._FAILOVER in nodes else None
         failover = failover and Failover.from_node(failover[1].version, failover[0])
 
-        return Cluster(initialize, config, leader, last_leader_operation, members, failover, sync, history)
+        return Cluster(initialize, config, leader, last_lsn, members, failover, sync, history, slots)
 
     def _load_cluster(self):
         cluster = self.cluster
@@ -237,12 +256,13 @@ class ZooKeeper(AbstractDCS):
                 self.cluster_watcher(None)
                 raise ZooKeeperError('ZooKeeper in not responding properly')
         # Optime ZNode was updated or doesn't exist and we are not leader
-        elif (self._fetch_optime and not self._fetch_cluster or not cluster.last_leader_operation) and\
+        elif (self._fetch_status and not self._fetch_cluster or not cluster.last_lsn
+              or cluster.has_permanent_logical_slots(self._name, False) and not cluster.slots) and\
                 not (cluster.leader and cluster.leader.name == self._name):
             try:
-                optime = self.get_leader_optime(cluster.leader)
-                cluster = Cluster(cluster.initialize, cluster.config, cluster.leader, optime,
-                                  cluster.members, cluster.failover, cluster.sync, cluster.history)
+                last_lsn, slots = self.get_status(cluster.leader)
+                cluster = Cluster(cluster.initialize, cluster.config, cluster.leader, last_lsn,
+                                  cluster.members, cluster.failover, cluster.sync, cluster.history, slots)
             except Exception:
                 pass
         return cluster
@@ -336,8 +356,11 @@ class ZooKeeper(AbstractDCS):
     def take_leader(self):
         return self.attempt_to_acquire_leader()
 
-    def _write_leader_optime(self, last_operation):
-        return self._set_or_create(self.leader_optime_path, last_operation)
+    def _write_leader_optime(self, last_lsn):
+        return self._set_or_create(self.leader_optime_path, last_lsn)
+
+    def _write_status(self, value):
+        return self._set_or_create(self.status_path, value)
 
     def _update_leader(self):
         return True
@@ -374,6 +397,6 @@ class ZooKeeper(AbstractDCS):
 
     def watch(self, leader_index, timeout):
         ret = super(ZooKeeper, self).watch(leader_index, timeout)
-        if ret and not self._fetch_optime:
+        if ret and not self._fetch_status:
             self._fetch_cluster = True
         return ret or self._fetch_cluster
diff --git a/patroni/ha.py b/patroni/ha.py
index 6adf437c..8d94a94d 100644
--- a/patroni/ha.py
+++ b/patroni/ha.py
@@ -66,7 +66,6 @@ class Ha(object):
         self.old_cluster = None
         self._is_leader = False
         self._is_leader_lock = RLock()
-        self._leader_access_is_restricted = False
         self._was_paused = False
         self._leader_timeline = None
         self.recovering = False
@@ -121,16 +120,12 @@ class Ha(object):
 
     def is_leader(self):
         with self._is_leader_lock:
-            return self._is_leader > time.time() and not self._leader_access_is_restricted
+            return self._is_leader > time.time()
 
     def set_is_leader(self, value):
         with self._is_leader_lock:
             self._is_leader = time.time() + self.dcs.ttl if value else 0
 
-    def set_leader_access_is_restricted(self, value):
-        with self._is_leader_lock:
-            self._leader_access_is_restricted = value
-
     def load_cluster_from_dcs(self):
         cluster = self.dcs.get_cluster()
 
@@ -145,20 +140,20 @@ class Ha(object):
         self._leader_timeline = None if cluster.is_unlocked() else cluster.leader.timeline
 
     def acquire_lock(self):
-        self.set_leader_access_is_restricted(self.cluster.has_permanent_logical_slots(self.state_handler.name))
         ret = self.dcs.attempt_to_acquire_leader()
         self.set_is_leader(ret)
         return ret
 
     def update_lock(self, write_leader_optime=False):
-        last_operation = None
+        last_lsn = slots = None
         if write_leader_optime:
             try:
-                last_operation = self.state_handler.last_operation()
+                last_lsn = self.state_handler.last_operation()
+                slots = self.state_handler.slots()
             except Exception:
                 logger.exception('Exception when called state_handler.last_operation()')
         try:
-            ret = self.dcs.update_leader(last_operation, self._leader_access_is_restricted)
+            ret = self.dcs.update_leader(last_lsn, slots)
         except Exception:
             logger.exception('Unexpected exception raised from update_leader, please report it as a BUG')
             ret = False
@@ -613,8 +608,6 @@ class Ha(object):
                     return 'Postponing promotion because synchronous replication state was updated by somebody else'
                 self.state_handler.config.set_synchronous_standby(['*'] if self.is_synchronous_mode_strict() else [])
             if self.state_handler.role != 'master':
-                self.set_leader_access_is_restricted(self.cluster.has_permanent_logical_slots(self.state_handler.name))
-
                 def on_success():
                     self._rewind.reset_state()
                     logger.info("cleared rewind state after becoming the leader")
@@ -622,8 +615,7 @@ class Ha(object):
                 with self._async_response:
                     self._async_response.reset()
                 self._async_executor.try_run_async('promote', self.state_handler.promote,
-                                                   args=(self.dcs.loop_wait, self._async_response, on_success,
-                                                         self._leader_access_is_restricted))
+                                                   args=(self.dcs.loop_wait, self._async_response, on_success))
             return promote_message
 
     def fetch_node_status(self, member):
@@ -653,14 +645,13 @@ class Ha(object):
         :param wal_position: Current wal position.
         :returns True when node is lagging
         """
-        lag = (self.cluster.last_leader_operation or 0) - wal_position
+        lag = (self.cluster.last_lsn or 0) - wal_position
         return lag > self.patroni.config.get('maximum_lag_on_failover', 0)
 
     def _is_healthiest_node(self, members, check_replication_lag=True):
         """This method tries to determine whether I am healthy enough to became a new leader candidate or not."""
 
-        # We don't call `last_operation()` here because it returns a string
-        _, my_wal_position, _ = self.state_handler.timeline_wal_position()
+        my_wal_position = self.state_handler.last_operation()
         if check_replication_lag and self.is_lagging(my_wal_position):
             logger.info('My wal position exceeds maximum replication lag')
             return False  # Too far behind last reported wal position on master
@@ -802,13 +793,13 @@ class Ha(object):
 
         return self._is_healthiest_node(members.values())
 
-    def _delete_leader(self, last_operation=None):
+    def _delete_leader(self, last_lsn=None):
         self.set_is_leader(False)
-        self.dcs.delete_leader(last_operation)
+        self.dcs.delete_leader(last_lsn)
         self.dcs.reset_cluster()
 
-    def release_leader_key_voluntarily(self, last_operation=None):
-        self._delete_leader(last_operation)
+    def release_leader_key_voluntarily(self, last_lsn=None):
+        self._delete_leader(last_lsn)
         self.touch_member()
         logger.info("Leader key released")
 
@@ -989,11 +980,6 @@ class Ha(object):
                 self._delete_leader()
                 return 'removed leader lock because postgres is not running as master'
 
-            if self.state_handler.is_leader() and self._leader_access_is_restricted:
-                self.state_handler.slots_handler.sync_replication_slots(self.cluster)
-                self.state_handler.call_nowait(ACTION_ON_ROLE_CHANGE)
-                self.set_leader_access_is_restricted(False)
-
             if self.update_lock(True):
                 msg = self.process_manual_failover_from_leader()
                 if msg is not None:
@@ -1261,7 +1247,6 @@ class Ha(object):
             self.cancel_initialization()
         self.dcs.initialize(create_new=(self.cluster.initialize is None), sysid=self.state_handler.sysid)
         self.dcs.set_config_value(json.dumps(self.patroni.config.dynamic_configuration, separators=(',', ':')))
-        self.state_handler.slots_handler.sync_replication_slots(self.cluster)
         self.dcs.take_leader()
         self.set_is_leader(True)
         self.state_handler.call_nowait(ACTION_ON_START)
@@ -1317,8 +1302,8 @@ class Ha(object):
     def _run_cycle(self):
         dcs_failed = False
         try:
-            self.state_handler.reset_cluster_info_state()
             self.load_cluster_from_dcs()
+            self.state_handler.reset_cluster_info_state(self.cluster, self.patroni.nofailover)
 
             if self.is_paused():
                 self.watchdog.disable()
@@ -1424,20 +1409,28 @@ class Ha(object):
 
             try:
                 if self.cluster.is_unlocked():
-                    return self.process_unhealthy_cluster()
+                    ret = self.process_unhealthy_cluster()
                 else:
                     msg = self.process_healthy_cluster()
-                    return self.evaluate_scheduled_restart() or msg
+                    ret = self.evaluate_scheduled_restart() or msg
             finally:
                 # 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 master)
                 if not self._async_executor.busy and not self.state_handler.is_starting():
-                    self.state_handler.slots_handler.sync_replication_slots(self.cluster)
+                    create_slots = self.state_handler.slots_handler.sync_replication_slots(self.cluster,
+                                                                                           self.patroni.nofailover)
                     if not self.state_handler.cb_called:
                         if not self.state_handler.is_leader():
                             self._rewind.trigger_check_diverged_lsn()
                         self.state_handler.call_nowait(ACTION_ON_START)
+                    if 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.leader, create_slots))
+                        if not err:
+                            ret = 'Copying logical slots {0} from the primary'.format(create_slots)
+            return ret
         except DCSError:
             dcs_failed = True
             logger.error('Error communicating with DCS')
@@ -1468,7 +1461,7 @@ class Ha(object):
             self.watchdog.disable()
         elif not self._join_aborted:
             # FIXME: If stop doesn't reach safepoint quickly enough keepalive is triggered. If shutdown checkpoint
-            # takes longer than ttl, then leader key is lost and replication might not have sent out all xlog.
+            # takes longer than ttl, then leader key is lost and replication might not have sent out all WAL.
             # This might not be the desired behavior of users, as a graceful shutdown of the host can mean lost data.
             # We probably need to something smarter here.
             disable_wd = self.watchdog.disable if self.watchdog.is_running else None
diff --git a/patroni/postgresql/__init__.py b/patroni/postgresql/__init__.py
index aced2d7c..b33fff8b 100644
--- a/patroni/postgresql/__init__.py
+++ b/patroni/postgresql/__init__.py
@@ -10,19 +10,20 @@ from contextlib import contextmanager
 from copy import deepcopy
 from dateutil import tz
 from datetime import datetime
-from patroni.postgresql.callback_executor import CallbackExecutor
-from patroni.postgresql.bootstrap import Bootstrap
-from patroni.postgresql.cancellable import CancellableSubprocess
-from patroni.postgresql.config import ConfigHandler, mtime
-from patroni.postgresql.connection import Connection, get_connection_cursor
-from patroni.postgresql.misc import parse_history, parse_lsn, postgres_major_version_to_int
-from patroni.postgresql.postmaster import PostmasterProcess
-from patroni.postgresql.slots import SlotsHandler
-from patroni.exceptions import PostgresConnectionException
-from patroni.utils import Retry, RetryFailedError, polling_loop, data_directory_is_empty, parse_int
 from psutil import TimeoutExpired
 from threading import current_thread, Lock
 
+from .callback_executor import CallbackExecutor
+from .bootstrap import Bootstrap
+from .cancellable import CancellableSubprocess
+from .config import ConfigHandler, mtime
+from .connection import Connection, get_connection_cursor
+from .misc import parse_history, parse_lsn, postgres_major_version_to_int
+from .postmaster import PostmasterProcess
+from .slots import SlotsHandler
+from ..exceptions import PostgresConnectionException
+from ..utils import Retry, RetryFailedError, polling_loop, data_directory_is_empty, parse_int
+
 
 logger = logging.getLogger(__name__)
 
@@ -54,7 +55,7 @@ class Postgresql(object):
               "pg_catalog.pg_current_{0}_{1}()), 1, 8))::bit(32)::int END, "  # master timeline
               "CASE WHEN pg_catalog.pg_is_in_recovery() THEN 0 "
               "ELSE pg_catalog.pg_{0}_{1}_diff(pg_catalog.pg_current_{0}_{1}(), '0/0')::bigint END, "  # write_lsn
-              "pg_catalog.pg_{0}_{1}_diff(pg_catalog.pg_last_{0}_replay_{1}(), '0/0')::bigint,  "
+              "pg_catalog.pg_{0}_{1}_diff(pg_catalog.pg_last_{0}_replay_{1}(), '0/0')::bigint, "
               "pg_catalog.pg_{0}_{1}_diff(COALESCE(pg_catalog.pg_last_{0}_receive_{1}(), '0/0'), '0/0')::bigint, "
               "pg_catalog.pg_is_in_recovery() AND pg_catalog.pg_is_{0}_replay_paused()")
 
@@ -101,6 +102,8 @@ class Postgresql(object):
         self._state_entry_timestamp = None
 
         self._cluster_info_state = {}
+        self._has_permanent_logical_slots = True
+        self._enforce_hot_standby_feedback = False
         self._cached_replica_timeline = None
 
         # Last known running process
@@ -152,14 +155,18 @@ class Postgresql(object):
     @property
     def cluster_info_query(self):
         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 = (", CASE WHEN latest_end_lsn IS NULL THEN NULL ELSE received_tli END,"
-                     " slot_name, conninfo FROM pg_catalog.pg_stat_get_wal_receiver()")
+                     " slot_name, conninfo, {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"
+            extra = "0, NULL, NULL, NULL, NULL"
 
         return ("SELECT " + self.TL_LSN + ", {2}").format(self.wal_name, self.lsn_name, extra)
 
@@ -299,16 +306,36 @@ class Postgresql(object):
             replica_methods = self.create_replica_methods
         return any(self.replica_method_can_work_without_replication_connection(m) for m in replica_methods)
 
-    def reset_cluster_info_state(self):
+    @property
+    def enforce_hot_standby_feedback(self):
+        return self._enforce_hot_standby_feedback
+
+    def set_enforce_hot_standby_feedback(self, value):
+        # If we enable or disable the hot_standby_feedback we need to update postgresql.conf and reload
+        if self._enforce_hot_standby_feedback != value:
+            self._enforce_hot_standby_feedback = value
+            if self.is_running():
+                self.config.write_postgresql_conf()
+                self.reload()
+
+    def reset_cluster_info_state(self, cluster, nofailover=None):
         self._cluster_info_state = {}
+        if cluster and cluster.config and cluster.config.modify_index:
+            self._has_permanent_logical_slots =\
+                cluster.has_permanent_logical_slots(self.name, nofailover, self.major_version)
+            self.set_enforce_hot_standby_feedback(
+                self._has_permanent_logical_slots or
+                cluster.should_enforce_hot_standby_feedback(self.name, nofailover, self.major_version))
 
     def _cluster_info_state_get(self, name):
         if not self._cluster_info_state:
             try:
                 result = self._is_leader_retry(self._query, self.cluster_info_query).fetchone()
-                self._cluster_info_state = dict(zip(['timeline', 'wal_position', 'replayed_location',
-                                                     'received_location', 'replay_paused', 'pg_control_timeline',
-                                                     'received_tli', 'slot_name', 'conninfo'], result))
+                cluster_info_state = dict(zip(['timeline', 'wal_position', 'replayed_location',
+                                               'received_location', 'replay_paused', 'pg_control_timeline',
+                                               'received_tli', 'slot_name', 'conninfo', 'slots'], result))
+                cluster_info_state['slots'] = self.slots_handler.process_permanent_slots(cluster_info_state['slots'])
+                self._cluster_info_state = cluster_info_state
             except RetryFailedError as e:  # SELECT failed two times
                 self._cluster_info_state = {'error': str(e)}
                 if not self.is_starting() and self.pg_isready() == STATE_REJECT:
@@ -325,6 +352,9 @@ class Postgresql(object):
     def received_location(self):
         return self._cluster_info_state_get('received_location')
 
+    def slots(self):
+        return self._cluster_info_state_get('slots')
+
     def primary_slot_name(self):
         return self._cluster_info_state_get('slot_name')
 
@@ -362,7 +392,7 @@ class Postgresql(object):
                 return self._postmaster_proc
             self._postmaster_proc = None
 
-        # we noticed that postgres was restarted, force syncing of replication
+        # we noticed that postgres was restarted, force syncing of replication slots and check of logical slots
         self.slots_handler.schedule()
 
         self._postmaster_proc = PostmasterProcess.from_pidfile(self._data_dir)
@@ -812,7 +842,7 @@ class Postgresql(object):
             logger.info('pre_promote script `%s` exited with %s', cmd, ret)
         return ret == 0
 
-    def promote(self, wait_seconds, task, on_success=None, access_is_restricted=False):
+    def promote(self, wait_seconds, task, on_success=None):
         if self.role == 'master':
             return True
 
@@ -829,13 +859,14 @@ class Postgresql(object):
             logger.info("PostgreSQL promote cancelled.")
             return False
 
+        self.slots_handler.on_promote()
+
         ret = self.pg_ctl('promote', '-W')
         if ret:
             self.set_role('master')
             if on_success is not None:
                 on_success()
-            if not access_is_restricted:
-                self.call_nowait(ACTION_ON_ROLE_CHANGE)
+            self.call_nowait(ACTION_ON_ROLE_CHANGE)
             ret = self._wait_promote(wait_seconds)
         return ret
 
@@ -873,8 +904,8 @@ class Postgresql(object):
             return None
 
     def last_operation(self):
-        return str(self._wal_position(self.is_leader(), self._cluster_info_state_get('wal_position'),
-                                      self.received_location(), self.replayed_location()))
+        return self._wal_position(self.is_leader(), self._cluster_info_state_get('wal_position'),
+                                  self.received_location(), self.replayed_location())
 
     def configure_server_parameters(self):
         self._major_version = self.get_major_version()
diff --git a/patroni/postgresql/config.py b/patroni/postgresql/config.py
index d6e70bc7..7ad794ef 100644
--- a/patroni/postgresql/config.py
+++ b/patroni/postgresql/config.py
@@ -387,15 +387,21 @@ class ConfigHandler(object):
         if 'custom_conf' not in self._config and not os.path.exists(self._postgresql_base_conf):
             os.rename(self._postgresql_conf, self._postgresql_base_conf)
 
+        configuration = configuration or self._server_parameters.copy()
+
         # In case we are using custom bootstrap from spilo image with PITR it fails if it contains increasing
         # values like Max_connections. We disable hot_standby so it will accept increasing values.
         if self._postgresql.bootstrap.running_custom_bootstrap:
             configuration['hot_standby'] = 'off'
 
+        # Due to the permanent logical replication slots configured we have to enable hot_standby_feedback
+        if self._postgresql.enforce_hot_standby_feedback:
+            configuration['hot_standby_feedback'] = 'on'
+
         with ConfigWriter(self._postgresql_conf) as f:
             include = self._config.get('custom_conf') or self._postgresql_base_conf_name
             f.writeline("include '{0}'\n".format(ConfigWriter.escape(include)))
-            for name, value in sorted((configuration or self._server_parameters).items()):
+            for name, value in sorted((configuration).items()):
                 value = transform_postgresql_parameter_value(self._postgresql.major_version, name, value)
                 if (not self._postgresql.bootstrap.running_custom_bootstrap or name != 'hba_file') \
                         and name not in self._RECOVERY_PARAMETERS and value is not None:
diff --git a/patroni/postgresql/misc.py b/patroni/postgresql/misc.py
index f5caffbd..1d93d3de 100644
--- a/patroni/postgresql/misc.py
+++ b/patroni/postgresql/misc.py
@@ -68,3 +68,8 @@ def parse_history(data):
                 yield values
             except (IndexError, ValueError):
                 logger.exception('Exception when parsing timeline history line "%s"', values)
+
+
+def format_lsn(lsn, full=False):
+    template = '{0:X}/{1:08X}' if full else '{0:X}/{1:X}'
+    return template.format(lsn >> 32, lsn & 0xFFFFFFFF)
diff --git a/patroni/postgresql/rewind.py b/patroni/postgresql/rewind.py
index 82591b51..519b0b24 100644
--- a/patroni/postgresql/rewind.py
+++ b/patroni/postgresql/rewind.py
@@ -7,7 +7,7 @@ import subprocess
 from threading import Lock, Thread
 
 from .connection import get_connection_cursor
-from .misc import parse_history, parse_lsn
+from .misc import format_lsn, parse_history, parse_lsn
 from ..async_executor import CriticalTask
 from ..dcs import Leader
 
@@ -17,11 +17,6 @@ REWIND_STATUS = type('Enum', (), {'INITIAL': 0, 'CHECKPOINT': 1, 'CHECK': 2, 'NE
                                   'NOT_NEED': 4, 'SUCCESS': 5, 'FAILED': 6})
 
 
-def format_lsn(lsn, full=False):
-    template = '{0:X}/{1:08X}' if full else '{0:X}/{1:X}'
-    return template.format(lsn >> 32, lsn & 0xFFFFFFFF)
-
-
 class Rewind(object):
 
     def __init__(self, postgresql):
diff --git a/patroni/postgresql/slots.py b/patroni/postgresql/slots.py
index 3738a25d..cee319d9 100644
--- a/patroni/postgresql/slots.py
+++ b/patroni/postgresql/slots.py
@@ -1,14 +1,33 @@
+import errno
 import logging
+import os
+import shutil
 
-from patroni.postgresql.connection import get_connection_cursor
 from collections import defaultdict
+from contextlib import contextmanager
+
+from .connection import get_connection_cursor
+from .misc import format_lsn
 
 logger = logging.getLogger(__name__)
 
 
-def compare_slots(s1, s2):
+def compare_slots(s1, s2, dbid='database'):
     return s1['type'] == s2['type'] and (s1['type'] == 'physical' or
-                                         s1['database'] == s2['database'] and s1['plugin'] == s2['plugin'])
+                                         s1.get(dbid) == s2.get(dbid) and s1['plugin'] == s2['plugin'])
+
+
+def fsync_dir(path):
+    if os.name != 'nt':
+        fd = os.open(path, os.O_DIRECTORY)
+        try:
+            os.fsync(fd)
+        except OSError as e:
+            # Some filesystems don't like fsyncing directories and raise EINVAL. Ignoring it is usually safe.
+            if e.errno != errno.EINVAL:
+                raise
+        finally:
+            os.close(fd)
 
 
 class SlotsHandler(object):
@@ -16,22 +35,66 @@ class SlotsHandler(object):
     def __init__(self, postgresql):
         self._postgresql = postgresql
         self._replication_slots = {}  # already existing replication slots
+        self._unready_logical_slots = set()
         self.schedule()
 
     def _query(self, sql, *params):
         return self._postgresql.query(sql, *params, retry=False)
 
+    @staticmethod
+    def _copy_items(src, dst, keys=None):
+        dst.update({key: src[key] for key in keys or ('datoid', 'catalog_xmin', 'confirmed_flush_lsn')})
+
+    def process_permanent_slots(self, slots):
+        """This methods solves three problems at once (I know, it is weird).
+
+        The cluster_info_query from `Postgresql` is executed every HA loop and returns
+        information about all replication slots that exists on the current host.
+        Based on this information we perform the following actions:
+        1. For the primary we want to expose to DCS permanent logical slots, therefore the method
+           builds (and returns) a dict, that maps permanent logical slot names and confirmed_flush_lsns.
+        2. This method also detects if one of the previously known permanent slots got missing and schedules resync.
+        3. Updates the local cache with the fresh catalog_xmin and confirmed_flush_lsn for every known slot.
+           This info is used when performing the check of logical slot readiness on standbys.
+        """
+        ret = {}
+
+        slots = {slot['slot_name']: slot for slot in slots or []}
+        if slots:
+            for name, value in slots.items():
+                if name in self._replication_slots:
+                    if compare_slots(value, self._replication_slots[name], 'datoid'):
+                        if value['type'] == 'logical':
+                            ret[name] = value['confirmed_flush_lsn']
+                            self._copy_items(value, self._replication_slots[name])
+                    else:
+                        self._schedule_load_slots = True
+
+        # It could happen that the slots was deleted in the background, we want to detect this case
+        if any(name not in slots for name in self._replication_slots.keys()):
+            self._schedule_load_slots = True
+
+        return ret
+
     def load_replication_slots(self):
         if self._postgresql.major_version >= 90400 and self._schedule_load_slots:
             replication_slots = {}
-            cursor = self._query('SELECT slot_name, slot_type, plugin, database FROM pg_catalog.pg_replication_slots')
+            extra = ", catalog_xmin, pg_catalog.pg_wal_lsn_diff(confirmed_flush_lsn, '0/0')::bigint"\
+                if self._postgresql.major_version >= 100000 else ""
+            cursor = self._query('SELECT slot_name, slot_type, plugin, database, datoid'
+                                 '{0} FROM pg_catalog.pg_replication_slots'.format(extra))
             for r in cursor:
                 value = {'type': r[1]}
                 if r[1] == 'logical':
-                    value.update({'plugin': r[2], 'database': r[3]})
+                    value.update(plugin=r[2], database=r[3], datoid=r[4])
+                    if self._postgresql.major_version >= 100000:
+                        value.update(catalog_xmin=r[5], confirmed_flush_lsn=r[6])
                 replication_slots[r[0]] = value
             self._replication_slots = replication_slots
             self._schedule_load_slots = False
+            if self._force_readiness_check:
+                self._unready_logical_slots = set(n for n, v in replication_slots.items() if v['type'] == 'logical')
+                self._force_readiness_check = False
 
     def ignore_replication_slot(self, cluster, name):
         slot = self._replication_slots[name]
@@ -47,65 +110,188 @@ class SlotsHandler(object):
         # In normal situation rowcount should be 1, otherwise either slot doesn't exists or it is still active
         return cursor.rowcount == 1
 
-    def sync_replication_slots(self, cluster):
+    def _drop_incorrect_slots(self, cluster, slots):
+        # drop old replication slots which are not presented in desired slots
+        for name in set(self._replication_slots) - set(slots):
+            if not self.ignore_replication_slot(cluster, name) and not self.drop_replication_slot(name):
+                logger.error("Failed to drop replication slot '%s'", name)
+                self._schedule_load_slots = True
+
+        for name, value in slots.items():
+            if name in self._replication_slots and not compare_slots(value, self._replication_slots[name]):
+                logger.info("Trying to drop replication slot '%s' because value is changing from %s to %s",
+                            name, self._replication_slots[name], value)
+                if self.drop_replication_slot(name):
+                    self._replication_slots.pop(name)
+                else:
+                    logger.error("Failed to drop replication slot '%s'", name)
+                    self._schedule_load_slots = True
+
+    def _ensure_physical_slots(self, slots):
+        immediately_reserve = ', true' if self._postgresql.major_version >= 90600 else ''
+        for name, value in slots.items():
+            if name not in self._replication_slots and value['type'] == 'physical':
+                try:
+                    self._query(("SELECT pg_catalog.pg_create_physical_replication_slot(%s{0})" +
+                                 " WHERE NOT EXISTS (SELECT 1 FROM pg_catalog.pg_replication_slots" +
+                                 " WHERE slot_type = 'physical' AND slot_name = %s)").format(
+                                     immediately_reserve), name, name)
+                except Exception:
+                    logger.exception("Failed to create physical replication slot '%s'", name)
+                self._schedule_load_slots = True
+
+    @contextmanager
+    def _get_local_connection_cursor(self, database):
+        conn_kwargs = self._postgresql.config.local_connect_kwargs
+        conn_kwargs['database'] = database
+        with get_connection_cursor(**conn_kwargs) as cur:
+            yield cur
+
+    def _ensure_logical_slots_primary(self, slots):
+        # Group logical slots to be created by database name
+        logical_slots = defaultdict(dict)
+        for name, value in slots.items():
+            if value['type'] == 'logical':
+                # If the logical already exists, copy some information about it into the original structure
+                if self._replication_slots.get(name, {}).get('datoid'):
+                    self._copy_items(self._replication_slots[name], value)
+                else:
+                    logical_slots[value['database']][name] = value
+
+        # Create new logical slots
+        for database, values in logical_slots.items():
+            with self._get_local_connection_cursor(database) as cur:
+                for name, value in values.items():
+                    try:
+                        cur.execute("SELECT pg_catalog.pg_create_logical_replication_slot(%s, %s)" +
+                                    " WHERE NOT EXISTS (SELECT 1 FROM pg_catalog.pg_replication_slots" +
+                                    " WHERE slot_type = 'logical' AND slot_name = %s)",
+                                    (name, value['plugin'], name))
+                    except Exception as e:
+                        logger.error("Failed to create logical replication slot '%s' plugin='%s': %r",
+                                     name, value['plugin'], e)
+                        slots.pop(name)
+                    self._schedule_load_slots = True
+
+    def _ensure_logical_slots_replica(self, cluster, slots):
+        advance_slots = defaultdict(dict)  # Group logical slots to be advanced by database name
+        create_slots = []  # And collect logical slots to be created on the replica
+        for name, value in slots.items():
+            if value['type'] == 'logical':
+                # If the logical already exists, copy some information about it into the original structure
+                if self._replication_slots.get(name, {}).get('datoid'):
+                    self._copy_items(self._replication_slots[name], value)
+                    if name in cluster.slots:
+                        try:  # Skip slots that doesn't need to be advanced
+                            if value['confirmed_flush_lsn'] < int(cluster.slots[name]):
+                                advance_slots[value['database']][name] = value
+                        except Exception as e:
+                            logger.error('Failed to parse "%s": %r', cluster.slots[name], e)
+                elif name in cluster.slots:  # We want to copy only slots with feedback in a DCS
+                    create_slots.append(name)
+
+        # Advance logical slots
+        for database, values in advance_slots.items():
+            with self._get_local_connection_cursor(database) as cur:
+                for name, value in values.items():
+                    try:
+                        cur.execute("SELECT pg_catalog.pg_replication_slot_advance(%s, %s)",
+                                    (name, format_lsn(int(cluster.slots[name]))))
+                    except Exception as e:
+                        logger.exception("Failed to advance logical replication slot '%s': %r", name, e)
+                        self._schedule_load_slots = True
+        return create_slots
+
+    def sync_replication_slots(self, cluster, nofailover, replicatefrom=None):
+        ret = None
         if self._postgresql.major_version >= 90400 and cluster.config:
             try:
                 self.load_replication_slots()
 
-                slots = cluster.get_replication_slots(self._postgresql.name, self._postgresql.role)
+                slots = cluster.get_replication_slots(self._postgresql.name, self._postgresql.role,
+                                                      nofailover, self._postgresql.major_version, True)
 
-                # drop old replication slots which are not presented in desired slots
-                for name in set(self._replication_slots) - set(slots):
-                    if not self.ignore_replication_slot(cluster, name) and not self.drop_replication_slot(name):
-                        logger.error("Failed to drop replication slot '%s'", name)
-                        self._schedule_load_slots = True
+                self._drop_incorrect_slots(cluster, slots)
+
+                self._ensure_physical_slots(slots)
+
+                if self._postgresql.is_leader():
+                    self._unready_logical_slots.clear()
+                    self._ensure_logical_slots_primary(slots)
+                elif cluster.slots and slots:
+                    self.check_logical_slots_readiness(cluster, nofailover, replicatefrom)
+
+                    ret = self._ensure_logical_slots_replica(cluster, slots)
 
-                immediately_reserve = ', true' if self._postgresql.major_version >= 90600 else ''
-
-                logical_slots = defaultdict(dict)
-                for name, value in slots.items():
-                    if name in self._replication_slots and not compare_slots(value, self._replication_slots[name]):
-                        logger.info("Trying to drop replication slot '%s' because value is changing from %s to %s",
-                                    name, self._replication_slots[name], value)
-                        if not self.drop_replication_slot(name):
-                            logger.error("Failed to drop replication slot '%s'", name)
-                            self._schedule_load_slots = True
-                            continue
-                        self._replication_slots.pop(name)
-                    if name not in self._replication_slots:
-                        if value['type'] == 'physical':
-                            try:
-                                self._query(("SELECT pg_catalog.pg_create_physical_replication_slot(%s{0})" +
-                                             " WHERE NOT EXISTS (SELECT 1 FROM pg_catalog.pg_replication_slots" +
-                                             " WHERE slot_type = 'physical' AND slot_name = %s)").format(
-                                                 immediately_reserve), name, name)
-                            except Exception:
-                                logger.exception("Failed to create physical replication slot '%s'", name)
-                                self._schedule_load_slots = True
-                        elif value['type'] == 'logical' and name not in self._replication_slots:
-                            logical_slots[value['database']][name] = value
-
-                # create new logical slots
-                for database, values in logical_slots.items():
-                    conn_kwargs = self._postgresql.config.local_connect_kwargs
-                    conn_kwargs['database'] = database
-                    with get_connection_cursor(**conn_kwargs) as cur:
-                        for name, value in values.items():
-                            try:
-                                cur.execute("SELECT pg_catalog.pg_create_logical_replication_slot(%s, %s)" +
-                                            " WHERE NOT EXISTS (SELECT 1 FROM pg_catalog.pg_replication_slots" +
-                                            " WHERE slot_type = 'logical' AND slot_name = %s)",
-                                            (name, value['plugin'], name))
-                            except Exception:
-                                logger.exception("Failed to create logical replication slot '%s' plugin='%s'",
-                                                 name, value['plugin'])
-                                self._schedule_load_slots = True
                 self._replication_slots = slots
             except Exception:
                 logger.exception('Exception when changing replication slots')
                 self._schedule_load_slots = True
+        return ret
+
+    @contextmanager
+    def _get_leader_connection_cursor(self, leader):
+        conn_kwargs = leader.conn_kwargs(self._postgresql.config.rewind_credentials)
+        conn_kwargs['database'] = self._postgresql.database
+        with get_connection_cursor(connect_timeout=3, options="-c statement_timeout=2000", **conn_kwargs) as cur:
+            yield cur
+
+    def check_logical_slots_readiness(self, cluster, nofailover, replicatefrom):
+        if self._unready_logical_slots:
+            slot_name = cluster.get_my_slot_name_on_primary(self._postgresql.name, replicatefrom)
+            try:
+                with self._get_leader_connection_cursor(cluster.leader) as cur:
+                    cur.execute("SELECT catalog_xmin FROM pg_catalog.pg_get_replication_slots()"
+                                " WHERE NOT pg_catalog.pg_is_in_recovery() AND slot_name = %s", (slot_name,))
+                    if cur.rowcount < 1:
+                        return logger.warning('Physical slot %s does not exist on the primary', slot_name)
+                    catalog_xmin = cur.fetchone()[0]
+            except Exception as e:
+                return logger.error("Failed to check %s physical slot on the primary: %r", slot_name, e)
+        for name in list(self._unready_logical_slots):
+            value = self._replication_slots.get(name)
+            if not value or catalog_xmin <= value['catalog_xmin']:
+                self._unready_logical_slots.remove(name)
+                if value:
+                    logger.info('Logical slot %s is safe to be used after a failover', name)
+
+    def copy_logical_slots(self, leader, slots):
+        with self._get_leader_connection_cursor(leader) as cur:
+            try:
+                cur.execute("SELECT slot_name, catalog_xmin, "
+                            "pg_catalog.pg_wal_lsn_diff(confirmed_flush_lsn, '0/0')::bigint, "
+                            "pg_catalog.pg_read_binary_file('pg_replslot/' || slot_name || '/state')"
+                            " FROM pg_catalog.pg_get_replication_slots() WHERE NOT pg_catalog.pg_is_in_recovery()"
+                            " AND slot_name = ANY(%s)", (slots,))
+                slots = {r[0]: {'catalog_xmin': r[1], 'confirmed_flush_lsn': r[2], 'data': r[3]} for r in cur}
+            except Exception as e:
+                logger.error("Failed to copy logical slots from the %s via postgresql connection: %r", leader.name, e)
+
+        if isinstance(slots, dict) and self._postgresql.stop():
+            pg_replslot_dir = os.path.join(self._postgresql.data_dir, 'pg_replslot')
+            for name, value in slots.items():
+                slot_dir = os.path.join(pg_replslot_dir, name)
+                slot_tmp_dir = slot_dir + '.tmp'
+                if os.path.exists(slot_tmp_dir):
+                    shutil.rmtree(slot_tmp_dir)
+                os.makedirs(slot_tmp_dir)
+                fsync_dir(slot_tmp_dir)
+                with open(os.path.join(slot_tmp_dir, 'state'), 'wb') as f:
+                    f.write(value['data'])
+                    f.flush()
+                    os.fsync(f.fileno())
+                os.rename(slot_tmp_dir, slot_dir)
+                fsync_dir(slot_dir)
+                self._unready_logical_slots.add(name)
+            fsync_dir(pg_replslot_dir)
+            self._postgresql.start()
 
     def schedule(self, value=None):
         if value is None:
             value = self._postgresql.major_version >= 90400
-        self._schedule_load_slots = value
+        self._schedule_load_slots = self._force_readiness_check = value
+
+    def on_promote(self):
+        if self._unready_logical_slots:
+            logger.warning('Logical replication slots that might be unsafe to use after promote: %s',
+                           self._unready_logical_slots)
diff --git a/patroni/utils.py b/patroni/utils.py
index 0bbddb4d..1cfad919 100644
--- a/patroni/utils.py
+++ b/patroni/utils.py
@@ -405,7 +405,7 @@ def is_standby_cluster(config):
 
 def cluster_as_json(cluster):
     leader_name = cluster.leader.name if cluster.leader else None
-    xlog_location_cluster = cluster.last_leader_operation or 0
+    cluster_lsn = cluster.last_lsn or 0
 
     ret = {'members': []}
     for m in cluster.members:
@@ -427,11 +427,11 @@ def cluster_as_json(cluster):
         member.update({n: m.data[n] for n in optional_attributes if n in m.data})
 
         if m.name != leader_name:
-            xlog_location = m.data.get('xlog_location')
-            if xlog_location is None:
+            lsn = m.data.get('xlog_location')
+            if lsn is None:
                 member['lag'] = 'unknown'
-            elif xlog_location_cluster >= xlog_location:
-                member['lag'] = xlog_location_cluster - xlog_location
+            elif cluster_lsn >= lsn:
+                member['lag'] = cluster_lsn - lsn
             else:
                 member['lag'] = 0
 


Test patch:
diff --git a/.github/workflows/run_tests.py b/.github/workflows/run_tests.py
index e2db5e93..fbda3999 100644
--- a/.github/workflows/run_tests.py
+++ b/.github/workflows/run_tests.py
@@ -23,18 +23,19 @@ def main():
 
     env = os.environ.copy()
     if sys.platform.startswith('linux'):
-        version = {'etcd': '9.6', 'etcd3': '9.6', 'consul': 10, 'exhibitor': 11, 'kubernetes': 12, 'raft': 13}.get(what)
+        version = {'etcd': '9.6', 'etcd3': '13', 'consul': 12, 'exhibitor': 11, 'kubernetes': 13, 'raft': 12}.get(what)
         path = '/usr/lib/postgresql/{0}/bin:.'.format(version)
         unbuffer = ['timeout', '600', 'unbuffer']
+        args = ['--tags=-skip'] if what == 'etcd' else []
     else:
         path = os.path.abspath(os.path.join('pgsql', 'bin'))
         if sys.platform == 'darwin':
             path += ':.'
-        unbuffer = []
+        args = unbuffer = []
     env['PATH'] = path + os.pathsep + env['PATH']
     env['DCS'] = what
 
-    ret = subprocess.call(unbuffer + [sys.executable, '-m', 'behave'], env=env)
+    ret = subprocess.call(unbuffer + [sys.executable, '-m', 'behave'] + args, env=env)
 
     if ret != 0:
         if subprocess.call('grep . features/output/*_failed/*postgres?.*', shell=True) != 0:
diff --git a/tests/__init__.py b/tests/__init__.py
index 23861881..4c433862 100644
--- a/tests/__init__.py
+++ b/tests/__init__.py
@@ -90,10 +90,14 @@ class MockCursor(object):
             raise psycopg2.OperationalError()
         elif sql.startswith('RetryFailedError'):
             raise RetryFailedError('retry')
+        elif sql.startswith('SELECT catalog_xmin'):
+            self.results = [(100, 501)]
+        elif sql.startswith('SELECT slot_name, catalog_xmin'):
+            self.results = [('ls', 100, 500, b'123456')]
         elif sql.startswith('SELECT slot_name'):
-            self.results = [('blabla', 'physical'), ('foobar', 'physical'), ('ls', 'logical', 'a', 'b')]
+            self.results = [('blabla', 'physical'), ('foobar', 'physical'), ('ls', 'logical', 'a', 'b', 5, 100, 500)]
         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, [{"slot_name": "ls", "confirmed_flush_lsn": 12345}])]
         elif sql.startswith('SELECT pg_catalog.pg_is_in_recovery()'):
             self.results = [(False, 2)]
         elif sql.startswith('SELECT pg_catalog.pg_postmaster_start_time'):
diff --git a/tests/test_api.py b/tests/test_api.py
index ec91267e..079753f4 100644
--- a/tests/test_api.py
+++ b/tests/test_api.py
@@ -162,7 +162,7 @@ class TestRestApiHandler(unittest.TestCase):
     _authorization = '\nAuthorization: Basic dGVzdDp0ZXN0'
 
     def test_do_GET(self):
-        MockPatroni.dcs.cluster.last_leader_operation = 20
+        MockPatroni.dcs.cluster.last_lsn = 20
         MockRestApiServer(RestApiHandler, 'GET /replica')
         MockRestApiServer(RestApiHandler, 'GET /replica?lag=1M')
         MockRestApiServer(RestApiHandler, 'GET /replica?lag=10MB')
diff --git a/tests/test_consul.py b/tests/test_consul.py
index a38a2869..6b5f773b 100644
--- a/tests/test_consul.py
+++ b/tests/test_consul.py
@@ -15,8 +15,7 @@ def kv_get(self, key, **kwargs):
         return None, None
     if key == 'service/good/leader':
         return '1', None
-    if key == 'service/good/':
-        return ('6429',
+    good_cls = ('6429',
                 [{'CreateIndex': 1334, 'Flags': 0, 'Key': key + 'failover', 'LockIndex': 0,
                   'ModifyIndex': 1334, 'Value': b''},
                  {'CreateIndex': 1334, 'Flags': 0, 'Key': key + 'initialize', 'LockIndex': 0,
@@ -34,7 +33,17 @@ def kv_get(self, key, **kwargs):
                  {'CreateIndex': 1085, 'Flags': 0, 'Key': key + 'optime/leader', 'LockIndex': 0,
                   'ModifyIndex': 6429, 'Value': b'4496294792'},
                  {'CreateIndex': 1085, 'Flags': 0, 'Key': key + 'sync', 'LockIndex': 0,
-                  'ModifyIndex': 6429, 'Value': b'{"leader": "leader", "sync_standby": null}'}])
+                  'ModifyIndex': 6429, 'Value': b'{"leader": "leader", "sync_standby": null}'},
+                 {'CreateIndex': 1085, 'Flags': 0, 'Key': key + 'status', 'LockIndex': 0,
+                  'ModifyIndex': 6429, 'Value': b'{"optime":4496294792, "slots":{"ls":12345}}'}])
+    if key == 'service/good/':
+        return good_cls
+    if key == 'service/broken/':
+        good_cls[1][-1]['Value'] = b'{'
+        return good_cls
+    if key == 'service/legacy/':
+        good_cls[1].pop()
+        return good_cls
     raise ConsulException
 
 
@@ -109,6 +118,10 @@ class TestConsul(unittest.TestCase):
         self.assertIsInstance(self.c.get_cluster(), Cluster)
         self.c._base_path = '/service/fail'
         self.assertRaises(ConsulError, self.c.get_cluster)
+        self.c._base_path = '/service/broken'
+        self.assertIsInstance(self.c.get_cluster(), Cluster)
+        self.c._base_path = '/service/legacy'
+        self.assertIsInstance(self.c.get_cluster(), Cluster)
         self.c._base_path = '/service/good'
         self.c._session = 'fd4f44fe-2cac-bba5-a60b-304b51ff39b8'
         self.assertIsInstance(self.c.get_cluster(), Cluster)
@@ -146,7 +159,7 @@ class TestConsul(unittest.TestCase):
 
     @patch.object(consul.Consul.Session, 'renew', Mock())
     def test_update_leader(self):
-        self.c.update_leader(None)
+        self.c.update_leader(12345)
 
     @patch.object(consul.Consul.KV, 'delete', Mock(return_value=True))
     def test_delete_leader(self):
diff --git a/tests/test_etcd.py b/tests/test_etcd.py
index 9648d4ec..b96082ae 100644
--- a/tests/test_etcd.py
+++ b/tests/test_etcd.py
@@ -66,7 +66,13 @@ def etcd_read(self, key, **kwargs):
                         "?application_name=http://127.0.0.1:8008/patroni",
                      "expiration": "2015-05-15T09:11:09.611860899Z", "ttl": 30,
                      "modifiedIndex": 20730, "createdIndex": 20730}],
-                 "modifiedIndex": 1581, "createdIndex": 1581}], "modifiedIndex": 1581, "createdIndex": 1581}}
+                 "modifiedIndex": 1581, "createdIndex": 1581},
+                {"key": "/service/batman5/status", "value": '{"optime":2164261704,"slots":{"ls":12345}}',
+                 "modifiedIndex": 1582, "createdIndex": 1582}], "modifiedIndex": 1581, "createdIndex": 1581}}
+    if key == '/service/legacy/':
+        response['node']['nodes'].pop()
+    if key == '/service/broken/':
+        response['node']['nodes'][-1]['value'] = '{'
     result = etcd.EtcdResult(**response)
     result.etcd_index = 0
     return result
@@ -246,6 +252,10 @@ class TestEtcd(unittest.TestCase):
         cluster = self.etcd.get_cluster()
         self.assertIsInstance(cluster, Cluster)
         self.assertFalse(cluster.is_synchronous_mode())
+        self.etcd._base_path = '/service/legacy'
+        self.assertIsInstance(self.etcd.get_cluster(), Cluster)
+        self.etcd._base_path = '/service/broken'
+        self.assertIsInstance(self.etcd.get_cluster(), Cluster)
         self.etcd._base_path = '/service/nocluster'
         cluster = self.etcd.get_cluster()
         self.assertIsInstance(cluster, Cluster)
diff --git a/tests/test_etcd3.py b/tests/test_etcd3.py
index 70f19e1d..abc593af 100644
--- a/tests/test_etcd3.py
+++ b/tests/test_etcd3.py
@@ -33,6 +33,8 @@ def mock_urlopen(self, method, url, **kwargs):
                  "value": base64_encode('foo'), "lease": "bla", "mod_revision": '1'},
                 {"key": base64_encode('/patroni/test/members/foo'),
                  "value": base64_encode('{}'), "lease": "123", "mod_revision": '1'},
+                {"key": base64_encode('/patroni/test/members/bar'),
+                 "value": base64_encode('{"version":"1.6.5"}'), "lease": "123", "mod_revision": '1'},
                 {"key": base64_encode('/patroni/test/failover'), "value": base64_encode('{}'), "mod_revision": '1'}
             ]
         })
@@ -172,6 +174,22 @@ class TestEtcd3(BaseTestEtcd3):
         self.assertIsInstance(self.etcd3.get_cluster(), Cluster)
         self.client._kv_cache = None
         with patch.object(urllib3.PoolManager, 'urlopen') as mock_urlopen:
+            mock_urlopen.return_value = MockResponse()
+            mock_urlopen.return_value.content = json.dumps({
+                "header": {"revision": "1"},
+                "kvs": [
+                    {"key": base64_encode('/patroni/test/status'),
+                     "value": base64_encode('{"optime":1234567,"slots":{"ls":12345}}'), "mod_revision": '1'}
+                ]
+            })
+            self.assertIsInstance(self.etcd3.get_cluster(), Cluster)
+            mock_urlopen.return_value.content = json.dumps({
+                "header": {"revision": "1"},
+                "kvs": [
+                    {"key": base64_encode('/patroni/test/status'), "value": base64_encode('{'), "mod_revision": '1'}
+                ]
+            })
+            self.assertIsInstance(self.etcd3.get_cluster(), Cluster)
             mock_urlopen.side_effect = UnsupportedEtcdVersion('')
             self.assertRaises(UnsupportedEtcdVersion, self.etcd3.get_cluster)
             mock_urlopen.side_effect = SleepException()
diff --git a/tests/test_ha.py b/tests/test_ha.py
index b81b7bba..ed534b27 100644
--- a/tests/test_ha.py
+++ b/tests/test_ha.py
@@ -38,7 +38,7 @@ def get_cluster(initialize, leader, members, failover, sync, cluster_config=None
     history = TimelineHistory(1, '[[1,67197376,"no recovery target specified","' + t + '"]]',
                               [(1, 67197376, 'no recovery target specified', t)])
     cluster_config = cluster_config or ClusterConfig(1, {'check_timeline': True}, 1)
-    return Cluster(initialize, cluster_config, leader, 10, members, failover, sync, history)
+    return Cluster(initialize, cluster_config, leader, 10, members, failover, sync, history, None)
 
 
 def get_cluster_not_initialized_without_leader(cluster_config=None):
@@ -66,7 +66,7 @@ def get_cluster_initialized_with_leader(failover=None, sync=None):
 
 def get_cluster_initialized_with_only_leader(failover=None, cluster_config=None):
     leader = get_cluster_initialized_without_leader(leader=True, failover=failover).leader
-    return get_cluster(True, leader, [leader], failover, None, cluster_config)
+    return get_cluster(True, leader, [leader.member], failover, None, cluster_config)
 
 
 def get_standby_cluster_initialized_with_only_leader(failover=None, sync=None):
@@ -152,13 +152,13 @@ def run_async(self, func, args=()):
 @patch.object(Postgresql, 'is_running', Mock(return_value=MockPostmaster()))
 @patch.object(Postgresql, 'is_leader', Mock(return_value=True))
 @patch.object(Postgresql, 'timeline_wal_position', Mock(return_value=(1, 10, 1)))
-@patch.object(Postgresql, '_cluster_info_state_get', Mock(return_value=3))
+@patch.object(Postgresql, '_cluster_info_state_get', Mock(return_value=10))
 @patch.object(Postgresql, 'data_directory_empty', Mock(return_value=False))
 @patch.object(Postgresql, 'controldata', Mock(return_value={
     'Database system identifier': SYSID,
     'Database cluster state': 'shut down',
     'Latest checkpoint location': '0/12345678'}))
-@patch.object(SlotsHandler, 'sync_replication_slots', Mock())
+@patch.object(SlotsHandler, 'load_replication_slots', Mock(side_effect=Exception))
 @patch.object(ConfigHandler, 'append_pg_hba', Mock())
 @patch.object(ConfigHandler, 'write_pgpass', Mock(return_value={}))
 @patch.object(ConfigHandler, 'write_recovery_conf', Mock())
@@ -384,13 +384,21 @@ class TestHa(PostgresInit):
         self.p.is_leader = false
         self.assertEqual(self.ha.run_cycle(), 'not promoting because failed to update leader lock in DCS')
 
+    @patch.object(Postgresql, 'major_version', PropertyMock(return_value=130000))
     def test_follow(self):
         self.ha.cluster.is_unlocked = false
         self.p.is_leader = false
         self.assertEqual(self.ha.run_cycle(), 'no action.  i am a secondary and i am following a leader')
         self.ha.patroni.replicatefrom = "foo"
         self.p.config.check_recovery_conf = Mock(return_value=(True, False))
+        self.ha.cluster.config.data.update({'slots': {'l': {'database': 'a', 'plugin': 'b'}}})
+        self.ha.cluster.members[1].data['tags']['replicatefrom'] = 'postgresql0'
+        self.ha.patroni.nofailover = True
+        self.assertEqual(self.ha.run_cycle(), 'no action.  i am a secondary and i am following a leader')
+        del self.ha.cluster.config.data['slots']
+        self.ha.cluster.config.data.update({'postgresql': {'use_slots': False}})
         self.assertEqual(self.ha.run_cycle(), 'no action.  i am a secondary and i am following a leader')
+        del self.ha.cluster.config.data['postgresql']['use_slots']
 
     def test_follow_in_pause(self):
         self.ha.cluster.is_unlocked = false
@@ -634,7 +642,7 @@ class TestHa(PostgresInit):
         # in synchronous_mode consider itself healthy if the former leader is accessible in read-only and ahead of us
         with patch.object(Ha, 'is_synchronous_mode', Mock(return_value=True)):
             self.assertTrue(self.ha._is_healthiest_node(self.ha.old_cluster.members))
-        with patch('patroni.postgresql.Postgresql.timeline_wal_position', return_value=(1, 1, 1)):
+        with patch('patroni.postgresql.Postgresql.last_operation', return_value=1):
             self.assertFalse(self.ha._is_healthiest_node(self.ha.old_cluster.members))
         with patch('patroni.postgresql.Postgresql.replica_cached_timeline', return_value=1):
             self.assertFalse(self.ha._is_healthiest_node(self.ha.old_cluster.members))
@@ -1110,6 +1118,7 @@ class TestHa(PostgresInit):
     @patch('psycopg2.connect', psycopg2_connect)
     def test_permanent_logical_slots_after_promote(self):
         config = ClusterConfig(1, {'slots': {'l': {'database': 'postgres', 'plugin': 'test_decoding'}}}, 1)
+        self.p.name = 'other'
         self.ha.cluster = get_cluster_initialized_without_leader(cluster_config=config)
         self.assertEqual(self.ha.run_cycle(), 'acquired session lock as a leader')
         self.ha.cluster = get_cluster_initialized_without_leader(leader=True, cluster_config=config)
@@ -1137,3 +1146,19 @@ class TestHa(PostgresInit):
 
         self.ha.has_lock = true
         self.assertEqual(self.ha.run_cycle(), 'PAUSE: released leader key voluntarily due to the system ID mismatch')
+
+    @patch('psycopg2.connect', psycopg2_connect)
+    @patch('os.path.exists', Mock(return_value=True))
+    @patch('shutil.rmtree', Mock())
+    @patch('os.makedirs', Mock())
+    @patch('os.open', Mock())
+    @patch('os.fsync', Mock())
+    @patch('os.close', Mock())
+    @patch('os.rename', Mock())
+    @patch('patroni.postgresql.Postgresql.is_starting', Mock(return_value=False))
+    @patch.object(builtins, 'open', mock_open())
+    @patch.object(SlotsHandler, 'sync_replication_slots', Mock(return_value=['foo']))
+    def test_follow_copy(self):
+        self.ha.cluster.is_unlocked = false
+        self.p.is_leader = false
+        self.assertTrue(self.ha.run_cycle().startswith('Copying logical slots'))
diff --git a/tests/test_kubernetes.py b/tests/test_kubernetes.py
index aa4493a9..92339b5a 100644
--- a/tests/test_kubernetes.py
+++ b/tests/test_kubernetes.py
@@ -16,7 +16,8 @@ def mock_list_namespaced_config_map(*args, **kwargs):
     metadata = {'resource_version': '1', 'labels': {'f': 'b'}, 'name': 'test-config',
                 'annotations': {'initialize': '123', 'config': '{}'}}
     items = [k8s_client.V1ConfigMap(metadata=k8s_client.V1ObjectMeta(**metadata))]
-    metadata.update({'name': 'test-leader', 'annotations': {'optime': '1234', 'leader': 'p-0', 'ttl': '30s'}})
+    metadata.update({'name': 'test-leader',
+                     'annotations': {'optime': '1234x', 'leader': 'p-0', 'ttl': '30s', 'slots': '{'}})
     items.append(k8s_client.V1ConfigMap(metadata=k8s_client.V1ObjectMeta(**metadata)))
     metadata.update({'name': 'test-failover', 'annotations': {'leader': 'p-0'}})
     items.append(k8s_client.V1ConfigMap(metadata=k8s_client.V1ObjectMeta(**metadata)))
@@ -260,10 +261,6 @@ class TestKubernetesEndpoints(BaseTestKubernetes):
         self.k._kinds._object_cache['test'].metadata.annotations['leader'] = 'p-1'
         self.assertFalse(self.k.update_leader('123'))
 
-    @patch.object(k8s_client.CoreV1Api, 'patch_namespaced_endpoints', mock_namespaced_kind, create=True)
-    def test_update_leader_with_restricted_access(self):
-        self.assertIsNotNone(self.k.update_leader('123', True))
-
     @patch.object(k8s_client.CoreV1Api, 'read_namespaced_endpoints', create=True)
     @patch.object(k8s_client.CoreV1Api, 'patch_namespaced_endpoints', create=True)
     def test__update_leader_with_retry(self, mock_patch, mock_read):
diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py
index 3a3937d4..479554fe 100644
--- a/tests/test_postgresql.py
+++ b/tests/test_postgresql.py
@@ -1,5 +1,4 @@
 import datetime
-import mock  # for the mock.call method, importing it without a namespace breaks python3
 import os
 import psutil
 import psycopg2
@@ -9,11 +8,10 @@ import time
 
 from mock import Mock, MagicMock, PropertyMock, patch, mock_open
 from patroni.async_executor import CriticalTask
-from patroni.dcs import Cluster, ClusterConfig, Member, RemoteMember, SyncState
+from patroni.dcs import Cluster, RemoteMember, SyncState
 from patroni.exceptions import PostgresConnectionException, PatroniException
 from patroni.postgresql import Postgresql, STATE_REJECT, STATE_NO_RESPONSE
 from patroni.postgresql.postmaster import PostmasterProcess
-from patroni.postgresql.slots import SlotsHandler
 from patroni.utils import RetryFailedError
 from six.moves import builtins
 from threading import Thread, current_thread
@@ -303,29 +301,6 @@ class TestPostgresql(BaseTestPostgresql):
         m = RemoteMember('1', {'restore_command': '2', 'primary_slot_name': 'foo', 'conn_kwargs': {'host': 'bar'}})
         self.p.follow(m)
 
-    @patch.object(Postgresql, 'is_running', Mock(return_value=True))
-    def test_sync_replication_slots(self):
-        self.p.start()
-        config = ClusterConfig(1, {'slots': {'test_3': {'database': 'a', 'plugin': 'b'},
-                                             'A': 0, 'ls': 0, 'b': {'type': 'logical', 'plugin': '1'}},
-                                   'ignore_slots': [{'name': 'blabla'}]}, 1)
-        cluster = Cluster(True, config, self.leader, 0, [self.me, self.other, self.leadermem], None, None, None)
-        with mock.patch('patroni.postgresql.Postgresql._query', Mock(side_effect=psycopg2.OperationalError)):
-            self.p.slots_handler.sync_replication_slots(cluster)
-        self.p.slots_handler.sync_replication_slots(cluster)
-        with mock.patch('patroni.postgresql.Postgresql.role', new_callable=PropertyMock(return_value='replica')):
-            self.p.slots_handler.sync_replication_slots(cluster)
-        with patch.object(SlotsHandler, 'drop_replication_slot', Mock(return_value=True)),\
-                patch('patroni.dcs.logger.error', new_callable=Mock()) as errorlog_mock:
-            alias1 = Member(0, 'test-3', 28, {'conn_url': 'postgres://replicator:rep-pass@127.0.0.1:5436/postgres'})
-            alias2 = Member(0, 'test.3', 28, {'conn_url': 'postgres://replicator:rep-pass@127.0.0.1:5436/postgres'})
-            cluster.members.extend([alias1, alias2])
-            self.p.slots_handler.sync_replication_slots(cluster)
-            self.assertEqual(errorlog_mock.call_count, 5)
-            ca = errorlog_mock.call_args_list[0][0][1]
-            self.assertTrue("test-3" in ca, "non matching {0}".format(ca))
-            self.assertTrue("test.3" in ca, "non matching {0}".format(ca))
-
     @patch.object(MockCursor, 'execute', Mock(side_effect=psycopg2.OperationalError))
     def test__query(self):
         self.assertRaises(PostgresConnectionException, self.p._query, 'blabla')
@@ -340,7 +315,7 @@ class TestPostgresql(BaseTestPostgresql):
     @patch.object(Postgresql, 'pg_isready', Mock(return_value=STATE_REJECT))
     def test_is_leader(self):
         self.assertTrue(self.p.is_leader())
-        self.p.reset_cluster_info_state()
+        self.p.reset_cluster_info_state(None)
         with patch.object(Postgresql, '_query', Mock(side_effect=RetryFailedError(''))):
             self.assertRaises(PostgresConnectionException, self.p.is_leader)
 
@@ -611,7 +586,7 @@ class TestPostgresql(BaseTestPostgresql):
 
     def test_pick_sync_standby(self):
         cluster = Cluster(True, None, self.leader, 0, [self.me, self.other, self.leadermem], None,
-                          SyncState(0, self.me.name, self.leadermem.name), None)
+                          SyncState(0, self.me.name, self.leadermem.name), None, None)
         mock_cursor = Mock()
         mock_cursor.fetchone.return_value = ('remote_apply',)
 
@@ -727,10 +702,14 @@ class TestPostgresql(BaseTestPostgresql):
     @patch.object(Postgresql, '_query', Mock(side_effect=RetryFailedError('')))
     def test_received_timeline(self):
         self.p.set_role('standby_leader')
-        self.p.reset_cluster_info_state()
+        self.p.reset_cluster_info_state(None)
         self.assertRaises(PostgresConnectionException, self.p.received_timeline)
 
     def test__write_recovery_params(self):
         self.p.config._write_recovery_params(Mock(), {'pause_at_recovery_target': 'false'})
         with patch.object(Postgresql, 'major_version', PropertyMock(return_value=90400)):
             self.p.config._write_recovery_params(Mock(), {'recovery_target_action': 'PROMOTE'})
+
+    @patch.object(Postgresql, 'is_running', Mock(return_value=True))
+    def test_set_enforce_hot_standby_feedback(self):
+        self.p.set_enforce_hot_standby_feedback(True)
diff --git a/tests/test_raft.py b/tests/test_raft.py
index 4b51ebe0..54197df8 100644
--- a/tests/test_raft.py
+++ b/tests/test_raft.py
@@ -129,14 +129,19 @@ class TestRaft(unittest.TestCase):
                      'retry_timeout': 10, 'data_dir': self._TMP})
         raft.set_retry_timeout(20)
         raft.set_ttl(60)
+        self.assertTrue(raft._sync_obj.set(raft.members_path + 'legacy', '{"version":"2.0.0"}'))
         self.assertTrue(raft.touch_member(''))
         self.assertTrue(raft.initialize())
         self.assertTrue(raft.cancel_initialization())
         self.assertTrue(raft.set_config_value('{}'))
         self.assertTrue(raft.write_sync_state('foo', 'bar'))
-        self.assertTrue(raft.update_leader('1'))
         self.assertTrue(raft.manual_failover('foo', 'bar'))
         raft.get_cluster()
+        self.assertTrue(raft._sync_obj.set(raft.status_path, '{"optime":1234567,"slots":{"ls":12345}}'))
+        raft.get_cluster()
+        self.assertTrue(raft.update_leader('1'))
+        self.assertTrue(raft._sync_obj.set(raft.status_path, '{'))
+        raft.get_cluster()
         self.assertTrue(raft.delete_sync_state())
         self.assertTrue(raft.delete_leader())
         self.assertTrue(raft.set_history_value(''))
diff --git a/tests/test_rewind.py b/tests/test_rewind.py
index cb14ce9d..c2607e3f 100644
--- a/tests/test_rewind.py
+++ b/tests/test_rewind.py
@@ -93,7 +93,7 @@ class TestRewind(BaseTestPostgresql):
             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,), Exception])):
+            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)
 
     @patch.object(CancellableSubprocess, 'call', mock_cancellable_call)
diff --git a/tests/test_slots.py b/tests/test_slots.py
new file mode 100644
index 00000000..e17c4779
--- /dev/null
+++ b/tests/test_slots.py
@@ -0,0 +1,124 @@
+import mock
+import os
+import psycopg2
+import unittest
+
+
+from mock import Mock, PropertyMock, patch
+
+from patroni.dcs import Cluster, ClusterConfig, Member
+from patroni.postgresql import Postgresql
+from patroni.postgresql.slots import SlotsHandler, fsync_dir
+
+from . import BaseTestPostgresql, psycopg2_connect, MockCursor
+
+
+@patch('subprocess.call', Mock(return_value=0))
+@patch('psycopg2.connect', psycopg2_connect)
+@patch.object(Postgresql, 'is_running', Mock(return_value=True))
+class TestSlotsHandler(BaseTestPostgresql):
+
+    @patch('subprocess.call', Mock(return_value=0))
+    @patch('os.rename', Mock())
+    @patch('patroni.postgresql.CallbackExecutor', Mock())
+    @patch.object(Postgresql, 'get_major_version', Mock(return_value=130000))
+    @patch.object(Postgresql, 'is_running', Mock(return_value=True))
+    def setUp(self):
+        super(TestSlotsHandler, self).setUp()
+        self.s = self.p.slots_handler
+        self.p.start()
+
+    def test_sync_replication_slots(self):
+        config = ClusterConfig(1, {'slots': {'test_3': {'database': 'a', 'plugin': 'b'},
+                                             'A': 0, 'ls': 0, 'b': {'type': 'logical', 'plugin': '1'}},
+                                   'ignore_slots': [{'name': 'blabla'}]}, 1)
+        cluster = Cluster(True, config, self.leader, 0,
+                          [self.me, self.other, self.leadermem], None, None, None, {'test_3': 10})
+        with mock.patch('patroni.postgresql.Postgresql._query', Mock(side_effect=psycopg2.OperationalError)):
+            self.s.sync_replication_slots(cluster, False)
+        self.p.set_role('standby_leader')
+        self.s.sync_replication_slots(cluster, False)
+        self.p.set_role('replica')
+        with patch.object(Postgresql, 'is_leader', Mock(return_value=False)):
+            self.s.sync_replication_slots(cluster, False)
+        self.p.set_role('master')
+        with mock.patch('patroni.postgresql.Postgresql.role', new_callable=PropertyMock(return_value='replica')):
+            self.s.sync_replication_slots(cluster, False)
+        with patch.object(SlotsHandler, 'drop_replication_slot', Mock(return_value=True)),\
+                patch('patroni.dcs.logger.error', new_callable=Mock()) as errorlog_mock:
+            alias1 = Member(0, 'test-3', 28, {'conn_url': 'postgres://replicator:rep-pass@127.0.0.1:5436/postgres'})
+            alias2 = Member(0, 'test.3', 28, {'conn_url': 'postgres://replicator:rep-pass@127.0.0.1:5436/postgres'})
+            cluster.members.extend([alias1, alias2])
+            self.s.sync_replication_slots(cluster, False)
+            self.assertEqual(errorlog_mock.call_count, 5)
+            ca = errorlog_mock.call_args_list[0][0][1]
+            self.assertTrue("test-3" in ca, "non matching {0}".format(ca))
+            self.assertTrue("test.3" in ca, "non matching {0}".format(ca))
+            with patch.object(Postgresql, 'major_version', PropertyMock(return_value=90618)):
+                self.s.sync_replication_slots(cluster, False)
+
+    def test_process_permanent_slots(self):
+        config = ClusterConfig(1, {'slots': {'ls': {'database': 'a', 'plugin': 'b'}},
+                                   'ignore_slots': [{'name': 'blabla'}]}, 1)
+        cluster = Cluster(True, config, self.leader, 0, [self.me, self.other, self.leadermem], None, None, None, None)
+
+        self.s.sync_replication_slots(cluster, False)
+        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,
+                [{"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,
+                [{"slot_name": "ls", "type": "logical", "datoid": 6, "plugin": "b",
+                  "confirmed_flush_lsn": 12345, "catalog_xmin": 105}])
+            self.assertEqual(self.p.slots(), {})
+
+    @patch.object(Postgresql, 'is_leader', Mock(return_value=False))
+    def test__ensure_logical_slots_replica(self):
+        self.p.set_role('replica')
+        config = ClusterConfig(1, {'slots': {'ls': {'database': 'a', 'plugin': 'b'}}}, 1)
+        cluster = Cluster(True, config, self.leader, 0,
+                          [self.me, self.other, self.leadermem], None, None, None, {'ls': 12346})
+        self.assertEqual(self.s.sync_replication_slots(cluster, False), [])
+        self.s._schedule_load_slots = False
+        with patch.object(MockCursor, 'execute', Mock(side_effect=psycopg2.OperationalError)):
+            self.assertEqual(self.s.sync_replication_slots(cluster, False), [])
+        cluster.slots['ls'] = 'a'
+        self.assertEqual(self.s.sync_replication_slots(cluster, False), [])
+        with patch.object(MockCursor, 'rowcount', PropertyMock(return_value=1), create=True):
+            self.assertEqual(self.s.sync_replication_slots(cluster, False), ['ls'])
+
+    @patch.object(MockCursor, 'execute', Mock(side_effect=psycopg2.OperationalError))
+    def test_copy_logical_slots(self):
+        self.s.copy_logical_slots(self.leader, ['foo'])
+
+    @patch.object(Postgresql, 'stop', Mock(return_value=True))
+    @patch.object(Postgresql, 'start', Mock(return_value=True))
+    @patch.object(Postgresql, 'is_leader', Mock(return_value=False))
+    def test_check_logical_slots_readiness(self):
+        self.s.copy_logical_slots(self.leader, ['ls'])
+        config = ClusterConfig(1, {'slots': {'ls': {'database': 'a', 'plugin': 'b'}}}, 1)
+        cluster = Cluster(True, config, self.leader, 0,
+                          [self.me, self.other, self.leadermem], None, None, None, {'ls': 12345})
+        self.assertEqual(self.s.sync_replication_slots(cluster, False), [])
+        with patch.object(MockCursor, 'rowcount', PropertyMock(return_value=1), create=True):
+            self.s.check_logical_slots_readiness(cluster, False, None)
+
+    @patch.object(Postgresql, 'stop', Mock(return_value=True))
+    @patch.object(Postgresql, 'start', Mock(return_value=True))
+    @patch.object(Postgresql, 'is_leader', Mock(return_value=False))
+    def test_on_promote(self):
+        self.s.copy_logical_slots(self.leader, ['ls'])
+        self.s.on_promote()
+
+    @unittest.skipIf(os.name == 'nt', "Windows not supported")
+    @patch('os.open', Mock())
+    @patch('os.close', Mock())
+    @patch('os.fsync', Mock(side_effect=OSError))
+    def test_fsync_dir(self):
+        self.assertRaises(OSError, fsync_dir, 'foo')
diff --git a/tests/test_zookeeper.py b/tests/test_zookeeper.py
index 8d00894a..ef368826 100644
--- a/tests/test_zookeeper.py
+++ b/tests/test_zookeeper.py
@@ -30,7 +30,9 @@ class MockKazooClient(Mock):
     def get(self, path, watch=None):
         if not isinstance(path, six.string_types):
             raise TypeError("Invalid type for 'path' (string expected)")
-        if path == '/no_node':
+        if path == '/broken/status':
+            return (b'{', ZnodeStat(0, 0, 0, 0, 0, 0, 0, -1, 0, 0, 0))
+        elif path in ('/no_node', '/legacy/status'):
             raise NoNodeError
         elif '/members/' in path:
             return (
@@ -45,6 +47,8 @@ class MockKazooClient(Mock):
             return (b'foo', ZnodeStat(0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0))
         elif path.endswith('/initialize'):
             return (b'foo', ZnodeStat(0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0))
+        elif path.endswith('/status'):
+            return (b'{"optime":500,"slots":{"ls":1234567}}', ZnodeStat(0, 0, 0, 0, 0, 0, 0, -1, 0, 0, 0))
         return (b'', ZnodeStat(0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0))
 
     @staticmethod
@@ -147,6 +151,10 @@ class TestZooKeeper(unittest.TestCase):
     def test__inner_load_cluster(self):
         self.zk._base_path = self.zk._base_path.replace('test', 'bla')
         self.zk._inner_load_cluster()
+        self.zk._base_path = self.zk._base_path = '/broken'
+        self.zk._inner_load_cluster()
+        self.zk._base_path = self.zk._base_path = '/legacy'
+        self.zk._inner_load_cluster()
         self.zk._base_path = self.zk._base_path = '/no_node'
         self.zk._inner_load_cluster()
 
@@ -156,11 +164,11 @@ class TestZooKeeper(unittest.TestCase):
         self.assertIsInstance(cluster.leader, Leader)
         self.zk.touch_member({'foo': 'foo'})
         self.zk._name = 'bar'
-        self.zk.optime_watcher(None)
+        self.zk.status_watcher(None)
         with patch.object(ZooKeeper, 'get_node', Mock(side_effect=Exception)):
             self.zk.get_cluster()
         cluster = self.zk.get_cluster()
-        self.assertEqual(cluster.last_leader_operation, 500)
+        self.assertEqual(cluster.last_lsn, 500)
 
     def test_delete_leader(self):
         self.assertTrue(self.zk.delete_leader())
@@ -203,10 +211,10 @@ class TestZooKeeper(unittest.TestCase):
             self.zk.take_leader()
 
     def test_update_leader(self):
-        self.assertTrue(self.zk.update_leader(None))
+        self.assertTrue(self.zk.update_leader(12345))
 
     def test_write_leader_optime(self):
-        self.zk.last_leader_operation = '0'
+        self.zk.last_lsn = '0'
         self.zk.write_leader_optime('1')
         with patch.object(MockKazooClient, 'create_async', Mock()):
             self.zk.write_leader_optime('1')
@@ -221,7 +229,7 @@ class TestZooKeeper(unittest.TestCase):
     def test_watch(self):
         self.zk.watch(None, 0)
         self.zk.event.isSet = Mock(return_value=True)
-        self.zk._fetch_optime = False
+        self.zk._fetch_status = False
         self.zk.watch(None, 0)
 
     def test__kazoo_connect(self):