Repository: zalando/patroni
Pull request #2852 (2023-09-07 13:32:12)

Permanent physical slots on standby nodes
Create permanent physical replication slots on standby nodes and use `pg_replication_slot_advance()` function to move them forward.

The `restart_lsn` is advanced based on values stored in the `/status` key by the primary node.

When slot is created on a replica it could be ahead the same slot on the primary and therefore there is some period of time when it doesn't protect WAL files from being recycled.

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

* **48** of **48**   **(100.0%)**  changed or added relevant lines in **3** files are covered.
* No unchanged relevant lines lost coverage.
* Overall coverage remained the same at **99.863%**

---



|  Totals | [![Coverage Status](https://coveralls.io/builds/62556644/badge)](https://coveralls.io/builds/62556644) |
| :-- | --: |
| Change from base [Build 6108812767](https://coveralls.io/builds/62496621): |  0.0% |
| Covered Lines: | 13100 |
| Relevant Lines: | 13118 |

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

hughcapet: 👍 
CyberDem0n: > it won't advance the LSN of a slot which matches the name of the current primary because the slot doesn't exist in the primary:

Nice catch! Fixed in 852205e6856f302d2519cfc2125ab04d39dd6822.
hughcapet: 👍 

Patch:
diff --git a/docs/conf.py b/docs/conf.py
index f94b9660..950fbc91 100644
--- a/docs/conf.py
+++ b/docs/conf.py
@@ -112,10 +112,10 @@ todo_include_todos = True
 # a list of builtin themes.
 #
 
+html_theme = 'sphinx_rtd_theme'
 on_rtd = os.environ.get('READTHEDOCS', None) == 'True'
 if not on_rtd:  # only import and set the theme if we're building docs locally
     import sphinx_rtd_theme
-    html_theme = 'sphinx_rtd_theme'
     html_theme_path = [sphinx_rtd_theme.get_html_theme_path()]
 
 # Theme options are theme-specific and customize the look and feel of a theme
diff --git a/docs/contributing_guidelines.rst b/docs/contributing_guidelines.rst
index 48710868..05d98987 100644
--- a/docs/contributing_guidelines.rst
+++ b/docs/contributing_guidelines.rst
@@ -143,12 +143,12 @@ If you want to disable this facility set the env var `OPEN_CMD` to the `:` no-op
 Behave tests
 ^^^^^^^^^^^^
 
-Behave tests with `-m behave` will build docker images based on PG_MAJOR version 11 through 15 and then run all
+Behave tests with `-m behave` will build docker images based on PG_MAJOR version 11 through 16 and then run all
 behave tests. This can take quite a long time to run so you might want to limit the scope to a select version of
 Postgres or to a specific feature set or steps.
 
 To specify the version of postgres include the full name of the dependent image build env that you want and then the
-behave env name. For instance if you want Postgres 15 use:
+behave env name. For instance if you want Postgres 14 use:
 
 .. code-block:: bash
 
diff --git a/docs/dynamic_configuration.rst b/docs/dynamic_configuration.rst
index f5bb4ee5..7f04ce33 100644
--- a/docs/dynamic_configuration.rst
+++ b/docs/dynamic_configuration.rst
@@ -46,7 +46,7 @@ In order to change the dynamic configuration you can use either ``patronictl edi
    -  **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. Permanent slots that don't exist will be created by Patroni. The physical slots are maintained only in the current primary. 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+.
+-  **slots**: define permanent replication slots. These slots will be preserved during switchover/failover. Permanent slots that don't exist will be created by Patroni. With PostgreSQL 11 onwards permanent physical slots are created on all nodes and their position is advanced every **loop_wait** seconds. For PostgreSQL versions older than 11 permanent physical replication slots are maintained only on the current primary. 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 replication slots requires **postgresql.use_slots** to be set to ``true``. If there are permanent logical replication slots defined Patroni will 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 the permanent replication slot. If the permanent slot name matches with the name of the current leader it will not be created. Please note that Patroni does not make checks for permanent slot names added to this configuration matching those that Patroni creates automatically for members. If those names are added, Patroni will ensure that any slots that were created are not removed even if the member becomes unresponsive, situation which would normally result in the slot's removal by Patroni. Although this can be useful in some situations, such as when importing existing members to a new Patroni cluster (see :ref:`Convert a Standalone to a Patroni Cluster <existing_data>` for details), caution should be exercised by the operator that these clashes in names are not persisted in the DCS due to its effect on normal functioning of Patroni.
 
@@ -80,4 +80,17 @@ Note: **slots** is a hashmap while **ignore_slots** is an array. For example:
             plugin: test_decoding
           - name: ignored_physical_slot_name
             type: physical
-          ...
\ No newline at end of file
+          ...
+
+Note: if cluster topology is static (fixed number of nodes that never change their names) you can configure permanent physical replication slots with names corresponding to names of nodes to avoid recycling of WAL files while replica is temporary down:
+
+.. code:: YAML
+
+        slots:
+          node_name1:
+            type: physical
+          node_name2:
+            type: physical
+          node_name3:
+            type: physical
+          ...
diff --git a/docs/releases.rst b/docs/releases.rst
index 7a05bf7e..5d4af2cc 100644
--- a/docs/releases.rst
+++ b/docs/releases.rst
@@ -3,6 +3,64 @@
 Release notes
 =============
 
+Version 3.1.1
+-------------
+
+**Bugfixes**
+
+- Reset failsafe state on promote (ChenChangAo)
+
+  If switchover/failover happened shortly after failsafe mode had been activated, the newly promoted primary was demoting itself after failsafe becomes inactive.
+
+- Silence useless warnings in ``patronictl`` (Alexander Kukushkin)
+
+  If ``patronictl`` uses the same patroni.yaml file as Patroni and can access ``PGDATA`` directory it might have been showing annoying warnings about incorrect values in the global configuration.
+
+- Explicitly enable synchronous mode for a corner case (Alexander Kukushkin)
+
+  Synchronous mode effectively was never activated if there are no replicas streaming from the primary.
+
+- Fixed bug with ``0`` integer values validation (Israel Barth Rubio)
+
+  In most cases, it didn't cause any issues, just warnings.
+
+- Don't return logical slots for standby cluster (Alexander Kukushkin)
+
+  Patroni can't create logical replication slots in the standby cluster, thus they should be ignored if they are defined in the global configuration.
+
+- Avoid showing docstring in ``patronictl --help`` output (Israel Barth Rubio)
+
+  The ``click`` module needs to get a special hint for that.
+
+- Fixed bug with ``kubernetes.standby_leader_label_value`` (Alexander Kukushkin)
+
+  This feature effectively never worked.
+
+- Returned cluster system identifier to the ``patronictl list`` output (Polina Bungina)
+
+  The problem was introduced while implementing the support for Citus, where we need to hide the identifier because it is different for coordinator and all workers.
+
+- Override ``write_leader_optime`` method in Kubernetes implementation (Alexander Kukushkin)
+
+  The method is supposed to write shutdown LSN to the leader Endpoint/ConfigMap when there are no healthy replicas available to become the new primary.
+
+- Don't start stopped postgres in pause (Alexander Kukushkin)
+
+  Due to a race condition, Patroni was falsely assuming that the standby should be restarted because some recovery parameters (``primary_conninfo`` or similar) were changed.
+
+- Fixed bug in ``patronictl query`` command (Israel Barth Rubio)
+
+  It didn't work when only ``-m`` argument was provided or when none of ``-r`` or ``-m`` were provided.
+
+- Properly treat integer parameters that are used in the command line to start postgres (Polina Bungina)
+
+  If values are supplied as strings and not casted to integer it was resulting in an incorrect calculation of ``max_prepared_transactions`` based on ``max_connections`` for Citus clusters.
+
+- Don't rely on ``pg_stat_wal_receiver`` when deciding on ``pg_rewind`` (Alexander Kukushkin)
+
+  It could happen that ``received_tli`` reported by ``pg_stat_wal_recevier`` is ahead of the actual replayed timeline, while the timeline reported by ``DENTIFY_SYSTEM`` via replication connection is always correct.
+
+
 Version 3.1.0
 -------------
 
diff --git a/features/ignored_slots.feature b/features/ignored_slots.feature
index 08a6dda5..e0c53ea3 100644
--- a/features/ignored_slots.feature
+++ b/features/ignored_slots.feature
@@ -25,10 +25,10 @@ Feature: ignored slots
     # but Patroni can actually end up dropping them almost immediately, so it's helpful
     # to verify they exist before we begin testing whether they persist through failover
     # cycles.
-    Then postgres1 has a logical replication slot named unmanaged_slot_0 with the test_decoding plugin
-    And postgres1 has a logical replication slot named unmanaged_slot_1 with the test_decoding plugin
-    And postgres1 has a logical replication slot named unmanaged_slot_2 with the test_decoding plugin
-    And postgres1 has a logical replication slot named unmanaged_slot_3 with the test_decoding plugin
+    Then postgres1 has a logical replication slot named unmanaged_slot_0 with the test_decoding plugin after 2 seconds
+    And postgres1 has a logical replication slot named unmanaged_slot_1 with the test_decoding plugin after 2 seconds
+    And postgres1 has a logical replication slot named unmanaged_slot_2 with the test_decoding plugin after 2 seconds
+    And postgres1 has a logical replication slot named unmanaged_slot_3 with the test_decoding plugin after 2 seconds
 
     When I start postgres0
     Then "members/postgres0" key in DCS has role=replica after 10 seconds
@@ -46,16 +46,16 @@ Feature: ignored slots
     And "members/postgres1" key in DCS has role=replica after 10 seconds
     # give Patroni time to sync replication slots
     And I sleep for 2 seconds
-    And postgres1 has a logical replication slot named unmanaged_slot_0 with the test_decoding plugin
-    And postgres1 has a logical replication slot named unmanaged_slot_1 with the test_decoding plugin
-    And postgres1 has a logical replication slot named unmanaged_slot_2 with the test_decoding plugin
-    And postgres1 has a logical replication slot named unmanaged_slot_3 with the test_decoding plugin
+    And postgres1 has a logical replication slot named unmanaged_slot_0 with the test_decoding plugin after 2 seconds
+    And postgres1 has a logical replication slot named unmanaged_slot_1 with the test_decoding plugin after 2 seconds
+    And postgres1 has a logical replication slot named unmanaged_slot_2 with the test_decoding plugin after 2 seconds
+    And postgres1 has a logical replication slot named unmanaged_slot_3 with the test_decoding plugin after 2 seconds
     And postgres1 does not have a logical replication slot named dummy_slot
 
     # 3. After a failover the server (now a primary) still has the slot.
     When I shut down postgres0
     Then "members/postgres1" key in DCS has role=master after 10 seconds
-    And postgres1 has a logical replication slot named unmanaged_slot_0 with the test_decoding plugin
-    And postgres1 has a logical replication slot named unmanaged_slot_1 with the test_decoding plugin
-    And postgres1 has a logical replication slot named unmanaged_slot_2 with the test_decoding plugin
-    And postgres1 has a logical replication slot named unmanaged_slot_3 with the test_decoding plugin
+    And postgres1 has a logical replication slot named unmanaged_slot_0 with the test_decoding plugin after 2 seconds
+    And postgres1 has a logical replication slot named unmanaged_slot_1 with the test_decoding plugin after 2 seconds
+    And postgres1 has a logical replication slot named unmanaged_slot_2 with the test_decoding plugin after 2 seconds
+    And postgres1 has a logical replication slot named unmanaged_slot_3 with the test_decoding plugin after 2 seconds
diff --git a/features/permanent_slots.feature b/features/permanent_slots.feature
new file mode 100644
index 00000000..656e6ade
--- /dev/null
+++ b/features/permanent_slots.feature
@@ -0,0 +1,34 @@
+Feature: permanent slots
+  Scenario: check that physical permanent slots are created
+    Given I start postgres0
+    Then postgres0 is a leader after 10 seconds
+    And there is a non empty initialize key in DCS after 15 seconds
+    When I issue a PATCH request to http://127.0.0.1:8008/config with {"slots": {"test_physical": {"type": "physical"}}, "postgresql": {"parameters": {"wal_level": "logical"}}}
+    Then I receive a response code 200
+    And Response on GET http://127.0.0.1:8008/config contains slots after 10 seconds
+    Then postgres0 has a physical replication slot named test_physical after 10 seconds
+    And I start postgres1
+
+  @slot-advance
+  Scenario: check that logical permanent slots are created
+    Given I run patronictl.py restart batman postgres0 --force
+    And I issue a PATCH request to http://127.0.0.1:8008/config with {"slots": {"test_logical": {"type": "logical", "database": "postgres", "plugin": "test_decoding"}}}
+    Then postgres0 has a logical replication slot named test_logical with the test_decoding plugin after 10 seconds
+
+  @slot-advance
+  Scenario: check that permanent slots are created on the replica
+    Given postgres1 has a logical replication slot named test_logical with the test_decoding plugin after 10 seconds
+    Then Logical slot test_logical is in sync between postgres0 and postgres1 after 10 seconds
+    And postgres1 has a physical replication slot named test_physical after 2 seconds
+
+  @slot-advance
+  Scenario: check that permanent slots are advanced on the replica
+    Given I add the table replicate_me to postgres0
+    And I get all changes from physical slot test_physical on postgres0
+    When I get all changes from logical slot test_logical on postgres0
+    Then Logical slot test_logical is in sync between postgres0 and postgres1 after 10 seconds
+    And Physical slot test_physical is in sync between postgres0 and postgres1 after 10 seconds
+
+  Scenario: check permanent physical replication slot after failover
+    Given I shut down postgres0
+    Then postgres1 has a physical replication slot named test_physical after 10 seconds
diff --git a/features/standby_cluster.feature b/features/standby_cluster.feature
index 850c7970..7ff07679 100644
--- a/features/standby_cluster.feature
+++ b/features/standby_cluster.feature
@@ -22,9 +22,6 @@ Feature: standby cluster
   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
diff --git a/features/steps/slots.py b/features/steps/slots.py
index 8a742b1c..f4a3cfa5 100644
--- a/features/steps/slots.py
+++ b/features/steps/slots.py
@@ -15,17 +15,25 @@ def create_logical_replication_slot(context, slot_name, pg_name, plugin):
         assert False, "Error creating slot {0} on {1} with plugin {2}".format(slot_name, pg_name, plugin)
 
 
-@then('{pg_name:w} has a logical replication slot named {slot_name} with the {plugin:w} plugin')
-def has_logical_replication_slot(context, pg_name, slot_name, plugin):
-    try:
-        row = context.pctl.query(pg_name, ("SELECT slot_type, plugin FROM pg_replication_slots"
-                                           " WHERE slot_name = '{0}'").format(slot_name)).fetchone()
-        assert row, "Couldn't find replication slot named {0}".format(slot_name)
-        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:
-        assert False, "Error looking for slot {0} on {1} with plugin {2}".format(slot_name, pg_name, plugin)
+@step('{pg_name:w} has a logical replication slot named {slot_name}'
+      ' with the {plugin:w} plugin after {time_limit:d} seconds')
+@then('{pg_name:w} has a logical replication slot named {slot_name}'
+      ' with the {plugin:w} plugin after {time_limit:d} seconds')
+def has_logical_replication_slot(context, pg_name, slot_name, plugin, time_limit):
+    time_limit *= context.timeout_multiplier
+    max_time = time.time() + int(time_limit)
+    while time.time() < max_time:
+        try:
+            row = context.pctl.query(pg_name, ("SELECT slot_type, plugin FROM pg_replication_slots"
+                                               f" WHERE slot_name = '{slot_name}'")).fetchone()
+            if row:
+                assert row[0] == "logical", f"Replication slot {slot_name} isn't a logical but {row[0]}"
+                assert row[1] == plugin, f"Replication slot {slot_name} using plugin {row[1]} rather than {plugin}"
+                return
+        except Exception:
+            pass
+        time.sleep(1)
+    assert False, f"Error looking for slot {slot_name} on {pg_name} with plugin {plugin}"
 
 
 @then('{pg_name:w} does not have a logical replication slot named {slot_name}')
@@ -38,13 +46,14 @@ def does_not_have_logical_replication_slot(context, pg_name, slot_name):
         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):
+@step('{slot_type:w} slot {slot_name:w} is in sync between {pg_name1:w} and {pg_name2:w} after {time_limit:d} seconds')
+def slots_in_sync(context, slot_type, slot_name, pg_name1, pg_name2, time_limit):
     time_limit *= context.timeout_multiplier
     max_time = time.time() + int(time_limit)
+    column = 'confirmed_flush_lsn' if slot_type.lower() == 'logical' else 'restart_lsn'
+    query = f"SELECT {column} FROM pg_replication_slots WHERE slot_name = '{slot_name}'"
     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]:
@@ -52,9 +61,31 @@ def logical_slots_in_sync(context, slot_name, pg_name1, pg_name2, time_limit):
         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)
