few fixes after buildfarm tests round

Author: kelvich

Diff for: src/ddd.c

@@ -376,7 +376,7 @@ MtmDetectGlobalDeadLockForXid(TransactionId xid)
 	ByteBufferAlloc(&buf);
 	EnumerateLocks(MtmSerializeLock, &buf);
 
-	Assert(replorigin_session_origin == InvalidRepOriginId);
+	// Assert(replorigin_session_origin == InvalidRepOriginId);
 	XLogFlush(LogLogicalMessage("L", buf.data, buf.used, false));
 
 	MtmGraphInit(&graph);

Diff for: src/include/receiver.h

@@ -6,7 +6,7 @@ typedef struct
 	int node_id;
 	bool is_recovery;
 	bool parallel_allowed;
-	TimestampTz session_id;
+	uint64 session_id;
 	XLogRecPtr end_lsn;
 } MtmReceiverContext;
 

Diff for: src/include/syncpoint.h

@@ -21,9 +21,15 @@ typedef struct
 	XLogRecPtr	local_lsn;
 } Syncpoint;
 
+/*
+ * Used as a hashkey in recovery filter.
+ *
+ * NB: make sure to memset this structure to zeroes before using as hashkey
+ * because it contains 4-byte padding hole in the middle.
+ */
 typedef struct
 {
-	int64		node_id;
+	int			node_id;
 	XLogRecPtr	origin_lsn;
 } FilterEntry;
 

Diff for: src/multimaster.c

@@ -739,6 +739,11 @@ drop_node_entry(int node_id)
 	PopActiveSnapshot();
 }
 
+/*
+ * Load mtm config.
+ *
+ * In case of absense of configured nodes will return cfg->n_nodes = 0.
+ */
 MtmConfig *
 MtmLoadConfig()
 {
@@ -774,8 +779,6 @@ MtmLoadConfig()
 		cfg->n_nodes = SPI_processed;
 		cfg->my_node_id = 0;
 
-		// XXX: allow zero nodes?
-
 		for (i = 0; i < cfg->n_nodes; i++)
 		{
 			HeapTuple	tup = SPI_tuptable->vals[i];

Diff for: src/pglogical_receiver.c

@@ -383,9 +383,13 @@ MtmFilterTransaction(char *record, int size, Syncpoint *spvector, HTAB *filter_m
 	}
 	else
 	{
-		FilterEntry	entry = {origin_node, tx_lsn};
+		FilterEntry	entry;
 		bool		found;
 
+		memset(&entry, '\0', sizeof(FilterEntry));
+		entry.node_id = origin_node;
+		entry.origin_lsn = tx_lsn;
+
 		hash_search(filter_map, &entry, HASH_FIND, &found);
 
 		{
@@ -580,7 +584,13 @@ pglogical_receiver_main(Datum main_arg)
 		/* Create new slot if needed */
 		query = createPQExpBuffer();
 
-		receiver_ctx.session_id = MtmGetIncreasingTimestamp();
+		/*
+		 * Hand-made uuid. First byte is node_id, then rest is timestamp which
+		 * is guaranteed to be uniq on this node.
+		 */
+		receiver_ctx.session_id = ((uint64) receiver_mtm_cfg->my_node_id << 56)
+								  + MtmGetIncreasingTimestamp();
+
 		receiver_ctx.is_recovery = mode == REPLMODE_RECOVERY;
 		receiver_ctx.parallel_allowed = false;
 

Diff for: src/state.c

@@ -998,7 +998,19 @@ MtmMonitor(Datum arg)
 	memcpy(&user_id, MyBgworkerEntry->bgw_extra + sizeof(Oid), sizeof(Oid));
 	BackgroundWorkerInitializeConnectionByOid(db_id, user_id, 0);
 
-	mtm_cfg = MtmLoadConfig();
+	/*
+	 * During init_node() our worker is started from transaction that created
+	 * mtm config, so we can get here before this transaction is committed,
+	 * so we won't see config yet. Just wait for it to became visible.
+	 */
+	for (;;)
+	{
+		mtm_cfg = MtmLoadConfig();
+		if (mtm_cfg->n_nodes > 0)
+			break;
+		pfree(mtm_cfg);
+		MtmSleep(USECS_PER_SEC);
+	}
 
 	/* Set proper values in Mtm structure */
 	MtmStateFill(mtm_cfg);

Diff for: src/syncpoint.c

@@ -524,9 +524,13 @@ RecoveryFilterLoad(int filter_node_id, Syncpoint *spvector, MtmConfig *mtm_cfg)
 
 		if (XLogRecGetRmid(xlogreader) == RM_XACT_ID)
 		{
-			uint32 info = XLogRecGetInfo(xlogreader);
-			FilterEntry				entry = {node_id, InvalidXLogRecPtr};
-			bool					found;
+			uint32		info = XLogRecGetInfo(xlogreader);
+			FilterEntry	entry;
+			bool		found;
+
+			memset(&entry, '\0', sizeof(FilterEntry));
+			entry.node_id = node_id;
+			entry.origin_lsn = InvalidXLogRecPtr;
 
 			switch (info & XLOG_XACT_OPMASK)
 			{
@@ -563,7 +567,7 @@ RecoveryFilterLoad(int filter_node_id, Syncpoint *spvector, MtmConfig *mtm_cfg)
 				continue;
 
 			Assert(entry.origin_lsn != InvalidXLogRecPtr);
-			mtm_log(MtmReceiverFilter, "load_filter_map: add {%ld, %"INT64_MODIFIER"x } %d (%"INT64_MODIFIER"x)",
+			mtm_log(MtmReceiverFilter, "load_filter_map: add {%d, %"INT64_MODIFIER"x } %d (%"INT64_MODIFIER"x)",
 					entry.node_id, entry.origin_lsn, info & XLOG_XACT_OPMASK, xlogreader->EndRecPtr);
 			hash_search(filter_map, &entry, HASH_ENTER, &found);
 			Assert(!found);