Use pg_current_wal_flush_lsn() starting from 9.6 (#2813)

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.

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

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:

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:

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

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)