+    assert False, \
+        f"{slot_type} slot {slot_name} is not in sync between {pg_name1} and {pg_name2} after {time_limit} seconds"
 
 
 @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))
+
+
+@step('I get all changes from physical slot {slot_name:w} on {pg_name:w}')
+def physical_slot_get_changes(context, slot_name, pg_name):
+    context.pctl.query(pg_name, f"SELECT * FROM pg_replication_slot_advance('{slot_name}', pg_current_wal_lsn())")
+
+
+@step('{pg_name:w} has a physical replication slot named {slot_name} after {time_limit:d} seconds')
+def has_physical_replication_slot(context, pg_name, slot_name, time_limit):
+    time_limit *= context.timeout_multiplier
+    max_time = time.time() + int(time_limit)
+    query = f"SELECT * FROM pg_catalog.pg_replication_slots WHERE slot_type = 'physical' AND slot_name = '{slot_name}'"
+    while time.time() < max_time:
+        try:
+            row = context.pctl.query(pg_name, query).fetchone()
+            if row:
+                return
+        except Exception:
+            pass
+        time.sleep(1)
+    assert False, f"Physical slot {slot_name} doesn't exist after {time_limit} seconds"
diff --git a/kubernetes/Dockerfile b/kubernetes/Dockerfile
index 2fa25f73..29a683bd 100644
--- a/kubernetes/Dockerfile
+++ b/kubernetes/Dockerfile
@@ -9,8 +9,8 @@ RUN export DEBIAN_FRONTEND=noninteractive \
             | xargs apt-get install -y vim-tiny curl jq locales git python3-pip python3-wheel \
     ## Make sure we have a en_US.UTF-8 locale available
     && localedef -i en_US -c -f UTF-8 -A /usr/share/locale/locale.alias en_US.UTF-8 \
