Merge branch 'master' of github.com:zalando/patroni into feature/quorum-commit

.github/workflows/install_deps.py

@@ -45,8 +45,8 @@ def install_packages(what):
     packages['exhibitor'] = packages['zookeeper']
     packages = packages.get(what, [])
     ver = versions.get(what)
-    if float(ver) >= 15:
-        packages += ['postgresql-{0}-citus-11.2'.format(ver)]
+    if float(ver) == 15:
+        packages += ['postgresql-{0}-citus-12.0'.format(ver)]
     subprocess.call(['sudo', 'apt-get', 'update', '-y'])
     return subprocess.call(['sudo', 'apt-get', 'install', '-y', 'postgresql-' + ver, 'expect-dev'] + packages)
 

.github/workflows/mapping.py

@@ -1 +1 @@
-versions = {'etcd': '9.6', 'etcd3': '14', 'consul': '13', 'exhibitor': '12', 'raft': '11', 'kubernetes': '15'}
+versions = {'etcd': '9.6', 'etcd3': '16', 'consul': '13', 'exhibitor': '12', 'raft': '11', 'kubernetes': '15'}

.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

README.rst

@@ -12,7 +12,7 @@ Patroni is a template for high availability (HA) PostgreSQL solutions using Pyth
 
 We call Patroni a "template" because it is far from being a one-size-fits-all or plug-and-play replication system. It will have its own caveats. Use wisely.
 
-Currently supported PostgreSQL versions: 9.3 to 15.
+Currently supported PostgreSQL versions: 9.3 to 16.
 
 **Note to Citus users**: Starting from 3.0 Patroni nicely integrates with the `Citus <https://github.com/citusdata/citus>`__ database extension to Postgres. Please check the `Citus support page <https://github.com/zalando/patroni/blob/master/docs/citus.rst>`__ in the Patroni documentation for more info about how to use Patroni high availability together with a Citus distributed cluster.
 

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

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
 

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.
 
@@ -81,3 +81,16 @@ Note: **slots** is a hashmap while **ignore_slots** is an array. For example:
           - name: ignored_physical_slot_name
             type: physical
           ...
+
+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
+          ...

docs/index.rst

@@ -10,7 +10,7 @@ Patroni is a template for high availability (HA) PostgreSQL solutions using Pyth
 
 We call Patroni a "template" because it is far from being a one-size-fits-all or plug-and-play replication system. It will have its own caveats. Use wisely. There are many ways to run high availability with PostgreSQL; for a list, see the `PostgreSQL Documentation <https://wiki.postgresql.org/wiki/Replication,_Clustering,_and_Connection_Pooling>`__.
 
-Currently supported PostgreSQL versions: 9.3 to 15.
+Currently supported PostgreSQL versions: 9.3 to 16.
 
 **Note to Citus users**: Starting from 3.0 Patroni nicely integrates with the `Citus <https://github.com/citusdata/citus>`__ database extension to Postgres. Please check the :ref:`Citus support page <citus>` in the Patroni documentation for more info about how to use Patroni high availability together with a Citus distributed cluster.
 

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
 -------------
 

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

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

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

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"

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 \

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

patroni/config.py

@@ -16,6 +16,7 @@ from .dcs import ClusterConfig, Cluster
 from .exceptions import ConfigParseError
 from .file_perm import pg_perm
 from .postgresql.config import ConfigHandler
+from .validator import IntValidator
 from .utils import deep_compare, parse_bool, parse_int, patch_config
 
 logger = logging.getLogger(__name__)
@@ -492,8 +493,9 @@ class Config(object):
             if name not in ConfigHandler.CMDLINE_OPTIONS:
                 pg_params[name] = value
             elif not is_local:
-                if ConfigHandler.CMDLINE_OPTIONS[name][1](value):
-                    pg_params[name] = value
+                validator = ConfigHandler.CMDLINE_OPTIONS[name][1]
+                if validator(value):
+                    pg_params[name] = int(value) if isinstance(validator, IntValidator) else value
                 else:
                     logger.warning("postgresql parameter %s=%s failed validation, defaulting to %s",
                                    name, value, ConfigHandler.CMDLINE_OPTIONS[name][0])

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.
@@ -976,24 +982,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]:
@@ -1025,7 +1049,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)
@@ -1073,7 +1097,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"
@@ -1085,7 +1109,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::
@@ -1101,6 +1126,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.
         """
@@ -1108,9 +1134,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.
@@ -1155,21 +1183,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,
@@ -1177,20 +1217,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:

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:

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(',')]
 

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()

patroni/ha.py

@@ -924,13 +924,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:

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
 
@@ -179,6 +179,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.
@@ -213,8 +218,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':
@@ -439,18 +445,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:
@@ -461,7 +464,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
@@ -573,7 +576,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]:
@@ -1028,7 +1031,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'),

patroni/postgresql/citus.py

@@ -397,7 +397,7 @@ class CitusHandler(Thread):
         parameters['shared_preload_libraries'] = ','.join(['citus'] + shared_preload_libraries)
 
         # if not explicitly set Citus overrides max_prepared_transactions to max_connections*2
-        if parameters.get('max_prepared_transactions') == 0:
+        if parameters['max_prepared_transactions'] == 0:
             parameters['max_prepared_transactions'] = parameters['max_connections'] * 2
 
         # Resharding in Citus implemented using logical replication

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 \

patroni/postgresql/rewind.py

@@ -158,7 +158,7 @@ class Rewind(object):
     def _get_local_timeline_lsn(self) -> Tuple[Optional[bool], Optional[int], Optional[int]]:
         if self._postgresql.is_running():  # if postgres is running - get timeline from replication connection
             in_recovery = True
-            timeline = self._postgresql.received_timeline() or self._postgresql.get_replica_timeline()
+            timeline = self._postgresql.get_replica_timeline()
             lsn = self._postgresql.replayed_location()
         else:  # otherwise analyze pg_controldata output
             in_recovery, timeline, lsn = self._get_local_timeline_lsn_from_controldata()

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.

patroni/version.py

@@ -2,4 +2,4 @@
 
 :var __version__: the current Patroni version.
 """
