Fix several new syncpoint machinery - add/stop node jogs.

1) Encode join_node syncpoints in hex;
2) Set Syncpoint.local_lsn of existing nodes in new node pulling from donor
  (otherwise their changes are skipped);
3) Flush WAL before filter loading (tests demonstrated empty xact doesn't flush
  WAL, so filter skips and we eventually get it twice);
4) Accurately wait for slot creation in 005_add_stop_node.pl tests before
  basebackup.

Author: arssher

Cluster.pm

@@ -1,6 +1,8 @@
 package Cluster;
+
 use strict;
 use warnings;
+
 use PostgresNode;
 use TestLib;
 use Test::More;
@@ -193,6 +195,14 @@ sub safe_psql
 	return $node->safe_psql($node->{dbname}, $query);
 }
 
+sub poll_query_until
+{
+	my ($self, $node_off, $query) = @_;
+	my $node = $self->{nodes}->[$node_off];
+	note("polling query $query");
+	return $node->poll_query_until($node->{dbname}, $query);
+}
+
 sub connstr
 {
 	my ($self, $node_off) = @_;

multimaster--1.0.sql

@@ -280,8 +280,8 @@ ALTER TABLE mtm.syncpoints ENABLE ALWAYS TRIGGER syncpoints_trigger;
  *    just a mark that node have applied everything up to origin_lsn.
  * 2) emitpoint: means all origin_node transactions with start_lsn >= origin_lsn
  *    at receiver have or will have start_lsn >= receiver_lsn. So recovering
- *    nodes should translate origin_lsn they need to the appropriate emitpoint of
- *    donor and request streaming since it.
+ *    node should translate origin_lsn it needs to the appropriate emitpoint of
+ *    donor and request streaming since it (and ack it in its reports).
  * In the current blocking implementation single record represents both
  * absorbpoint and emitpoint. To make future development easier, hints are put
  * in places where the distinction would matter.

src/multimaster.c

@@ -1167,8 +1167,6 @@ mtm_join_node(PG_FUNCTION_ARGS)
 		while (!MtmAllApplyWorkersFinished())
 			MtmSleep(USECS_PER_SEC / 10);
 