-    && pip3 install setuptools \
-    && pip3 install 'git+https://github.com/zalando/patroni.git#egg=patroni[kubernetes]' \
+    && pip3 install --break-system-packages setuptools \
+    && pip3 install --break-system-packages 'git+https://github.com/zalando/patroni.git#egg=patroni[kubernetes]' \
     && PGHOME=/home/postgres \
     && mkdir -p $PGHOME \
     && chown postgres $PGHOME \
diff --git a/kubernetes/Dockerfile.citus b/kubernetes/Dockerfile.citus
index 1ae242bf..f9564521 100644
--- a/kubernetes/Dockerfile.citus
+++ b/kubernetes/Dockerfile.citus
@@ -10,12 +10,24 @@ RUN export DEBIAN_FRONTEND=noninteractive \
             | xargs apt-get install -y busybox vim-tiny curl jq less locales git python3-pip python3-wheel lsb-release \
     ## Make sure we have a en_US.UTF-8 locale available
     && localedef -i en_US -c -f UTF-8 -A /usr/share/locale/locale.alias en_US.UTF-8 \
-    && echo "deb [signed-by=/etc/apt/trusted.gpg.d/citusdata_community.gpg] https://packagecloud.io/citusdata/community/debian/ $(lsb_release -cs) main" > /etc/apt/sources.list.d/citusdata_community.list \
-    && curl -sL https://packagecloud.io/citusdata/community/gpgkey | gpg --dearmor > /etc/apt/trusted.gpg.d/citusdata_community.gpg \
-    && apt-get update -y \
-    && apt-get -y install postgresql-$PG_MAJOR-citus-11.3 \
-    && pip3 install setuptools \
-    && pip3 install 'git+https://github.com/zalando/patroni.git#egg=patroni[kubernetes]' \
+    && if [ $(dpkg --print-architecture) = 'arm64' ]; then \
+        apt-get install -y postgresql-server-dev-15 \
+                           gcc make autoconf \
+                           libc6-dev flex libcurl4-gnutls-dev \
+                           libicu-dev libkrb5-dev liblz4-dev \
+                           libpam0g-dev libreadline-dev libselinux1-dev\
+                           libssl-dev libxslt1-dev libzstd-dev uuid-dev \
+        && git clone -b "main" https://github.com/citusdata/citus.git \
+        && MAKEFLAGS="-j $(grep -c ^processor /proc/cpuinfo)" \
+        && cd citus && ./configure && make install && cd ../ && rm -rf /citus; \
+    else \
+        echo "deb [signed-by=/etc/apt/trusted.gpg.d/citusdata_community.gpg] https://packagecloud.io/citusdata/community/debian/ $(lsb_release -cs) main" > /etc/apt/sources.list.d/citusdata_community.list \
+        && curl -sL https://packagecloud.io/citusdata/community/gpgkey | gpg --dearmor > /etc/apt/trusted.gpg.d/citusdata_community.gpg \
+        && apt-get update -y \
+        && apt-get -y install postgresql-15-citus-12.0; \
+    fi \
+    && pip3 install --break-system-packages setuptools \
+    && pip3 install --break-system-packages 'git+https://github.com/zalando/patroni.git#egg=patroni[kubernetes]' \
     && PGHOME=/home/postgres \
     && mkdir -p $PGHOME \
     && chown postgres $PGHOME \
@@ -26,6 +38,9 @@ RUN export DEBIAN_FRONTEND=noninteractive \
     && chmod 664 /etc/passwd \
     # Clean up
     && apt-get remove -y git python3-pip python3-wheel \
