Thank you for your interest to the patch.
I understand you questions, but I fully support Alexander Korotkov idea
to commit the minimal required functionality. And then keep working on
other improvements.
On 2024-03-24 05:39, Bharath Rupireddy wrote:
On Fri, Mar 22, 2024 at 4:28 AM Peter Eisentraut <pe...@eisentraut.org>
wrote:
I had written in [0] about my questions related to using this with
connection poolers. I don't think this was addressed at all. I
haven't
seen any discussion about how to make this kind of facility usable in
a
full system. You have to manually query and send LSNs; that seems
pretty cumbersome. Sure, this is part of something that could be
useful, but how would an actual user with actual application code get
to
use this?
[0]:
https://www.postgresql.org/message-id/8b5b172f-0ae7-d644-8358-e2851dded43b%40enterprisedb.com
But I wonder how a client is going to get the LSN. How would all of
this be used by a client? I can think of a scenarios where you have
an application that issues a bunch of SQL commands and you have some
kind of pooler in the middle that redirects those commands to
different hosts, and what you really want is to have it transparently
behave as if it's just a single host. Do we want to inject a bunch
of "SELECT pg_get_lsn()", "SELECT pg_wait_lsn()" calls into that?
As I understand your question, application make dml on the primary
server, get LSN of changes and send bunch SQL read-only commands to
pooler. Transparent behave we can get using #synchronous_commit, but
it is very slow.
I'm tempted to think this could be a protocol-layer facility. Every
query automatically returns the current LSN, and every query can also
send along an LSN to wait for, and the client library would just keep
track of the LSN for (what it thinks of as) the connection. So you
get some automatic serialization without having to modify your client
code.
Thank you, it is a good question for future versions.
You say about a protocol-layer facility, what you meen. May be we can
use signals, like hot_standby_feedback.
I share the same concern as yours and had proposed something upthread
[1]. The idea is something like how each query takes a snapshot at the
beginning of txn/query (depending on isolation level), the same way
the standby can wait for the primary's current LSN as of the moment
(at the time of taking snapshot). And, primary keeps sending its
current LSN as part of regular WAL to standbys so that the standbys
doesn't have to make connections to the primary to know its current
LSN every time. Perhps, this may not even fully guarantee (considered
to be achieving) the read-after-write consistency on standbys unless
there's a way for the application to tell the wait LSN.
Thoughts?
[1]
https://www.postgresql.org/message-id/CALj2ACUfS7LH1PaWmSZ5KwH4BpQxO9izeMw4qC3a1DAwi6nfbQ%40mail.gmail.com
+1 to have support for implicit txns. A strawman solution I can think
of is to let primary send its current insert LSN to the standby every
time it sends a bunch of WAL, and the standby waits for that LSN to be
replayed on it at the start of every implicit txn automatically.
And how standby will get lsn to wait for? All solutions I can think of
are very invasive and poorly scalable.
For example, every dml can send back LSN if dml is success. And
application could use it to wait actual changes.
The new BEGIN syntax requires application code changes. This led me to
think how one can achieve read-after-write consistency today in a
primary - standby set up. All the logic of this patch, that is, waiting
for the standby to pass a given primary LSN needs to be done in the
application code (or in proxy or in load balancer?). I believe there
might be someone doing this already, it's good to hear from them.
You may use #synchronous_commit mode but it slow. So my implementation
don`t make primary to wait all standby to sent its feedbacks.
--
Ivan Kartyshov
Postgres Professional: www.postgrespro.com
diff --git a/doc/src/sgml/func.sgml b/doc/src/sgml/func.sgml
index 8ecc02f2b9..8042201bab 100644
--- a/doc/src/sgml/func.sgml
+++ b/doc/src/sgml/func.sgml
@@ -28086,10 +28086,55 @@ postgres=# SELECT '0/0'::pg_lsn + pd.segment_number * ps.setting::int + :offset
extension.
</para></entry>
</row>
+
+ <row>
+ <entry role="func_table_entry"><para role="func_signature">
+ <indexterm>
+ <primary>pg_wait_lsn</primary>
+ </indexterm>
+ <function>pg_wait_lsn</function> (trg_lsn pg_lsn, delay int8 DEFAULT 0)
+ </para>
+ <para>
+ Returns ERROR if target LSN was not replayed on standby.
+ Parameter <parameter>delay</parameter> sets seconds, the time to wait to the LSN.
+ </para></entry>
+ </row>
</tbody>
</tgroup>
</table>
+ <para>
+ <function>pg_wait_lsn</function> waits till <parameter>wait_lsn</parameter>
+ to be replayed on standby to achieve read-your-writes-consistency, while
+ using async replica for reads and primary for writes. In that case lsn of
+ last modification is stored inside application.
+ Note: pg_wait_lsn(trg_lsn pg_lsn, delay int8 DEFAULT 0)
+ </para>
+
+ <para>
+ You can use <function>pg_wait_lsn</function> to wait the <type>pg_lsn</type>
+ value. For example:
+ <programlisting>
+ Replica:
+ postgresql.conf
+ recovery_min_apply_delay = 10s;
+
+ Primary: Application update table and get lsn of changes
+ postgres=# UPDATE films SET kind = 'Dramatic' WHERE kind = 'Drama';
+ postgres=# SELECT pg_current_wal_lsn();
+ pg_current_wal_lsn
+ --------------------
+ 0/306EE20
+ (1 row)
+
+ Replica: Wait and read updated data
+ postgres=# CALL pg_wait_lsn('0/306EE20', 100); // Timeout 100ms is insufficent
+ ERROR: canceling waiting for LSN due to timeout
+ postgres=# CALL pg_wait_lsn('0/306EE20');
+ postgres=# SELECT * FROM films WHERE kind = 'Drama';
+ </programlisting>
+ </para>
+
<para>
The functions shown in <xref
linkend="functions-recovery-control-table"/> control the progress of recovery.
diff --git a/src/backend/access/transam/xlogrecovery.c b/src/backend/access/transam/xlogrecovery.c
index 29c5bec084..0b783dc733 100644
--- a/src/backend/access/transam/xlogrecovery.c
+++ b/src/backend/access/transam/xlogrecovery.c
@@ -43,6 +43,7 @@
#include "backup/basebackup.h"
#include "catalog/pg_control.h"
#include "commands/tablespace.h"
+#include "commands/waitlsn.h"
#include "common/file_utils.h"
#include "miscadmin.h"
#include "pgstat.h"
@@ -1828,6 +1829,14 @@ PerformWalRecovery(void)
break;
}
+ /*
+ * If we replayed an LSN that someone was waiting for, set latches
+ * in shared memory array to notify the waiter.
+ */
+ if (waitLSN &&
+ (XLogRecoveryCtl->lastReplayedEndRecPtr >= pg_atomic_read_u64(&waitLSN->minLSN)))
+ WaitLSNSetLatches(XLogRecoveryCtl->lastReplayedEndRecPtr);
+
/* Else, try to fetch the next WAL record */
record = ReadRecord(xlogprefetcher, LOG, false, replayTLI);
} while (record != NULL);
diff --git a/src/backend/catalog/system_functions.sql b/src/backend/catalog/system_functions.sql
index fe2bb50f46..5c4f9c78a1 100644
--- a/src/backend/catalog/system_functions.sql
+++ b/src/backend/catalog/system_functions.sql
@@ -414,6 +414,9 @@ CREATE OR REPLACE FUNCTION
json_populate_recordset(base anyelement, from_json json, use_json_as_text boolean DEFAULT false)
RETURNS SETOF anyelement LANGUAGE internal STABLE ROWS 100 AS 'json_populate_recordset' PARALLEL SAFE;
+CREATE OR REPLACE PROCEDURE pg_wait_lsn(trg_lsn pg_lsn, delay float8 DEFAULT 0)
+ LANGUAGE internal AS 'pg_wait_lsn';
+
CREATE OR REPLACE FUNCTION pg_logical_slot_get_changes(
IN slot_name name, IN upto_lsn pg_lsn, IN upto_nchanges int, VARIADIC options text[] DEFAULT '{}',
OUT lsn pg_lsn, OUT xid xid, OUT data text)
diff --git a/src/backend/commands/Makefile b/src/backend/commands/Makefile
index 48f7348f91..cede90c3b9 100644
--- a/src/backend/commands/Makefile
+++ b/src/backend/commands/Makefile
@@ -61,6 +61,7 @@ OBJS = \
vacuum.o \
vacuumparallel.o \
variable.o \
- view.o
+ view.o \
+ waitlsn.o
include $(top_srcdir)/src/backend/common.mk
diff --git a/src/backend/commands/meson.build b/src/backend/commands/meson.build
index 6dd00a4abd..7549be5dc3 100644
--- a/src/backend/commands/meson.build
+++ b/src/backend/commands/meson.build
@@ -50,4 +50,5 @@ backend_sources += files(
'vacuumparallel.c',
'variable.c',
'view.c',
+ 'waitlsn.c',
)
diff --git a/src/backend/commands/waitlsn.c b/src/backend/commands/waitlsn.c
new file mode 100644
index 0000000000..bd9ea6b7db
--- /dev/null
+++ b/src/backend/commands/waitlsn.c
@@ -0,0 +1,303 @@
+/*-------------------------------------------------------------------------
+ *
+ * waitlsn.c
+ * Implements waiting for the given LSN, which is used in
+ * CALL pg_wait_lsn(wait_lsn pg_lsn, timeout int).
+ *
+ * Copyright (c) 2024, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION
+ * src/backend/commands/waitlsn.c
+ *
+ *-------------------------------------------------------------------------
+ */
+
+#include "postgres.h"
+
+#include <float.h>
+#include <math.h>
+
+#include "pgstat.h"
+#include "fmgr.h"
+#include "access/transam.h"
+#include "access/xact.h"
+#include "access/xlog.h"
+#include "access/xlogdefs.h"
+#include "access/xlogrecovery.h"
+#include "catalog/pg_type.h"
+#include "commands/waitlsn.h"
+#include "executor/spi.h"
+#include "funcapi.h"
+#include "miscadmin.h"
+#include "storage/ipc.h"
+#include "storage/latch.h"
+#include "storage/pmsignal.h"
+#include "storage/proc.h"
+#include "storage/shmem.h"
+#include "storage/sinvaladt.h"
+#include "utils/builtins.h"
+#include "utils/pg_lsn.h"
+#include "utils/snapmgr.h"
+#include "utils/timestamp.h"
+#include "utils/fmgrprotos.h"
+
+/* Add to / delete from shared memory array */
+static void addLSNWaiter(XLogRecPtr lsn);
+static void deleteLSNWaiter(void);
+
+struct WaitLSNState *waitLSN = NULL;
+static volatile sig_atomic_t haveShmemItem = false;
+
+/*
+ * Report the amount of shared memory space needed for WaitLSNState
+ */
+Size
+WaitLSNShmemSize(void)
+{
+ Size size;
+
+ size = offsetof(WaitLSNState, procInfos);
+ size = add_size(size, mul_size(MaxBackends, sizeof(WaitLSNProcInfo)));
+ return size;
+}
+
+/* Initialize the WaitLSNState in the shared memory */
+void
+WaitLSNShmemInit(void)
+{
+ bool found;
+
+ waitLSN = (WaitLSNState *) ShmemInitStruct("WaitLSNState",
+ WaitLSNShmemSize(),
+ &found);
+ if (!found)
+ {
+ SpinLockInit(&waitLSN->mutex);
+ waitLSN->numWaitedProcs = 0;
+ pg_atomic_init_u64(&waitLSN->minLSN, PG_UINT64_MAX);
+ }
+}
+
+/*
+ * Add the information about the LSN waiter backend to the shared memory
+ * array.
+ */
+static void
+addLSNWaiter(XLogRecPtr lsn)
+{
+ WaitLSNProcInfo cur;
+ int i;
+
+ SpinLockAcquire(&waitLSN->mutex);
+
+ cur.procnum = MyProcNumber;
+ cur.waitLSN = lsn;
+
+ for (i = 0; i < waitLSN->numWaitedProcs; i++)
+ {
+ if (waitLSN->procInfos[i].waitLSN >= cur.waitLSN)
+ {
+ WaitLSNProcInfo tmp;
+
+ tmp = waitLSN->procInfos[i];
+ waitLSN->procInfos[i] = cur;
+ cur = tmp;
+ }
+ }
+ waitLSN->procInfos[i] = cur;
+ waitLSN->numWaitedProcs++;
+
+ pg_atomic_write_u64(&waitLSN->minLSN, waitLSN->procInfos[i].waitLSN);
+ SpinLockRelease(&waitLSN->mutex);
+}
+
+/*
+ * Delete the information about the LSN waiter backend from the shared memory
+ * array.
+ */
+static void
+deleteLSNWaiter(void)
+{
+ int i;
+ bool found = false;
+
+ SpinLockAcquire(&waitLSN->mutex);
+
+ for (i = 0; i < waitLSN->numWaitedProcs; i++)
+ {
+ if (waitLSN->procInfos[i].procnum == MyProcNumber)
+ found = true;
+
+ if (found && i < waitLSN->numWaitedProcs - 1)
+ {
+ waitLSN->procInfos[i] = waitLSN->procInfos[i + 1];
+ }
+ }
+
+ if (!found)
+ {
+ SpinLockRelease(&waitLSN->mutex);
+ return;
+ }
+ waitLSN->numWaitedProcs--;
+
+ if (waitLSN->numWaitedProcs != 0)
+ pg_atomic_write_u64(&waitLSN->minLSN, waitLSN->procInfos[i].waitLSN);
+ else
+ pg_atomic_write_u64(&waitLSN->minLSN, PG_UINT64_MAX);
+
+ SpinLockRelease(&waitLSN->mutex);
+}
+
+/*
+ * Set all latches in shared memory to signal that new LSN has been replayed
+*/
+void
+WaitLSNSetLatches(XLogRecPtr curLSN)
+{
+ uint32 i,
+ numWakeUpProcs;
+
+ SpinLockAcquire(&waitLSN->mutex);
+
+ /*
+ * Set latches for process, whose waited LSNs are already replayed.
+ */
+ for (i = 0; i < waitLSN->numWaitedProcs; i++)
+ {
+ PGPROC *backend;
+
+ if (waitLSN->procInfos[i].waitLSN > curLSN)
+ break;
+
+ backend = GetPGProcByNumber(waitLSN->procInfos[i].procnum);
+ SetLatch(&backend->procLatch);
+ }
+
+ /*
+ * Immediately remove those processes from the shmem array. Otherwise,
+ * shmem array items will be here till corresponding processes wake up and
+ * delete themselves.
+ */
+ numWakeUpProcs = i;
+ for (i = 0; i < waitLSN->numWaitedProcs - numWakeUpProcs; i++)
+ waitLSN->procInfos[i] = waitLSN->procInfos[i + numWakeUpProcs];
+ waitLSN->numWaitedProcs -= numWakeUpProcs;
+
+ if (waitLSN->numWaitedProcs != 0)
+ pg_atomic_write_u64(&waitLSN->minLSN, waitLSN->procInfos[i].waitLSN);
+ else
+ pg_atomic_write_u64(&waitLSN->minLSN, PG_UINT64_MAX);
+
+ SpinLockRelease(&waitLSN->mutex);
+}
+
+/*
+ * Delete our item from shmem array if any.
+ */
+void
+WaitLSNCleanup(void)
+{
+ if (haveShmemItem)
+ deleteLSNWaiter();
+}
+
+/*
+ * Wait using MyLatch to wait till the given LSN is replayed, the postmaster dies or
+ * timeout happens.
+ */
+void
+WaitForLSN(XLogRecPtr lsn, float8 sec)
+{
+ XLogRecPtr curLSN;
+ int latch_events;
+ TimestampTz endtime;
+
+ /* Shouldn't be called when shmem isn't initialized */
+ Assert(waitLSN);
+
+ /* Should be only called by a backend */
+ Assert(MyBackendType == B_BACKEND);
+
+ if (!RecoveryInProgress())
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("recovery is not in progress"),
+ errhint("Waiting for LSN can only be executed during recovery.")));
+
+ endtime = GetCurrentTimestamp() + (int64)(sec * 1000 * 1000);
+
+ latch_events = WL_TIMEOUT | WL_LATCH_SET | WL_EXIT_ON_PM_DEATH;
+ addLSNWaiter(lsn);
+ haveShmemItem = true;
+
+ for (;;)
+ {
+ int rc;
+ long delay_ms;
+
+ /* Check if the waited LSN has been replayed */
+ curLSN = GetXLogReplayRecPtr(NULL);
+ if (lsn <= curLSN)
+ break;
+
+ if (sec >= 1)
+ delay_ms = (endtime - GetCurrentTimestamp()) / 1000;
+ else
+ /* If no timeout is set then wake up in 1 minute for interrupts */
+ delay_ms = 60000;
+
+ if (delay_ms <= 0)
+ break;
+
+ /*
+ * If received an interruption from CHECK_FOR_INTERRUPTS, then delete
+ * the current event from array.
+ */
+ CHECK_FOR_INTERRUPTS();
+
+ /* If postmaster dies, finish immediately */
+ if (!PostmasterIsAlive())
+ break;
+
+ rc = WaitLatch(MyLatch, latch_events, delay_ms,
+ WAIT_EVENT_WAIT_FOR_STANDBY_CONFIRMATION);
+
+ if (rc & WL_LATCH_SET)
+ ResetLatch(MyLatch);
+ }
+
+ if (lsn > curLSN)
+ {
+ deleteLSNWaiter();
+ haveShmemItem = false;
+ ereport(ERROR,
+ (errcode(ERRCODE_QUERY_CANCELED),
+ errmsg("canceling waiting for LSN due to timeout")));
+ }
+ else
+ {
+ haveShmemItem = false;
+ }
+}
+
+Datum
+pg_wait_lsn(PG_FUNCTION_ARGS)
+{
+ XLogRecPtr trg_lsn = PG_GETARG_LSN(0);
+ float8 delay = PG_GETARG_FLOAT8(1);
+ CallContext *context = (CallContext *) fcinfo->context;
+
+ if (context->atomic)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("pg_wait_lsn() must be only called in non-atomic context")));
+
+ if (ActiveSnapshotSet())
+ PopActiveSnapshot();
+ Assert(!ActiveSnapshotSet());
+
+ (void) WaitForLSN(trg_lsn, delay);
+
+ PG_RETURN_VOID();
+}
diff --git a/src/backend/storage/ipc/ipci.c b/src/backend/storage/ipc/ipci.c
index 521ed5418c..5aed90c935 100644
--- a/src/backend/storage/ipc/ipci.c
+++ b/src/backend/storage/ipc/ipci.c
@@ -25,6 +25,7 @@
#include "access/xlogprefetcher.h"
#include "access/xlogrecovery.h"
#include "commands/async.h"
+#include "commands/waitlsn.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "postmaster/autovacuum.h"
@@ -152,6 +153,7 @@ CalculateShmemSize(int *num_semaphores)
size = add_size(size, WaitEventExtensionShmemSize());
size = add_size(size, InjectionPointShmemSize());
size = add_size(size, SlotSyncShmemSize());
+ size = add_size(size, WaitLSNShmemSize());
#ifdef EXEC_BACKEND
size = add_size(size, ShmemBackendArraySize());
#endif
@@ -244,6 +246,11 @@ CreateSharedMemoryAndSemaphores(void)
/* Initialize subsystems */
CreateOrAttachShmemStructs();
+ /*
+ * Init array of Latches in shared memory for wait lsn
+ */
+ WaitLSNShmemInit();
+
#ifdef EXEC_BACKEND
/*
diff --git a/src/backend/storage/lmgr/proc.c b/src/backend/storage/lmgr/proc.c
index 162b1f919d..4b830dc3c8 100644
--- a/src/backend/storage/lmgr/proc.c
+++ b/src/backend/storage/lmgr/proc.c
@@ -36,6 +36,7 @@
#include "access/transam.h"
#include "access/twophase.h"
#include "access/xlogutils.h"
+#include "commands/waitlsn.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "postmaster/autovacuum.h"
@@ -862,6 +863,11 @@ ProcKill(int code, Datum arg)
*/
LWLockReleaseAll();
+ /*
+ * Cleanup waiting for LSN if any.
+ */
+ WaitLSNCleanup();
+
/* Cancel any pending condition variable sleep, too */
ConditionVariableCancelSleep();
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index 0d26e5b422..cb325b3be0 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -12138,6 +12138,11 @@
prorettype => 'bytea', proargtypes => 'pg_brin_minmax_multi_summary',
prosrc => 'brin_minmax_multi_summary_send' },
+{ oid => '16387', descr => 'wait for LSN until timeout',
+ proname => 'pg_wait_lsn', prokind => 'p', prorettype => 'void',
+ proargtypes => 'pg_lsn float8', proargnames => '{trg_lsn,delay}',
+ prosrc => 'pg_wait_lsn' },
+
{ oid => '6291', descr => 'arbitrary value from among input values',
proname => 'any_value', prokind => 'a', proisstrict => 'f',
prorettype => 'anyelement', proargtypes => 'anyelement',
diff --git a/src/include/commands/waitlsn.h b/src/include/commands/waitlsn.h
new file mode 100644
index 0000000000..53ab9b7b6b
--- /dev/null
+++ b/src/include/commands/waitlsn.h
@@ -0,0 +1,43 @@
+/*-------------------------------------------------------------------------
+ *
+ * waitlsn.h
+ * Declarations for LSN waiting routines.
+ *
+ * Copyright (c) 2024, PostgreSQL Global Development Group
+ *
+ * src/include/commands/waitlsn.h
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef WAIT_LSN_H
+#define WAIT_LSN_H
+
+#include "postgres.h"
+#include "port/atomics.h"
+#include "storage/spin.h"
+#include "tcop/dest.h"
+
+/* Shared memory structure */
+typedef struct WaitLSNProcInfo
+{
+ int procnum;
+ XLogRecPtr waitLSN;
+} WaitLSNProcInfo;
+
+typedef struct WaitLSNState
+{
+ pg_atomic_uint64 minLSN;
+ slock_t mutex;
+ int numWaitedProcs;
+ WaitLSNProcInfo procInfos[FLEXIBLE_ARRAY_MEMBER];
+} WaitLSNState;
+
+extern PGDLLIMPORT struct WaitLSNState *waitLSN;
+
+extern void WaitForLSN(XLogRecPtr lsn, float8 sec);
+extern Size WaitLSNShmemSize(void);
+extern void WaitLSNShmemInit(void);
+extern void WaitLSNSetLatches(XLogRecPtr curLSN);
+extern void WaitLSNCleanup(void);
+
+#endif /* WAIT_LSN_H */
diff --git a/src/test/recovery/meson.build b/src/test/recovery/meson.build
index b1eb77b1ec..bc47c93902 100644
--- a/src/test/recovery/meson.build
+++ b/src/test/recovery/meson.build
@@ -51,6 +51,7 @@ tests += {
't/040_standby_failover_slots_sync.pl',
't/041_checkpoint_at_promote.pl',
't/042_low_level_backup.pl',
+ 't/043_wait_lsn.pl',
],
},
}
diff --git a/src/test/recovery/t/043_wait_lsn.pl b/src/test/recovery/t/043_wait_lsn.pl
new file mode 100644
index 0000000000..1875702c93
--- /dev/null
+++ b/src/test/recovery/t/043_wait_lsn.pl
@@ -0,0 +1,77 @@
+# Checks waiting for lsn on standby pg_wait_lsn()
+use strict;
+use warnings;
+
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+# Initialize primary node
+my $node_primary = PostgreSQL::Test::Cluster->new('primary');
+$node_primary->init(allows_streaming => 1);
+$node_primary->start;
+
+# And some content and take a backup
+$node_primary->safe_psql('postgres',
+ "CREATE TABLE wait_test AS SELECT generate_series(1,10) AS a");
+my $backup_name = 'my_backup';
+$node_primary->backup($backup_name);
+
+# Create a streaming standby with a 1 second delay from the backup
+my $node_standby = PostgreSQL::Test::Cluster->new('standby');
+my $delay = 3;
+$node_standby->init_from_backup($node_primary, $backup_name,
+ has_streaming => 1);
+$node_standby->append_conf(
+ 'postgresql.conf', qq[
+ recovery_min_apply_delay = '${delay}s'
+]);
+$node_standby->start;
+
+
+# Make sure that pg_wait_lsn() works: add new content to primary and memorize
+# primary's new LSN, then wait for primary's LSN on standby. Prove that pg_wait_lsn() is
+# able to setup an infinite waiting loop and exit it if given no wait timeout.
+$node_primary->safe_psql('postgres',
+ "INSERT INTO wait_test VALUES (generate_series(11, 20))");
+my $lsn1 =
+ $node_primary->safe_psql('postgres', "SELECT pg_current_wal_lsn()");
+my $output = $node_standby->safe_psql(
+ 'postgres', qq[
+ CALL pg_wait_lsn('${lsn1}', 1000000);
+ SELECT pg_lsn_cmp(pg_last_wal_replay_lsn(), '${lsn1}'::pg_lsn);
+]);
+
+# Get the current LSN on standby and make sure it's the same as primary's LSN
+ok($output eq 0, "standby reached the same LSN as primary pg_wait_lsn()");
+
+my $lsn2 =
+ $node_primary->safe_psql('postgres', "SELECT pg_current_wal_lsn() + 1");
+my $stderr;
+$node_standby->safe_psql('postgres', "CALL pg_wait_lsn('${lsn1}', 1);");
+$node_standby->psql(
+ 'postgres',
+ "CALL pg_wait_lsn('${lsn2}', 1);",
+ stderr => \$stderr);
+ok( $stderr =~ /canceling waiting for LSN due to timeout/,
+ "get timeout on waiting for unreachable LSN");
+
+# Make sure that pg_wait_lsn() works: add new content to primary and memorize
+# primary's new LSN, then wait for primary's LSN on standby. Prove that pg_wait_lsn() is
+# able to setup an infinite waiting loop and exit it if current LSN replayed.
+$node_primary->safe_psql('postgres',
+ "INSERT INTO wait_test VALUES (generate_series(21, 30))");
+my $lsn3 =
+ $node_primary->safe_psql('postgres', "SELECT pg_current_wal_lsn()");
+$output = $node_standby->safe_psql(
+ 'postgres', qq[
+ CALL pg_wait_lsn('${lsn3}');
+ SELECT pg_lsn_cmp(pg_last_wal_replay_lsn(), '${lsn3}'::pg_lsn);
+]);
+
+# Get the current LSN on standby and make sure it's the same as primary's LSN
+ok($output eq 0, "standby reached the same LSN as primary");
+
+$node_standby->stop;
+$node_primary->stop;
+done_testing();