-		curr_lsn = GetXLogWriteRecPtr();
-
 		for (i = 0; i < cfg->n_nodes; i++)
 		{
 			char	   *ro_name;
@@ -1183,7 +1181,7 @@ mtm_join_node(PG_FUNCTION_ARGS)
 			ro_id = replorigin_by_name(ro_name, false);
 			olsn = replorigin_get_progress(ro_id, false);
 
-			msg = psprintf("F_%d_" UINT64_FORMAT,
+			msg = psprintf("F_%d_%" INT64_MODIFIER "X",
 						   cfg->nodes[i].node_id, olsn);
 
 			replorigin_session_origin = cfg->nodes[i].origin_id;
@@ -1225,7 +1223,11 @@ mtm_join_node(PG_FUNCTION_ARGS)
 	}
 	ConditionVariableBroadcast(&Mtm->receiver_barrier_cv);
 
-	/* as we going to write that lsn on a new node, let's sync it */
+	/*
+	 * Ensure new node will get 'Z' only after eating all syncpoints written
+	 * above.
+	 */
+	curr_lsn = GetXLogWriteRecPtr();
 	XLogFlush(curr_lsn);
 
 	/* fill MTM_NODES with a adjusted list of nodes */

src/pglogical_receiver.c

@@ -305,7 +305,8 @@ MtmFilterTransaction(char *record, int size, Syncpoint *spvector,
 			default:
 				Assert(false);
 		}
-	} else if (msgtype == 'M')
+	}
+	else if (msgtype == 'M')
 	{
 		char		action = pq_getmsgbyte(&s);
 		int			messageSize;
@@ -671,6 +672,8 @@ pglogical_receiver_main(Datum main_arg)
 			 */
 			if (receiver_mtm_cfg->backup_node_id > 0)
 			{
+				int i;
+
 				spvector = palloc0(MTM_MAX_NODES * sizeof(Syncpoint));
 				/*
 				 * Immediately after basebackup we don't have any
@@ -681,6 +684,15 @@ pglogical_receiver_main(Datum main_arg)
 				 */
 				spvector[sender - 1] = SyncpointGetLatest(sender);
 				remote_start = receiver_mtm_cfg->backup_end_lsn;
+				/*
+				 * all other origins are also filtered from filter position
+				 * to donor
+				 */
+				for (i = 0; i < receiver_mtm_cfg->n_nodes; i++)
+				{
+					spvector[receiver_mtm_cfg->nodes[i].node_id - 1].local_lsn =
+						spvector[sender - 1].local_lsn;
+				}
 				/*
 				 * we must filter out all xacts since sp to donor in this
 				 * special add_node case, so MtmInvalidNodeId
@@ -721,7 +733,11 @@ pglogical_receiver_main(Datum main_arg)
 			appendStringInfo(message, ", syncpoint_vector (origin/local) = {");
 			for (i = 0; i < MTM_MAX_NODES; i++)
 			{
-				if (spvector[i].origin_lsn != InvalidXLogRecPtr || spvector[i].local_lsn != InvalidXLogRecPtr)
+				/*
+				 * local_lsn must be always (even before first syncpoint)
+				 * non-zero in used cell; it is filter slot position.
+				 */
+				if (spvector[i].local_lsn != InvalidXLogRecPtr)
 				{
 					appendStringInfo(message, "%d: " LSN_FMT "/" LSN_FMT ", ",
 									 i + 1,

src/syncpoint.c

@@ -232,7 +232,9 @@ SyncpointGetLatest(int origin_node_id)
 		Assert(TupleDescAttr(tupdesc, 0)->atttypid == LSNOID);
 
 		sp.local_lsn = DatumGetLSN(SPI_getbinval(tup, tupdesc, 2, &isnull));
-		Assert(!isnull);
+		/* xxx c.f. similar in SyncpointGetAllLatest */
+		if (isnull)
+			mtm_log(ERROR, "fill_filter_since is null for node %d", origin_node_id);
 		Assert(TupleDescAttr(tupdesc, 1)->atttypid == LSNOID);
 	}
 	else if (rc == SPI_OK_SELECT && SPI_processed == 0)
@@ -307,12 +309,18 @@ SyncpointGetAllLatest(int sender_node_id)
 			origin_lsn = DatumGetLSN(SPI_getbinval(tup, tupdesc, 2, &isnull));
 			Assert(!isnull);
 			Assert(TupleDescAttr(tupdesc, 1)->atttypid == LSNOID);
+			spvector[node_id - 1].origin_lsn = origin_lsn;
 
 			local_lsn = DatumGetLSN(SPI_getbinval(tup, tupdesc, 3, &isnull));
-			Assert(!isnull);
 			Assert(TupleDescAttr(tupdesc, 2)->atttypid == LSNOID);
-
-			spvector[node_id - 1].origin_lsn = origin_lsn;
+			/*
+			 * XXX: there is no barrier between filter slot creation and
+			 * receiver start, so in theory (really unlikely) fastest receiver
+			 * may spin up before some slot is created by monitor and
+			 * fill_filter_since would be null. Shout in this case.
+			 */
+			if (isnull)
+				mtm_log(ERROR, "fill_filter_since is null for node %d", node_id);
 			spvector[node_id - 1].local_lsn = local_lsn;
 		}
 	}
@@ -392,9 +400,16 @@ RecoveryFilterLoad(int filter_node_id, Syncpoint *spvector, MtmConfig *mtm_cfg)
 	HTAB	   *filter_map;
 	int			estimate_size;
 	XLogRecPtr	start_lsn = PG_UINT64_MAX;
-	XLogRecPtr	current_last_lsn = GetFlushRecPtr();
+	XLogRecPtr	current_last_lsn;
 	int			i;
 
+	/*
+	 * ensure we will scan everything written up to this point
+	 * (if xact was empty it hadn't performed flush on commit)
+	 */
+	XLogFlush(GetXLogWriteRecPtr());
+	current_last_lsn = GetFlushRecPtr();
+
 	Assert(current_last_lsn != InvalidXLogRecPtr);
 
 	/* start from minimal among all of syncpoints */

t/005_add_stop_node.pl

@@ -1,5 +1,7 @@
 use strict;
 use warnings;
+
+use Carp;
 use PostgresNode;
 use Cluster;
 use TestLib;
@@ -101,8 +103,10 @@
 {
 	$cluster->pgbench(0, ('-N', '-n', -t => '100') );
 }
-sleep(2); # XXX sleep a bit to ensure monitor creates slot for new node on donor
-          #  which will be used by basebackup
+# ensure monitor creates slot for new node on donor which will be used by
+# basebackup before proceeding
+$cluster->poll_query_until(0, "select exists(select * from pg_replication_slots where slot_name = 'mtm_filter_slot_${new_node_id}');")
+    or croak "timed out waiting for slot creation";
 my $end_lsn = $cluster->backup_and_init(0, $new_node_off, $new_node_id);
 $cluster->release_socket($sock);
 $cluster->{nodes}->[$new_node_off]->append_conf('postgresql.conf', q{unix_socket_directories = ''