Add wait_for_replay_catchup wrapper to Cluster.pm

This simplifies a few lines of Perl test code a bit.

Author: Bertrand Drouvot
Discussion: https://postgr.es/m/846724b5-0723-f4c2-8b13-75301ec7509e@gmail.com

src/bin/pg_rewind/t/007_standby_source.pl

@@ -160,7 +160,7 @@ in A, after C was promoted
 $node_a->safe_psql('postgres',
 	"INSERT INTO tbl1 values ('in A, after rewind')");
 
-$node_b->wait_for_catchup('node_c', 'replay', $node_a->lsn('write'));
+$node_b->wait_for_replay_catchup('node_c', $node_a);
 
 check_query(
 	'SELECT * FROM tbl1',

src/test/modules/brin/t/02_wal_consistency.pl

@@ -70,6 +70,6 @@ my ($ret, $out, $err) = $whiskey->psql(
 	});
 cmp_ok($out, '>=', 1);
 
-$whiskey->wait_for_catchup($charlie, 'replay', $whiskey->lsn('insert'));
+$whiskey->wait_for_replay_catchup($charlie);
 
 done_testing();

src/test/perl/PostgreSQL/Test/Cluster.pm

@@ -2711,6 +2711,29 @@ sub wait_for_catchup
 
 =pod
 
+=item $node->wait_for_replay_catchup($standby_name [, $base_node ])
+
+Wait for the replication connection with application_name I<$standby_name>
+until its B<replay> replication column in pg_stat_replication in I<$node>
+equals or passes the I<$base_node>'s B<replay_lsn>. If I<$base_node> is
+omitted, the LSN to wait for is obtained from I<$node>.
+
+The replication connection must be in a streaming state.
+
+Requires that the 'postgres' db exists and is accessible.
+
+This is not a test. It die()s on failure.
+
+=cut
+
+sub wait_for_replay_catchup
+{
+	my ($self, $standby_name, $node) = @_;
+	$node = defined($node) ? $node : $self;
+
+	$self->wait_for_catchup($standby_name, 'replay', $node->lsn('flush'));
+}
+
 =item $node->wait_for_slot_catchup(slot_name, mode, target_lsn)
 
 Wait for the named replication slot to equal or pass the supplied target_lsn.

src/test/recovery/t/001_stream_rep.pl

@@ -47,9 +47,8 @@ $node_primary->safe_psql('postgres',
 	"CREATE TABLE tab_int AS SELECT generate_series(1,1002) AS a");
 
 # Wait for standbys to catch up
-my $primary_lsn = $node_primary->lsn('write');
-$node_primary->wait_for_catchup($node_standby_1, 'replay', $primary_lsn);
-$node_standby_1->wait_for_catchup($node_standby_2, 'replay', $primary_lsn);
+$node_primary->wait_for_replay_catchup($node_standby_1);
+$node_standby_1->wait_for_replay_catchup($node_standby_2, $node_primary);
 
 my $result =
   $node_standby_1->safe_psql('postgres', "SELECT count(*) FROM tab_int");
@@ -66,9 +65,8 @@ $node_primary->safe_psql('postgres',
 	"CREATE SEQUENCE seq1; SELECT nextval('seq1')");
 
 # Wait for standbys to catch up
-$primary_lsn = $node_primary->lsn('write');
-$node_primary->wait_for_catchup($node_standby_1, 'replay', $primary_lsn);
-$node_standby_1->wait_for_catchup($node_standby_2, 'replay', $primary_lsn);
+$node_primary->wait_for_replay_catchup($node_standby_1);
+$node_standby_1->wait_for_replay_catchup($node_standby_2, $node_primary);
 
 $result = $node_standby_1->safe_psql('postgres', "SELECT * FROM seq1");
 print "standby 1: $result\n";
@@ -372,10 +370,8 @@ sub replay_check
 	my $newval = $node_primary->safe_psql('postgres',
 		'INSERT INTO replayed(val) SELECT coalesce(max(val),0) + 1 AS newval FROM replayed RETURNING val'
 	);
-	my $primary_lsn = $node_primary->lsn('write');
-	$node_primary->wait_for_catchup($node_standby_1, 'replay', $primary_lsn);
-	$node_standby_1->wait_for_catchup($node_standby_2, 'replay',
-		$primary_lsn);
+	$node_primary->wait_for_replay_catchup($node_standby_1);
+	$node_standby_1->wait_for_replay_catchup($node_standby_2, $node_primary);
 
 	$node_standby_1->safe_psql('postgres',
 		qq[SELECT 1 FROM replayed WHERE val = $newval])

src/test/recovery/t/027_stream_regress.pl

@@ -86,8 +86,7 @@ $node_primary->psql('regression',
 	"select setval(seqrelid, nextval(seqrelid)) from pg_sequence");
 
 # Wait for standby to catch up