+               postgresql-server-dev-15 gcc make autoconf \
+               libc6-dev flex libicu-dev libkrb5-dev liblz4-dev \
+               libpam0g-dev libreadline-dev libselinux1-dev libssl-dev libxslt1-dev libzstd-dev uuid-dev \
     && apt-get autoremove -y \
     && apt-get clean -y \
     && rm -rf /var/lib/apt/lists/* /root/.cache
diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py
index 87d3c9c1..08112651 100644
--- a/patroni/dcs/__init__.py
+++ b/patroni/dcs/__init__.py
@@ -28,6 +28,7 @@ from ..tags import Tags
 if TYPE_CHECKING:  # pragma: no cover
     from ..config import Config
 
+SLOT_ADVANCE_AVAILABLE_VERSION = 110000
 CITUS_COORDINATOR_GROUP_ID = 0
 citus_group_re = re.compile('^(0|[1-9][0-9]*)$')
 slot_name_re = re.compile('^[a-z0-9_]{1,63}$')
@@ -350,6 +351,11 @@ class Member(Tags, NamedTuple('Member',
                 logger.debug('Failed to parse Patroni version %s', version)
         return None
 
+    @property
+    def lsn(self) -> Optional[int]:
+        """Current LSN (receive/flush/replay)."""
+        return self.data.get('xlog_location')
+
 
 class RemoteMember(Member):
     """Represents a remote member (typically a primary) for a standby cluster.
@@ -964,24 +970,42 @@ class Cluster(NamedTuple('Cluster',
         candidates = [m for m in self.members if m.clonefrom and m.is_running and m.name not in exclude]
         return candidates[randint(0, len(candidates) - 1)] if candidates else self.leader
 
+    @staticmethod
+    def is_physical_slot(value: Union[Any, Dict[str, Any]]) -> bool:
+        """Check whether provided configuration is for permanent physical replication slot.
+
+        :returns: ``True`` if this is a physical replication slot, otherwise ``False``.
+        """
+        return not value or isinstance(value, dict) and value.get('type', 'physical') == 'physical'
+
     @property
     def __permanent_slots(self) -> Dict[str, Union[Dict[str, Any], Any]]:
         """Dictionary of permanent replication slots with their known LSN."""
-        ret = deepcopy(self.config.permanent_slots if self.config else {})
-        # If primary reported flush LSN for permanent slots we want to enrich our structure with it
-        for name, lsn in (self.slots or {}).items():
-            if name in ret:
-                if not ret[name]:
-                    ret[name] = {}
-                if isinstance(ret[name], dict):
-                    ret[name]['lsn'] = lsn
+        leader = self.leader and self.leader.member
+        leader_name = slot_name_from_member_name(leader.name) if leader and leader.lsn else None
+
+        slots = self.slots or {}
+        ret: Dict[str, Union[Dict[str, Any], Any]] = deepcopy(self.config.permanent_slots if self.config else {})
+
+        for name, value in list(ret.items()):
+            if not value:
+                value = ret[name] = {}
+            if isinstance(value, dict):
+                if name in slots:
+                    # If primary reported flush LSN for permanent slots we want to enrich our structure with it
+                    value['lsn'] = slots[name]
+                elif self.is_physical_slot(value) and name == leader_name and leader and leader.lsn:
+                    # there is no slot on the leader for itself, use `lsn` from the member key.
+                    value['lsn'] = leader.lsn
+                else:
+                    # Don't let anyone set 'lsn' in the global configuration :)
+                    value.pop('lsn', None)
         return ret
 
     @property
     def __permanent_physical_slots(self) -> Dict[str, Any]:
         """Dictionary of permanent ``physical`` replication slots."""
-        return {name: value for name, value in self.__permanent_slots.items()
-                if not value or isinstance(value, dict) and value.get('type', 'physical') == 'physical'}
+        return {name: value for name, value in self.__permanent_slots.items() if self.is_physical_slot(value)}
 
     @property
     def __permanent_logical_slots(self) -> Dict[str, Any]:
@@ -1013,7 +1037,7 @@ class Cluster(NamedTuple('Cluster',
         :returns: final dictionary of slot names, after merging with permanent slots and performing sanity checks.
         """
         slots: Dict[str, Dict[str, str]] = self._get_members_slots(my_name, role)
-        permanent_slots: Dict[str, Any] = self._get_permanent_slots(is_standby_cluster, role, nofailover)
+        permanent_slots: Dict[str, Any] = self._get_permanent_slots(is_standby_cluster, role, nofailover, major_version)
 
         disabled_permanent_logical_slots: List[str] = self._merge_permanent_slots(
             slots, permanent_slots, my_name, major_version)
@@ -1061,7 +1085,7 @@ class Cluster(NamedTuple('Cluster',
                     continue
 
                 if value['type'] == 'logical' and value.get('database') and value.get('plugin'):
-                    if major_version < 110000:
+                    if major_version < SLOT_ADVANCE_AVAILABLE_VERSION:
                         disabled_permanent_logical_slots.append(name)
                     elif name in slots:
                         logger.error("Permanent logical replication slot {'%s': %s} is conflicting with"
@@ -1073,7 +1097,8 @@ class Cluster(NamedTuple('Cluster',
             logger.error("Bad value for slot '%s' in permanent_slots: %s", name, permanent_slots[name])
         return disabled_permanent_logical_slots
 
-    def _get_permanent_slots(self, is_standby_cluster: bool, role: str, nofailover: bool) -> Dict[str, Any]:
+    def _get_permanent_slots(self, is_standby_cluster: bool, role: str,
+                             nofailover: bool, major_version: int) -> Dict[str, Any]:
         """Get configured permanent replication slots.
 
         .. note::
@@ -1089,6 +1114,7 @@ class Cluster(NamedTuple('Cluster',
                                    the outside because we want to protect from the ``/config`` key removal.
         :param role: role of this node -- ``primary``, ``standby_leader`` or ``replica``.
         :param nofailover: ``True`` if this node is tagged to not be a failover candidate.
+        :param major_version: postgresql major version.
 
         :returns: dictionary of permanent slot names mapped to attributes.
         """
@@ -1096,9 +1122,11 @@ class Cluster(NamedTuple('Cluster',
             return {}
 
         if is_standby_cluster:
-            return self.__permanent_physical_slots if role == 'standby_leader' else {}
+            return self.__permanent_physical_slots \
+                if major_version >= SLOT_ADVANCE_AVAILABLE_VERSION or role == 'standby_leader' else {}
 
-        return self.__permanent_slots if role in ('master', 'primary') else self.__permanent_logical_slots
+        return self.__permanent_slots if major_version >= SLOT_ADVANCE_AVAILABLE_VERSION\
+            or role in ('master', 'primary') else self.__permanent_logical_slots
 
     def _get_members_slots(self, my_name: str, role: str) -> Dict[str, Dict[str, str]]:
         """Get physical replication slots configuration for members that sourcing from this node.
@@ -1143,21 +1171,33 @@ class Cluster(NamedTuple('Cluster',
                                    for k, v in slot_conflicts.items() if len(v) > 1))
         return slots
 
-    def has_permanent_logical_slots(self, my_name: str, nofailover: bool, major_version: int = 110000) -> bool:
+    def has_permanent_slots(self, my_name: str, nofailover: bool = False) -> bool:
+        """Check if the given member node has permanent replication slots configured.
+
+        :param my_name: name of the member node to check.
+        :param nofailover: ``True`` if this node is tagged to not be a failover candidate.
+
+        :returns: ``True`` if there are permanent replication slots configured, otherwise ``False``.
+        """
+        members_slots: Dict[str, Dict[str, str]] = self._get_members_slots(my_name, 'replica')
+        permanent_slots: Dict[str, Any] = self._get_permanent_slots(nofailover, 'replica', False,
+                                                                    SLOT_ADVANCE_AVAILABLE_VERSION)
+        slots = deepcopy(members_slots)
+        self._merge_permanent_slots(slots, permanent_slots, my_name, SLOT_ADVANCE_AVAILABLE_VERSION)
+        return len(slots) > len(members_slots)
+
+    def _has_permanent_logical_slots(self, my_name: str, nofailover: bool) -> bool:
         """Check if the given member node has permanent ``logical`` replication slots configured.
 
         :param my_name: name of the member node to check.
         :param nofailover: ``True`` if this node is tagged to not be a failover candidate.
-        :param major_version: the PostgreSQL major version number.
 
-        :returns: ``False`` if PostgreSQL is < 11, ``True`` if any detected replications slots are ``logical``.
+        :returns: ``True`` if any detected replications slots are ``logical``, otherwise ``False``.
         """
-        if major_version < 110000:
-            return False
-        slots = self.get_replication_slots(my_name, 'replica', nofailover, major_version).values()
+        slots = self.get_replication_slots(my_name, 'replica', nofailover, SLOT_ADVANCE_AVAILABLE_VERSION).values()
         return any(v for v in slots if v.get("type") == "logical")
 
-    def should_enforce_hot_standby_feedback(self, my_name: str, nofailover: bool, major_version: int) -> bool:
+    def should_enforce_hot_standby_feedback(self, my_name: str, nofailover: bool) -> bool:
         """Determine whether ``hot_standby_feedback`` should be enabled for the given member.
 
         The ``hot_standby_feedback`` must be enabled if the current replica has ``logical`` slots,
@@ -1165,20 +1205,16 @@ class Cluster(NamedTuple('Cluster',
 
         :param my_name: name of the member node to check.
         :param nofailover: ``True`` if this node is tagged to not be a failover candidate.
-        :param major_version: PostgreSQL major version number.
 
-        :returns: ``True`` if this node or any member replicating from this node has permanent logical slots.
-                 ``False`` if PostgreSQL major version is < 11.
+        :returns: ``True`` if this node or any member replicating from this node has
+                  permanent logical slots, otherwise ``False``.
         """
-        if major_version < 110000:
-            return False
-
-        if self.has_permanent_logical_slots(my_name, nofailover, major_version):
+        if self._has_permanent_logical_slots(my_name, nofailover):
             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 any(self.should_enforce_hot_standby_feedback(m.name, m.nofailover) for m in members)
         return False
 
     def get_my_slot_name_on_primary(self, my_name: str, replicatefrom: Optional[str]) -> str:
diff --git a/patroni/dcs/etcd3.py b/patroni/dcs/etcd3.py
index 0a71caa0..5cbd813f 100644
--- a/patroni/dcs/etcd3.py
+++ b/patroni/dcs/etcd3.py
@@ -205,7 +205,7 @@ class Etcd3Client(AbstractEtcdClientWithFailover):
 
     def __init__(self, config: Dict[str, Any], dns_resolver: DnsCachingResolver, cache_ttl: int = 300) -> None:
         self._token = None
-        self._cluster_version: Tuple[int] = tuple()
+        self._cluster_version: Tuple[int, ...] = tuple()
         super(Etcd3Client, self).__init__({**config, 'version_prefix': '/v3beta'}, dns_resolver, cache_ttl)
 
         try:
diff --git a/patroni/dcs/kubernetes.py b/patroni/dcs/kubernetes.py
index 3be7cee3..d4158368 100644
--- a/patroni/dcs/kubernetes.py
+++ b/patroni/dcs/kubernetes.py
@@ -836,7 +836,7 @@ class Kubernetes(AbstractDCS):
         self._api.configure_timeouts(self.loop_wait, self._retry.deadline, self.ttl)
 
         # retriable_http_codes supposed to be either int, list of integers or comma-separated string with integers.
-        retriable_http_codes = config.get('retriable_http_codes', [])
+        retriable_http_codes: Union[str, List[Union[str, int]]] = config.get('retriable_http_codes', [])
         if not isinstance(retriable_http_codes, list):
             retriable_http_codes = [c.strip() for c in str(retriable_http_codes).split(',')]
 
diff --git a/patroni/dcs/zookeeper.py b/patroni/dcs/zookeeper.py
index e4af1b1f..6390aa2e 100644
--- a/patroni/dcs/zookeeper.py
+++ b/patroni/dcs/zookeeper.py
@@ -90,7 +90,7 @@ class ZooKeeper(AbstractDCS):
     def __init__(self, config: Dict[str, Any]) -> None:
         super(ZooKeeper, self).__init__(config)
 
-        hosts = config.get('hosts', [])
+        hosts: Union[str, List[str]] = config.get('hosts', [])
         if isinstance(hosts, list):
             hosts = ','.join(hosts)
 
@@ -301,7 +301,7 @@ class ZooKeeper(AbstractDCS):
                 raise ZooKeeperError('ZooKeeper in not responding properly')
         # The /status ZNode was updated or doesn't exist
         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:
+                or cluster.has_permanent_slots(self._name) and not cluster.slots:
             # If current node is the leader just clear the event without fetching anything (we are updating the /status)
             if cluster.leader and cluster.leader.name == self._name:
                 self.event.clear()
diff --git a/patroni/ha.py b/patroni/ha.py
index a8274ee3..d5d6e16b 100644
--- a/patroni/ha.py
+++ b/patroni/ha.py
@@ -772,13 +772,13 @@ class Ha(object):
             if cluster_history:
                 self.dcs.set_history_value('[]')
         elif not cluster_history or cluster_history[-1][0] != primary_timeline - 1 or len(cluster_history[-1]) != 5:
-            cluster_history = {line[0]: line for line in cluster_history}
+            cluster_history_dict: Dict[int, List[Any]] = {line[0]: list(line) for line in cluster_history}
             history: List[List[Any]] = list(map(list, self.state_handler.get_history(primary_timeline)))
             if self.cluster.config:
                 history = history[-self.cluster.config.max_timelines_history:]
             for line in history:
                 # enrich current history with promotion timestamps stored in DCS
-                cluster_history_line = list(cluster_history.get(line[0], []))
+                cluster_history_line = cluster_history_dict.get(line[0], [])
                 if len(line) == 3 and len(cluster_history_line) >= 4 and cluster_history_line[1] == line[1]:
                     line.append(cluster_history_line[3])
                     if len(cluster_history_line) == 5:
diff --git a/patroni/postgresql/__init__.py b/patroni/postgresql/__init__.py
index a37e15e1..af48886f 100644
--- a/patroni/postgresql/__init__.py
+++ b/patroni/postgresql/__init__.py
@@ -27,7 +27,7 @@ from .sync import SyncHandler
 from .. import psycopg
 from ..async_executor import CriticalTask
 from ..collections import CaseInsensitiveSet
-from ..dcs import Cluster, Leader, Member
+from ..dcs import Cluster, Leader, Member, SLOT_ADVANCE_AVAILABLE_VERSION
 from ..exceptions import PostgresConnectionException
 from ..utils import Retry, RetryFailedError, polling_loop, data_directory_is_empty, parse_int
 
@@ -112,7 +112,7 @@ class Postgresql(object):
         self._state_entry_timestamp = 0
 
         self._cluster_info_state = {}
-        self._has_permanent_logical_slots = True
+        self._has_permanent_slots = True
         self._enforce_hot_standby_feedback = False
         self._cached_replica_timeline = None
 
@@ -174,6 +174,11 @@ class Postgresql(object):
         """:returns: `True` if Postgres version supports more than one synchronous node."""
         return self._major_version >= 90600
 
+    @property
+    def can_advance_slots(self) -> bool:
+        """``True`` if :attr:``major_version`` is greater than 110000."""
+        return self.major_version >= SLOT_ADVANCE_AVAILABLE_VERSION
+
     @property
     def cluster_info_query(self) -> str:
         """Returns the monitoring query with a fixed number of fields.
@@ -208,8 +213,9 @@ class Postgresql(object):
             extra = ("pg_catalog.current_setting('restore_command')" if self._major_version >= 120000 else "NULL") +\
                 ", " + ("(SELECT pg_catalog.json_agg(s.*) FROM (SELECT slot_name, slot_type as type, datoid::bigint, "
                         "plugin, catalog_xmin, pg_catalog.pg_wal_lsn_diff(confirmed_flush_lsn, '0/0')::bigint"
-                        " AS confirmed_flush_lsn FROM pg_catalog.pg_get_replication_slots()) AS s)"
-                        if self._has_permanent_logical_slots and self._major_version >= 110000 else "NULL") + extra
+                        " AS confirmed_flush_lsn, pg_catalog.pg_wal_lsn_diff(restart_lsn, '0/0')::bigint"
+                        " AS restart_lsn FROM pg_catalog.pg_get_replication_slots()) AS s)"
+                        if self._has_permanent_slots and self.can_advance_slots else "NULL") + extra
             extra = (", CASE WHEN latest_end_lsn IS NULL THEN NULL ELSE received_tli END,"
                      " slot_name, conninfo, status, {0} FROM pg_catalog.pg_stat_get_wal_receiver()").format(extra)
             if self.role == 'standby_leader':
@@ -434,18 +440,15 @@ class Postgresql(object):
             return
 
         if self._global_config.is_standby_cluster:
+            self._has_permanent_slots = False
             # Standby cluster can't have logical replication slots, and we don't need to enforce hot_standby_feedback
-            self._has_permanent_logical_slots = False
             self.set_enforce_hot_standby_feedback(False)
         elif cluster and cluster.config and cluster.config.modify_version:
-            self._has_permanent_logical_slots =\
-                cluster.has_permanent_logical_slots(self.name, nofailover, self.major_version)
-
+            self._has_permanent_slots = cluster.has_permanent_slots(self.name, nofailover)
             # We want to enable hot_standby_feedback if the replica is supposed
             # to have a logical slot or in case if it is the cascading replica.
             self.set_enforce_hot_standby_feedback(
-                self._has_permanent_logical_slots
-                or cluster.should_enforce_hot_standby_feedback(self.name, nofailover, self.major_version))
+                self.can_advance_slots and cluster.should_enforce_hot_standby_feedback(self.name, nofailover))
 
     def _cluster_info_state_get(self, name: str) -> Optional[Any]:
         if not self._cluster_info_state:
@@ -456,7 +459,7 @@ class Postgresql(object):
                                                'received_tli', 'slot_name', 'conninfo', 'receiver_state',
                                                'restore_command', 'slots', 'synchronous_commit',
                                                'synchronous_standby_names', 'pg_stat_replication'], result))
-                if self._has_permanent_logical_slots:
+                if self._has_permanent_slots and self.can_advance_slots:
                     cluster_info_state['slots'] =\
                         self.slots_handler.process_permanent_slots(cluster_info_state['slots'])
                 self._cluster_info_state = cluster_info_state
@@ -568,7 +571,7 @@ class Postgresql(object):
                              r'lsn: ([0-9A-Fa-f]+/[0-9A-Fa-f]+), prev ([0-9A-Fa-f]+/[0-9A-Fa-f]+), '
                              r'.*?desc: (.+)', out.decode('utf-8'))
             if match:
-                return match.groups()
+                return match.group(1), match.group(2), match.group(3), match.group(4)
         return None, None, None, None
 
     def latest_checkpoint_location(self) -> Optional[int]:
@@ -1023,7 +1026,7 @@ class Postgresql(object):
             return None, None
 
     @contextmanager
-    def get_replication_connection_cursor(self, host: Optional[str] = None, port: int = 5432,
+    def get_replication_connection_cursor(self, host: Optional[str] = None, port: Union[int, str] = 5432,
                                           **kwargs: Any) -> Iterator[Union['cursor', 'Cursor[Any]']]:
         conn_kwargs = self.config.replication.copy()
         conn_kwargs.update(host=host, port=int(port) if port else None, user=conn_kwargs.pop('username'),
diff --git a/patroni/postgresql/config.py b/patroni/postgresql/config.py
index 315bf8c7..a918a2cf 100644
--- a/patroni/postgresql/config.py
+++ b/patroni/postgresql/config.py
@@ -1026,17 +1026,14 @@ class ConfigHandler(object):
         # "notify" connection_pool about the "new" local connection address
         self._postgresql.connection_pool.conn_kwargs = local_conn_kwargs
 
-    def _get_pg_settings(
-            self, names: Collection[str]
-    ) -> Dict[str, Tuple[str, str, Optional[str], str, str, Optional[str]]]:
+    def _get_pg_settings(self, names: Collection[str]) -> Dict[Any, Tuple[Any, ...]]:
         return {r[0]: r for r in self._postgresql.query(('SELECT name, setting, unit, vartype, context, sourcefile'
                                                          + ' FROM pg_catalog.pg_settings '
                                                          + ' WHERE pg_catalog.lower(name) = ANY(%s)'),
                                                         [n.lower() for n in names])}
 
     @staticmethod
-    def _handle_wal_buffers(old_values: Dict[str, Tuple[str, str, Optional[str], str, str, Optional[str]]],
-                            changes: CaseInsensitiveDict) -> None:
+    def _handle_wal_buffers(old_values: Dict[Any, Tuple[Any, ...]], changes: CaseInsensitiveDict) -> None:
         wal_block_size = parse_int(old_values['wal_block_size'][1]) or 8192
         wal_segment_size = old_values['wal_segment_size']
         wal_segment_unit = parse_int(wal_segment_size[2], 'B') or 8192 \
diff --git a/patroni/postgresql/slots.py b/patroni/postgresql/slots.py
index 7391f543..29bbb130 100644
--- a/patroni/postgresql/slots.py
+++ b/patroni/postgresql/slots.py
@@ -16,6 +16,7 @@ from .misc import format_lsn, fsync_dir
 from ..dcs import Cluster, Leader
 from ..file_perm import pg_perm
 from ..psycopg import OperationalError
+from ..utils import parse_int
 
 if TYPE_CHECKING:  # pragma: no cover
     from psycopg import Cursor
@@ -231,15 +232,16 @@ class SlotsHandler:
         ret: Dict[str, int] = {}
 
         slots_dict: Dict[str, Dict[str, Any]] = {slot['slot_name']: slot for slot in slots or []}
-        if slots_dict:
-            for name, value in slots_dict.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])
+        for name, value in slots_dict.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
+                        self._replication_slots[name]['restart_lsn'] = ret[name] = value['restart_lsn']
+                else:
+                    self._schedule_load_slots = True
 
         # It could happen that the slot was deleted in the background, we want to detect this case
         if any(name not in slots_dict for name in self._replication_slots.keys()):
@@ -260,16 +262,19 @@ class SlotsHandler:
         """
         if self._postgresql.major_version >= 90400 and self._schedule_load_slots:
             replication_slots: Dict[str, Dict[str, Any]] = {}
-            extra = ", catalog_xmin, pg_catalog.pg_wal_lsn_diff(confirmed_flush_lsn, '0/0')::bigint" \
+            pg_wal_lsn_diff = f"pg_catalog.pg_{self._postgresql.wal_name}_{self._postgresql.lsn_name}_diff"
+            extra = f", catalog_xmin, {pg_wal_lsn_diff}(confirmed_flush_lsn, '0/0')::bigint" \
                 if self._postgresql.major_version >= 100000 else ""
             skip_temp_slots = ' WHERE NOT temporary' if self._postgresql.major_version >= 100000 else ''
-            for r in self._query('SELECT slot_name, slot_type, plugin, database, datoid'
-                                 f'{extra} FROM pg_catalog.pg_replication_slots{skip_temp_slots}'):
+            for r in self._query(f"SELECT slot_name, slot_type, {pg_wal_lsn_diff}(restart_lsn, '0/0')::bigint, plugin,"
+                                 f" database, datoid{extra} FROM pg_catalog.pg_replication_slots{skip_temp_slots}"):
                 value = {'type': r[1]}
                 if r[1] == 'logical':
-                    value.update(plugin=r[2], database=r[3], datoid=r[4])
+                    value.update(plugin=r[3], database=r[4], datoid=r[5])
                     if self._postgresql.major_version >= 100000:
-                        value.update(catalog_xmin=r[5], confirmed_flush_lsn=r[6])
+                        value.update(catalog_xmin=r[6], confirmed_flush_lsn=r[7])
+                else:
+                    value['restart_lsn'] = r[2]
                 replication_slots[r[0]] = value
             self._replication_slots = replication_slots
             self._schedule_load_slots = False
@@ -313,7 +318,7 @@ class SlotsHandler:
                             ' true AS dropped FROM slots WHERE not active) '
                             'SELECT active, COALESCE(dropped, false) FROM slots'
                             ' FULL OUTER JOIN dropped ON true'), name)
-        return rows[0] if rows else (False, False)
+        return (rows[0][0], rows[0][1]) if rows else (False, False)
 
     def _drop_incorrect_slots(self, cluster: Cluster, slots: Dict[str, Any], paused: bool) -> None:
         """Compare required slots and configured as permanent slots with those found, dropping extraneous ones.
@@ -353,7 +358,7 @@ class SlotsHandler:
                     self._schedule_load_slots = True
 
     def _ensure_physical_slots(self, slots: Dict[str, Any]) -> None:
-        """Create any missing physical replication *slots*.
+        """Create or advance physical replication *slots*.
 
         Any failures are logged and do not interrupt creation of all *slots*.
 
@@ -362,7 +367,9 @@ class SlotsHandler:
         """
         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':
+            if value['type'] != 'physical':
+                continue
+            if name not in self._replication_slots:
                 try:
                     self._query(f"SELECT pg_catalog.pg_create_physical_replication_slot(%s{immediately_reserve})"
                                 f" WHERE NOT EXISTS (SELECT 1 FROM pg_catalog.pg_replication_slots"
@@ -371,6 +378,16 @@ class SlotsHandler:
                 except Exception:
                     logger.exception("Failed to create physical replication slot '%s'", name)
                 self._schedule_load_slots = True
+            elif not self._postgresql.is_primary() and self._postgresql.can_advance_slots \
+                    and self._replication_slots[name]['type'] == 'physical':
+                value['restart_lsn'] = self._replication_slots[name]['restart_lsn']
+                lsn = parse_int(value.get('lsn'))
+                if lsn and lsn > value['restart_lsn']:  # The slot has feedback in DCS and needs to be advanced
+                    try:
+                        lsn = format_lsn(lsn)
+                        self._query("SELECT pg_catalog.pg_replication_slot_advance(%s, %s)", name, lsn)
+                    except Exception as exc:
+                        logger.error("Error while advancing replication slot %s to position '%s': %r", name, lsn, exc)
 
     @contextmanager
     def get_local_connection_cursor(self, **kwargs: Any) -> Iterator[Union['cursor', 'Cursor[Any]']]:
@@ -484,10 +501,11 @@ class SlotsHandler:
                                replicatefrom: Optional[str] = None, paused: bool = False) -> List[str]:
         """During the HA loop read, check and alter replication slots found in the cluster.
 
-        Read physical and logical slots found on the primary, then compare to those configured in the DCS.
+        Read physical and logical slots from ``pg_replication_slots``, then compare to those configured in the DCS.
         Drop any slots that do not match those required by configuration and are not configured as permanent.
-        Create any missing physical slots. If we are the leader then logical slots too, otherwise if logical slots
-        are known and active create them on replica nodes.
+        Create any missing physical slots, or advance their position according to feedback stored in DCS.
+        If we are the primary then create logical slots, otherwise if logical slots are known and active create
+        them on replica nodes by copying slot files from the primary.
 
         :param cluster: object containing stateful information for the cluster.
         :param nofailover: ``True`` if this node has been tagged to not be a failover candidate.
diff --git a/patroni/version.py b/patroni/version.py
index 87eff52e..ff98d3c0 100644
--- a/patroni/version.py
+++ b/patroni/version.py
@@ -2,4 +2,4 @@
 
 :var __version__: the current Patroni version.
 """
-__version__ = '3.1.0'
+__version__ = '3.1.1'
diff --git a/tox.ini b/tox.ini
index b145f483..bf66caef 100644
--- a/tox.ini
+++ b/tox.ini
@@ -6,6 +6,7 @@ postgres_matrix =
     pg13: PG_MAJOR = 13
     pg14: PG_MAJOR = 14
     pg15: PG_MAJOR = 15
+    pg16: PG_MAJOR = 16
 psycopg_deps =
     py{37,38,39,310,311}-{lin,win}: psycopg[binary]
     mac: psycopg2-binary
@@ -106,7 +107,7 @@ description = Reformat code with black
 deps = black
 commands = black {posargs:patroni tests}
 
-[testenv:pg{12,13,14,15}-docker-build]
+[testenv:pg{12,13,14,15,16}-docker-build]
 description = Build docker containers needed for testing
 labels =
     behave
@@ -124,7 +125,7 @@ commands =
         --file features/Dockerfile
 allowlist_externals = docker
 
-[testenv:pg{12,13,14,15}-docker-behave-{etcd}-{lin,mac}]
+[testenv:pg{12,13,14,15,16}-docker-behave-{etcd}-{lin,mac}]
 description = Run behaviour tests in patroni-dev docker container
 setenv =
     etcd: DCS=etcd
@@ -133,7 +134,7 @@ setenv =
 labels =
     behave
 depends =
-    pg{11,12,13,14,15}-docker-build
+    pg{11,12,13,14,15,16}-docker-build
 
 # There's a bug which affects calling multiple envs on the command line
 # This should be a valid command: tox -e 'py{36,37,38,39,310,311}-behave-{env:DCS}-lin'


Test patch:
diff --git a/.github/workflows/tests.yaml b/.github/workflows/tests.yaml
index bc6a1be3..aba3e751 100644
--- a/.github/workflows/tests.yaml
+++ b/.github/workflows/tests.yaml
@@ -5,6 +5,7 @@ on:
   push:
     branches:
     - master
+    - 'REL_[0-9]+_[0-9]+'
 
 env:
   CODACY_PROJECT_TOKEN: ${{ secrets.CODACY_PROJECT_TOKEN }}
@@ -173,7 +174,7 @@ jobs:
 
     - uses: jakebailey/pyright-action@v1
       with:
-        version: 1.1.320
+        version: 1.1.326
 
   docs:
     runs-on: ubuntu-latest
diff --git a/tests/__init__.py b/tests/__init__.py
index 68961af7..5240cec6 100644
--- a/tests/__init__.py
+++ b/tests/__init__.py
@@ -104,12 +104,14 @@ class MockCursor(object):
         elif sql.startswith('SELECT slot_name, slot_type, datname, plugin, catalog_xmin'):
             self.results = [('ls', 'logical', 'a', 'b', 100, 500, b'123456')]
         elif sql.startswith('SELECT slot_name'):
-            self.results = [('blabla', 'physical'), ('foobar', 'physical'), ('ls', 'logical', 'b', 'a', 5, 100, 500)]
+            self.results = [('blabla', 'physical', 12345),
+                            ('foobar', 'physical', 12345),
+                            ('ls', 'logical', 499, 'b', 'a', 5, 100, 500)]
         elif sql.startswith('WITH slots AS (SELECT slot_name, active'):
             self.results = [(False, True)] if self.rowcount == 1 else []
         elif sql.startswith('SELECT CASE WHEN pg_catalog.pg_is_in_recovery()'):
             self.results = [(1, 2, 1, 0, False, 1, 1, None, None, 'streaming', '',
-                             [{"slot_name": "ls", "confirmed_flush_lsn": 12345}],
+                             [{"slot_name": "ls", "confirmed_flush_lsn": 12345, "restart_lsn": 12344}],
                              'on', 'n1', None)]
         elif sql.startswith('SELECT pg_catalog.pg_is_in_recovery()'):
             self.results = [(False, 2)]
@@ -252,8 +254,8 @@ class BaseTestPostgresql(PostgresInit):
         if not os.path.exists(self.p.data_dir):
             os.makedirs(self.p.data_dir)
 
-        self.leadermem = Member(0, 'leader', 28, {
-            'state': 'running', 'conn_url': 'postgres://replicator:rep-pass@127.0.0.1:5435/postgres'})
+        self.leadermem = Member(0, 'leader', 28, {'xlog_location': 100, 'state': 'running',
+                                                  'conn_url': 'postgres://replicator:rep-pass@127.0.0.1:5435/postgres'})
         self.leader = Leader(-1, 28, self.leadermem)
         self.other = Member(0, 'test-1', 28, {'conn_url': 'postgres://replicator:rep-pass@127.0.0.1:5433/postgres',
                                               'state': 'running', 'tags': {'replicatefrom': 'leader'}})
diff --git a/tests/test_slots.py b/tests/test_slots.py
index 884f76b1..d1b8d458 100644
--- a/tests/test_slots.py
+++ b/tests/test_slots.py
@@ -53,6 +53,7 @@ class TestSlotsHandler(BaseTestPostgresql):
         self.p.set_role('replica')
         with patch.object(Postgresql, 'is_primary', Mock(return_value=False)), \
                 patch.object(SlotsHandler, 'drop_replication_slot') as mock_drop:
+            config.data['slots'].pop('ls')
             self.s.sync_replication_slots(cluster, False, paused=True)
             mock_drop.assert_not_called()
         self.p.set_role('primary')
@@ -69,6 +70,8 @@ class TestSlotsHandler(BaseTestPostgresql):
             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)
+                self.p.set_role('replica')
+                self.s.sync_replication_slots(cluster, False)
 
     def test_cascading_replica_sync_replication_slots(self):
         """Test sync with a cascading replica so physical slots are present on a replica."""
@@ -82,12 +85,12 @@ class TestSlotsHandler(BaseTestPostgresql):
         self.p.set_role('replica')
         with patch.object(Postgresql, '_query') as mock_query, \
                 patch.object(Postgresql, 'is_primary', Mock(return_value=False)):
-            mock_query.return_value = [('ls', 'logical', 'b', 'a', 5, 12345, 105)]
+            mock_query.return_value = [('ls', 'logical', 104, 'b', 'a', 5, 12345, 105)]
             ret = self.s.sync_replication_slots(cluster, False)
         self.assertEqual(ret, [])
 
     def test_process_permanent_slots(self):
-        config = ClusterConfig(1, {'slots': {'ls': {'database': 'a', 'plugin': 'b'}},
+        config = ClusterConfig(1, {'slots': {'ls': {'database': 'a', 'plugin': 'b'}, 'blabla': {'type': 'physical'}},
                                    'ignore_slots': [{'name': 'blabla'}]}, 1)
         cluster = Cluster(True, config, self.leader, Status.empty(), [self.me, self.other, self.leadermem],
                           None, SyncState.empty(), None, None)
@@ -98,8 +101,10 @@ class TestSlotsHandler(BaseTestPostgresql):
             mock_query.return_value = [(
                 1, 0, 0, 0, 0, 0, 0, 0, 0, None, None,
                 [{"slot_name": "ls", "type": "logical", "datoid": 5, "plugin": "b",
-                  "confirmed_flush_lsn": 12345, "catalog_xmin": 105}])]
-            self.assertEqual(self.p.slots(), {'ls': 12345})
+                  "confirmed_flush_lsn": 12345, "catalog_xmin": 105, "restart_lsn": 12344},
+                 {"slot_name": "blabla", "type": "physical", "datoid": None, "plugin": None,
+                  "confirmed_flush_lsn": None, "catalog_xmin": 105, "restart_lsn": 12344}])]
+            self.assertEqual(self.p.slots(), {'ls': 12345, 'blabla': 12344})
 
             self.p.reset_cluster_info_state(None)
             mock_query.return_value = [(
@@ -114,8 +119,8 @@ class TestSlotsHandler(BaseTestPostgresql):
         self.cluster.slots['ls'] = 12346
         with patch.object(SlotsHandler, 'check_logical_slots_readiness', Mock(return_value=False)):
             self.assertEqual(self.s.sync_replication_slots(self.cluster, False), [])
-        self.s._schedule_load_slots = False
-        with patch.object(MockCursor, 'execute', Mock(side_effect=psycopg.OperationalError)), \
+        with patch.object(SlotsHandler, '_query', Mock(return_value=[('ls', 'logical', 499, 'b', 'a', 5, 100, 500)])), \
+                patch.object(MockCursor, 'execute', Mock(side_effect=psycopg.OperationalError)), \
                 patch.object(SlotsAdvanceThread, 'schedule', Mock(return_value=(True, ['ls']))), \
                 patch.object(psycopg.OperationalError, 'diag') as mock_diag:
             type(mock_diag).sqlstate = PropertyMock(return_value='58P01')
@@ -177,3 +182,18 @@ class TestSlotsHandler(BaseTestPostgresql):
         with patch.object(SlotsHandler, 'get_local_connection_cursor', Mock(side_effect=Exception)):
             self.s.schedule_advance_slots({'foo': {'bar': 100}})
             self.s._advance.sync_slots()
+
+    @patch.object(Postgresql, 'is_primary', Mock(return_value=False))
+    def test_advance_physical_slots(self):
+        config = ClusterConfig(1, {'slots': {'blabla': {'type': 'physical'}, 'leader': None}}, 1)
+        cluster = Cluster(True, config, self.leader, Status(0, {'blabla': 12346}),
+                          [self.me, self.other, self.leadermem], None, SyncState.empty(), None, None)
+        self.s.sync_replication_slots(cluster, False)
+        with patch.object(SlotsHandler, '_query', Mock(side_effect=[[('blabla', 'physical', 12345, None, None, None,
+                                                                      None, None)], Exception])) as mock_query, \
+                patch('patroni.postgresql.slots.logger.error') as mock_error:
+            self.s.sync_replication_slots(cluster, False)
+            self.assertEqual(mock_query.call_args[0],
+                             ("SELECT pg_catalog.pg_replication_slot_advance(%s, %s)", "blabla", '0/303A'))
+            self.assertEqual(mock_error.call_args[0][0],
+                             "Error while advancing replication slot %s to position '%s': %r")