Repository: zalando/patroni
Pull request #2813 (2023-08-14 08:06:54)

Use pg_current_wal_flush_lsn() starting from 9.6
Due to historical reasons (not available before 9.6) we used `pg_current_wal_lsn()`/`pg_current_xlog_location()` functions to get current WAL LSN on the primary. But, this LSN is not necessarily synced to disk, and could be lost if the primary node crashed.

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

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

---



|  Totals | [![Coverage Status](https://coveralls.io/builds/61961815/badge)](https://coveralls.io/builds/61961815) |
| :-- | --: |
| Change from base [Build 5830551883](https://coveralls.io/builds/61927667): |  0.0% |
| Covered Lines: | 13043 |
| Relevant Lines: | 13061 |

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

hughcapet: 👍 
CyberDem0n: :+1:

Patch:
diff --git a/patroni/api.py b/patroni/api.py
index 8d991d5f..3397d8c3 100644
--- a/patroni/api.py
+++ b/patroni/api.py
@@ -448,7 +448,7 @@ class RestApiHandler(BaseHTTPRequestHandler):
         * ``patroni_postmaster_start_time``: epoch timestamp since Postmaster was started;
         * ``patroni_master``: ``1`` if this node holds the leader lock, else ``0``;
         * ``patroni_primary``: same as ``patroni_master``;
-        * ``patroni_xlog_location``: ``pg_wal_lsn_diff(pg_current_wal_lsn(), '0/0')`` if leader, else ``0``;
+        * ``patroni_xlog_location``: ``pg_wal_lsn_diff(pg_current_wal_flush_lsn(), '0/0')`` if leader, else ``0``;
         * ``patroni_standby_leader``: ``1`` if standby leader node, else ``0``;
         * ``patroni_replica``: ``1`` if a replica, else ``0``;
         * ``patroni_sync_standby``: ``1`` if a sync replica, else ``0``;
@@ -1141,7 +1141,7 @@ class RestApiHandler(BaseHTTPRequestHandler):
             * ``server_version``: Postgres version without periods, e.g. ``150002`` for Postgres ``15.2``;
             * ``xlog``: dictionary. Its structure depends on ``role``:
                 * If ``master``:
-                    * ``location``: ``pg_current_wal_lsn()``
+                    * ``location``: ``pg_current_wal_flush_lsn()``
                 * If ``replica``:
                     * ``received_location``: ``pg_wal_lsn_diff(pg_last_wal_receive_lsn(), '0/0')``;
                     * ``replayed_location``: ``pg_wal_lsn_diff(pg_last_wal_replay_lsn(), '0/0)``;
@@ -1179,8 +1179,8 @@ class RestApiHandler(BaseHTTPRequestHandler):
                     " application_name, client_addr, w.state, sync_state, sync_priority"
                     " FROM pg_catalog.pg_stat_get_wal_senders() w, pg_catalog.pg_stat_get_activity(pid)) AS ri")
 
-            row = self.query(stmt.format(postgresql.wal_name, postgresql.lsn_name), retry=retry)[0]
-
+            row = self.query(stmt.format(postgresql.wal_name, postgresql.lsn_name,
+                                         postgresql.wal_flush), retry=retry)[0]
             result = {
                 'state': postgresql.state,
                 'postmaster_start_time': row[0],
diff --git a/patroni/postgresql/__init__.py b/patroni/postgresql/__init__.py
index 172e575e..943a5832 100644
--- a/patroni/postgresql/__init__.py
+++ b/patroni/postgresql/__init__.py
@@ -57,8 +57,8 @@ class Postgresql(object):
     TL_LSN = ("CASE WHEN pg_catalog.pg_is_in_recovery() THEN 0 "
               "ELSE ('x' || pg_catalog.substr(pg_catalog.pg_{0}file_name("
               "pg_catalog.pg_current_{0}_{1}()), 1, 8))::bit(32)::int END, "  # primary timeline
-              "CASE WHEN pg_catalog.pg_is_in_recovery() THEN 0 "
-              "ELSE pg_catalog.pg_{0}_{1}_diff(pg_catalog.pg_current_{0}_{1}(), '0/0')::bigint END, "  # write_lsn
+              "CASE WHEN pg_catalog.pg_is_in_recovery() THEN 0 ELSE "
+              "pg_catalog.pg_{0}_{1}_diff(pg_catalog.pg_current_{0}{2}_{1}(), '0/0')::bigint END, "  # wal(_flush)?_lsn
               "pg_catalog.pg_{0}_{1}_diff(pg_catalog.pg_last_{0}_replay_{1}(), '0/0')::bigint, "
               "pg_catalog.pg_{0}_{1}_diff(COALESCE(pg_catalog.pg_last_{0}_receive_{1}(), '0/0'), '0/0')::bigint, "
               "pg_catalog.pg_is_in_recovery() AND pg_catalog.pg_is_{0}_replay_paused()")
@@ -159,6 +159,11 @@ class Postgresql(object):
     def wal_name(self) -> str:
         return 'wal' if self._major_version >= 100000 else 'xlog'
 
+    @property
+    def wal_flush(self) -> str:
+        """For PostgreSQL 9.6 onwards we want to use pg_current_wal_flush_lsn()/pg_current_xlog_flush_location()."""
+        return '_flush' if self._major_version >= 90600 else ''
+
     @property
     def lsn_name(self) -> str:
         return 'lsn' if self._major_version >= 100000 else 'location'
@@ -211,7 +216,7 @@ class Postgresql(object):
         else:
             extra = "0, NULL, NULL, NULL, NULL, NULL, NULL" + extra
 
-        return ("SELECT " + self.TL_LSN + ", {2}").format(self.wal_name, self.lsn_name, extra)
+        return ("SELECT " + self.TL_LSN + ", {3}").format(self.wal_name, self.lsn_name, self.wal_flush, extra)
 
     @property
     def available_gucs(self) -> CaseInsensitiveSet:
diff --git a/patroni/postgresql/sync.py b/patroni/postgresql/sync.py
index 3193a7e7..97a6fd46 100644
--- a/patroni/postgresql/sync.py
+++ b/patroni/postgresql/sync.py
@@ -182,7 +182,7 @@ class _ReplicaList(List[_Replica]):
     swapping, but only if lag on this member is exceeding a threshold (``maximum_lag_on_syncnode``).
 
     :ivar max_lsn: maximum value of ``_Replica.lsn`` among all values. In case if there is just one
-                   element in the list we take value of ``pg_current_wal_lsn()``.
+                   element in the list we take value of ``pg_current_wal_flush_lsn()``.
     """
 
     def __init__(self, postgresql: 'Postgresql', cluster: Cluster) -> None:


Test patch:
diff --git a/tests/test_api.py b/tests/test_api.py
index eb19ebca..b5aeff8a 100644
--- a/tests/test_api.py
+++ b/tests/test_api.py
@@ -36,6 +36,7 @@ class MockPostgresql(object):
     pending_restart = True
     wal_name = 'wal'
     lsn_name = 'lsn'
+    wal_flush = '_flush'
     POSTMASTER_START_TIME = 'pg_catalog.pg_postmaster_start_time()'
     TL_LSN = 'CASE WHEN pg_catalog.pg_is_in_recovery()'
     citus_handler = Mock()
diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py
index a703f235..4f1e36da 100644
--- a/tests/test_postgresql.py
+++ b/tests/test_postgresql.py
@@ -955,3 +955,10 @@ class TestPostgresql2(BaseTestPostgresql):
         gucs = self.p.available_gucs
         self.assertIsInstance(gucs, CaseInsensitiveSet)
         self.assertEqual(gucs, mock_available_gucs.return_value)
+
+    def test_cluster_info_query(self):
+        self.assertIn('diff(pg_catalog.pg_current_wal_flush_lsn(', self.p.cluster_info_query)
+        self.p._major_version = 90600
+        self.assertIn('diff(pg_catalog.pg_current_xlog_flush_location(', self.p.cluster_info_query)
+        self.p._major_version = 90500
+        self.assertIn('diff(pg_catalog.pg_current_xlog_location(', self.p.cluster_info_query)