-__version__ = '3.1.0'
+__version__ = '3.1.1'

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:[email protected]:5435/postgres'})
+        self.leadermem = Member(0, 'leader', 28, {'xlog_location': 100, 'state': 'running',
+                                                  'conn_url': 'postgres://replicator:[email protected]:5435/postgres'})
         self.leader = Leader(-1, 28, self.leadermem)
         self.other = Member(0, 'test-1', 28, {'conn_url': 'postgres://replicator:[email protected]:5433/postgres',
                                               'state': 'running', 'tags': {'replicatefrom': 'leader'}})

tests/test_bootstrap.py

@@ -238,7 +238,8 @@ class TestBootstrap(BaseTestPostgresql):
         self.p.reload_config({'authentication': {'superuser': {'username': 'p', 'password': 'p'},
                                                  'replication': {'username': 'r', 'password': 'r'},
                                                  'rewind': {'username': 'rw', 'password': 'rw'}},
-                              'listen': '*', 'retry_timeout': 10, 'parameters': {'wal_level': '', 'hba_file': 'foo'}})
+                              'listen': '*', 'retry_timeout': 10,
+                              'parameters': {'wal_level': '', 'hba_file': 'foo', 'max_prepared_transactions': 10}})
         with patch.object(Postgresql, 'major_version', PropertyMock(return_value=110000)), \
                 patch.object(Postgresql, 'restart', Mock()) as mock_restart:
             self.b.post_bootstrap({}, task)

tests/test_config.py

@@ -3,6 +3,7 @@ import sys
 import unittest
 import io
 
+from copy import deepcopy
 from mock import MagicMock, Mock, patch
 from patroni.config import Config, ConfigParseError
 
@@ -22,7 +23,7 @@ class TestConfig(unittest.TestCase):
             self.assertFalse(self.config.set_dynamic_configuration({'foo': 'bar'}))
         self.assertTrue(self.config.set_dynamic_configuration({'standby_cluster': {}, 'postgresql': {
             'parameters': {'cluster_name': 1, 'hot_standby': 1, 'wal_keep_size': 1,
-                           'track_commit_timestamp': 1, 'wal_level': 1}}}))
+                           'track_commit_timestamp': 1, 'wal_level': 1, 'max_connections': '100'}}}))
 
     def test_reload_local_configuration(self):
         os.environ.update({
@@ -149,3 +150,25 @@ class TestConfig(unittest.TestCase):
     @patch('os.path.isdir', Mock(return_value=False))
     def test_invalid_path(self):
         self.assertRaises(ConfigParseError, Config, 'postgres0')
+
+    def test__process_postgresql_parameters(self):
+        expected_params = {
+            'f.oo': 'bar',  # not in ConfigHandler.CMDLINE_OPTIONS
+            'max_connections': 100,  # IntValidator
+            'wal_level': 'hot_standby',  # EnumValidator
+        }
+        input_params = deepcopy(expected_params)
+
+        input_params['max_connections'] = '100'
+        self.assertEqual(self.config._process_postgresql_parameters(input_params), expected_params)
+
+        expected_params['f.oo'] = input_params['f.oo'] = '100'
+        self.assertEqual(self.config._process_postgresql_parameters(input_params), expected_params)
+
+        input_params['wal_level'] = 'cold_standby'
+        expected_params.pop('wal_level')
+        self.assertEqual(self.config._process_postgresql_parameters(input_params), expected_params)
+
+        input_params['max_connections'] = 10
+        expected_params.pop('max_connections')
+        self.assertEqual(self.config._process_postgresql_parameters(input_params), expected_params)

tests/test_postgresql.py

@@ -688,7 +688,7 @@ class TestPostgresql(BaseTestPostgresql):
             self.assertIsNone(self.p.wait_for_startup())
 
     def test_get_server_parameters(self):
-        config = {'parameters': {'wal_level': 'hot_standby'}, 'listen': '0'}
+        config = {'parameters': {'wal_level': 'hot_standby', 'max_prepared_transactions': 100}, 'listen': '0'}
         self.p._global_config = GlobalConfig({'synchronous_mode': True})
         self.p.config.get_server_parameters(config)
         self.p._global_config = GlobalConfig({'synchronous_mode': True, 'synchronous_mode_strict': True})

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")

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'