-$node_primary->wait_for_catchup($node_standby_1, 'replay',
-	$node_primary->lsn('insert'));
+$node_primary->wait_for_replay_catchup($node_standby_1);
 
 # Perform a logical dump of primary and standby, and check that they match
 command_ok(

src/test/recovery/t/030_stats_cleanup_replica.pl

@@ -38,8 +38,7 @@ drop_table_by_oid('postgres', $tableoid);
 drop_function_by_oid('postgres', $funcoid);
 
 $sect = 'post drop';
-my $primary_lsn = $node_primary->lsn('flush');
-$node_primary->wait_for_catchup($node_standby, 'replay', $primary_lsn);
+$node_primary->wait_for_replay_catchup($node_standby);
 test_standby_func_tab_stats_status('postgres',
 	$dboid, $tableoid, $funcoid, 'f');
 
@@ -49,8 +48,7 @@ test_standby_func_tab_stats_status('postgres',
 $sect = "schema creation";
 
 $node_primary->safe_psql('postgres', "CREATE SCHEMA drop_schema_test1");
-$primary_lsn = $node_primary->lsn('flush');
-$node_primary->wait_for_catchup($node_standby, 'replay', $primary_lsn);
+$node_primary->wait_for_replay_catchup($node_standby);
 
 ($dboid, $tableoid, $funcoid) =
   populate_standby_stats('postgres', 'drop_schema_test1');
@@ -61,8 +59,7 @@ $node_primary->safe_psql('postgres', "DROP SCHEMA drop_schema_test1 CASCADE");
 
 $sect = "post schema drop";
 
-$primary_lsn = $node_primary->lsn('flush');
-$node_primary->wait_for_catchup($node_standby, 'replay', $primary_lsn);
+$node_primary->wait_for_replay_catchup($node_standby);
 
 # verify table and function stats removed from standby
 test_standby_func_tab_stats_status('postgres',
@@ -74,8 +71,7 @@ test_standby_func_tab_stats_status('postgres',
 $sect = "createdb";
 
 $node_primary->safe_psql('postgres', "CREATE DATABASE test");
-$primary_lsn = $node_primary->lsn('flush');
-$node_primary->wait_for_catchup($node_standby, 'replay', $primary_lsn);
+$node_primary->wait_for_replay_catchup($node_standby);
 
 ($dboid, $tableoid, $funcoid) = populate_standby_stats('test', 'public');
 
@@ -84,9 +80,8 @@ test_standby_func_tab_stats_status('test', $dboid, $tableoid, $funcoid, 't');
 test_standby_db_stats_status('test', $dboid, 't');
 
 $node_primary->safe_psql('postgres', "DROP DATABASE test");
-$sect        = "post dropdb";
-$primary_lsn = $node_primary->lsn('flush');
-$node_primary->wait_for_catchup($node_standby, 'replay', $primary_lsn);
+$sect = "post dropdb";
+$node_primary->wait_for_replay_catchup($node_standby);
 
 # Test that the stats were cleaned up on standby
 # Note that this connects to 'postgres' but provides the dboid of dropped db
@@ -137,8 +132,7 @@ sub populate_standby_stats
 	$node_primary->safe_psql($connect_db,
 		"CREATE FUNCTION $schema.drop_func_test1() RETURNS VOID AS 'select 2;' LANGUAGE SQL IMMUTABLE"
 	);
-	my $primary_lsn = $node_primary->lsn('flush');
-	$node_primary->wait_for_catchup($node_standby, 'replay', $primary_lsn);
+	$node_primary->wait_for_replay_catchup($node_standby);
 
 	# collect object oids
 	my $dboid = $node_standby->safe_psql($connect_db,

src/test/recovery/t/031_recovery_conflict.pl

@@ -63,8 +63,7 @@ CREATE TABLE ${table1}(a int, b int);
 INSERT INTO $table1 SELECT i % 3, 0 FROM generate_series(1,20) i;
 CREATE TABLE ${table2}(a int, b int);
 ]);
-my $primary_lsn = $node_primary->lsn('flush');
-$node_primary->wait_for_catchup($node_standby, 'replay', $primary_lsn);
+$node_primary->wait_for_replay_catchup($node_standby);
 
 
 # a longrunning psql that we can use to trigger conflicts
@@ -97,8 +96,7 @@ $node_primary->safe_psql(
 	BEGIN; LOCK $table1; COMMIT;
 	]);
 
-$primary_lsn = $node_primary->lsn('flush');
-$node_primary->wait_for_catchup($node_standby, 'replay', $primary_lsn);
+$node_primary->wait_for_replay_catchup($node_standby);
 
 my $cursor1 = "test_recovery_conflict_cursor";
 
@@ -124,8 +122,7 @@ $node_primary->safe_psql($test_db, qq[VACUUM $table1;]);
 # finished, so waiting for catchup ensures that there is no race between
 # encountering the recovery conflict which causes the disconnect and checking
 # the logfile for the terminated connection.
-$primary_lsn = $node_primary->lsn('flush');
-$node_primary->wait_for_catchup($node_standby, 'replay', $primary_lsn);
+$node_primary->wait_for_replay_catchup($node_standby);
 
 check_conflict_log("User was holding shared buffer pin for too long");
 reconnect_and_clear();
@@ -138,8 +135,7 @@ $expected_conflicts++;
 
 $node_primary->safe_psql($test_db,
 	qq[INSERT INTO $table1 SELECT i, 0 FROM generate_series(1,20) i]);
-$primary_lsn = $node_primary->lsn('flush');
-$node_primary->wait_for_catchup($node_standby, 'replay', $primary_lsn);
+$node_primary->wait_for_replay_catchup($node_standby);
 
 # DECLARE and FETCH from cursor on the standby
 $psql_standby{stdin} .= qq[
@@ -160,8 +156,7 @@ $node_primary->safe_psql($test_db,
 $node_primary->safe_psql($test_db, qq[VACUUM $table1;]);
 
 # Wait for attempted replay of PRUNE records
-$primary_lsn = $node_primary->lsn('flush');
-$node_primary->wait_for_catchup($node_standby, 'replay', $primary_lsn);
+$node_primary->wait_for_replay_catchup($node_standby);
 
 check_conflict_log(
 	"User query might have needed to see row versions that must be removed");
@@ -184,8 +179,7 @@ ok(pump_until_standby(qr/^1$/m), "$sect: conflicting lock acquired");
 # DROP TABLE containing block which standby has in a pinned buffer
 $node_primary->safe_psql($test_db, qq[DROP TABLE $table1;]);
 
-$primary_lsn = $node_primary->lsn('flush');
-$node_primary->wait_for_catchup($node_standby, 'replay', $primary_lsn);
+$node_primary->wait_for_replay_catchup($node_standby);
 
 check_conflict_log("User was holding a relation lock for too long");
 reconnect_and_clear();
@@ -213,8 +207,7 @@ ok(pump_until_standby(qr/^6000$/m),
 # standby
 $node_primary->safe_psql($test_db, qq[DROP TABLESPACE $tablespace1;]);
 
-$primary_lsn = $node_primary->lsn('flush');
-$node_primary->wait_for_catchup($node_standby, 'replay', $primary_lsn);
+$node_primary->wait_for_replay_catchup($node_standby);
 
 check_conflict_log(
 	"User was or might have been using tablespace that must be dropped");
@@ -255,8 +248,7 @@ INSERT INTO $table1(a) VALUES (170);
 SELECT txid_current();
 ]);
 
-$primary_lsn = $node_primary->lsn('flush');
-$node_primary->wait_for_catchup($node_standby, 'replay', $primary_lsn);
+$node_primary->wait_for_replay_catchup($node_standby);
 
 $psql_standby{stdin} .= qq[
     BEGIN;
@@ -282,8 +274,7 @@ SELECT 'waiting' FROM pg_locks WHERE locktype = 'relation' AND NOT granted;
 # VACUUM will prune away rows, causing a buffer pin conflict, while standby
 # psql is waiting on lock
 $node_primary->safe_psql($test_db, qq[VACUUM $table1;]);
-$primary_lsn = $node_primary->lsn('flush');
-$node_primary->wait_for_catchup($node_standby, 'replay', $primary_lsn);
+$node_primary->wait_for_replay_catchup($node_standby);
 
 check_conflict_log("User transaction caused buffer deadlock with recovery.");
 reconnect_and_clear();
@@ -311,8 +302,7 @@ $sect = "database conflict";
 
 $node_primary->safe_psql('postgres', qq[DROP DATABASE $test_db;]);
 
-$primary_lsn = $node_primary->lsn('flush');
-$node_primary->wait_for_catchup($node_standby, 'replay', $primary_lsn);
+$node_primary->wait_for_replay_catchup($node_standby);
 
 check_conflict_log("User was connected to a database that must be dropped");
 

src/test/recovery/t/033_replay_tsp_drops.pl

@@ -42,8 +42,7 @@ sub test_tablespace
 	$node_standby->start;
 
 	# Make sure the connection is made
-	$node_primary->wait_for_catchup($node_standby, 'write',
-		$node_primary->lsn('write'));
+	$node_primary->wait_for_catchup($node_standby, 'write');
 
 	# Do immediate shutdown just after a sequence of CREATE DATABASE / DROP
 	# DATABASE / DROP TABLESPACE. This causes CREATE DATABASE WAL records
@@ -65,8 +64,7 @@ sub test_tablespace
 	$query =~ s/<STRATEGY>/$strategy/g;
 
 	$node_primary->safe_psql('postgres', $query);
-	$node_primary->wait_for_catchup($node_standby, 'write',
-		$node_primary->lsn('write'));
+	$node_primary->wait_for_catchup($node_standby, 'write');
 
 	# show "create missing directory" log message
 	$node_standby->safe_psql('postgres',