Fix unreleased lock in GetPreparedTransactionState

Author: knizhnik

multimaster.c

@@ -1773,6 +1773,10 @@ void MtmRecoveryCompleted(void)
 	MTM_LOG1("Recovery of node %d is completed, disabled mask=%llx, connectivity mask=%llx, endLSN=%lx, live nodes=%d",
 			 MtmNodeId, (long long) Mtm->disabledNodeMask, 
 			 (long long)SELF_CONNECTIVITY_MASK, GetXLogInsertRecPtr(), Mtm->nLiveNodes);
+	if (Mtm->nAllNodes >= 3) { 
+		elog(WARNING, "restartLSNs at the end of recovery: {%lx, %lx, %lx}", 
+			 Mtm->nodes[0].restartLSN, Mtm->nodes[1].restartLSN, Mtm->nodes[2].restartLSN);
+	}
 	MtmLock(LW_EXCLUSIVE);
 	Mtm->recoverySlot = 0;
 	Mtm->recoveredLSN = GetXLogInsertRecPtr();
@@ -3244,7 +3248,12 @@ MtmReplicationMode MtmGetReplicationMode(int nodeId, sig_atomic_t volatile* shut
 				|| Mtm->recoverySlot == nodeId) 
 			{ 
 				/* Choose for recovery first available slot or slot of donor node (if any) */
-				elog(WARNING, "Process %d starts recovery from node %d", MyProcPid, nodeId);
+				if (Mtm->nAllNodes >= 3) { 
+					elog(WARNING, "Process %d starts recovery from node %d restartLSNs={%lx, %lx, %lx}", 
+						 MyProcPid, nodeId, Mtm->nodes[0].restartLSN, Mtm->nodes[1].restartLSN, Mtm->nodes[2].restartLSN);
+				} else {
+					elog(WARNING, "Process %d starts recovery from node %d", MyProcPid, nodeId);
+				}
 				Mtm->recoverySlot = nodeId;
 				Mtm->nReceivers = 0;
 				Mtm->nSenders = 0;
@@ -3383,7 +3392,7 @@ MtmReplicationStartupHook(struct PGLogicalStartupHookArgs* args)
 				sscanf(strVal(elem->arg), "%lx", &recoveredLSN);
 				MTM_LOG1("Recovered position of node %d is %lx", MtmReplicationNodeId, recoveredLSN); 
 				if (Mtm->nodes[MtmReplicationNodeId-1].restartLSN < recoveredLSN) { 
-					MTM_LOG2("[restartlsn] node %d: %lx -> %lx (MtmReplicationStartupHook)", MtmReplicationNodeId, Mtm->nodes[MtmReplicationNodeId-1].restartLSN, recoveredLSN);
+					MTM_LOG1("Advance restartLSN for node %d from %lx to %lx (MtmReplicationStartupHook)", MtmReplicationNodeId, Mtm->nodes[MtmReplicationNodeId-1].restartLSN, recoveredLSN);
 					Assert(Mtm->nodes[MtmReplicationNodeId-1].restartLSN == InvalidXLogRecPtr 
 						   || recoveredLSN < Mtm->nodes[MtmReplicationNodeId-1].restartLSN + MtmMaxRecoveryLag);
 					Mtm->nodes[MtmReplicationNodeId-1].restartLSN = recoveredLSN;
@@ -3587,9 +3596,13 @@ bool MtmFilterTransaction(char* record, int size)
 	origin_node = pq_getmsgbyte(&s);
 	origin_lsn = pq_getmsgint64(&s);
 
-	Assert(replication_node == MtmReplicationNodeId && 
-		   origin_node != 0 &&
-		   (Mtm->status == MTM_RECOVERY || origin_node == replication_node));	
+	Assert(replication_node == MtmReplicationNodeId);
+	if (!(origin_node != 0 &&
+		  (Mtm->status == MTM_RECOVERY || origin_node == replication_node)))
+	{
+		elog(WARNING, "Receive redirected commit event %d from node %d origin node %d origin LSN %lx in %s mode",
+			 event, replication_node, origin_node, origin_lsn, MtmNodeStatusMnem[Mtm->status]);
+	}
 
     switch (event)
     {
@@ -3616,8 +3629,8 @@ bool MtmFilterTransaction(char* record, int size)
 	}
 
 	if (duplicate) {
-		MTM_LOG1("Ignore transaction %s from node %d event=%x because our LSN position %lx for origin node %d is greater or equal than LSN %lx of this transaction (end_lsn=%lx, origin_lsn=%lx)", 
-				 gid, replication_node, event, Mtm->nodes[origin_node-1].restartLSN, origin_node, restart_lsn, end_lsn, origin_lsn);
+		MTM_LOG1("Ignore transaction %s from node %d event=%x because our LSN position %lx for origin node %d is greater or equal than LSN %lx of this transaction (end_lsn=%lx, origin_lsn=%lx) mode %s", 
+				 gid, replication_node, event, Mtm->nodes[origin_node-1].restartLSN, origin_node, restart_lsn, end_lsn, origin_lsn, MtmNodeStatusMnem[Mtm->status]);
 	} else {
 		MTM_LOG2("Apply transaction %s from node %d lsn %lx, event=%x, origin node %d, original lsn=%lx, current lsn=%lx", 
 				 gid, replication_node, end_lsn, event, origin_node, origin_lsn, restart_lsn);

pglogical_receiver.c

@@ -342,11 +342,11 @@ pglogical_receiver_main(Datum main_arg)
 				 * Them are either empty, either new node is synchronized using base_backup.
 				 * So we assume that LSNs are the same for local and remote node
 				 */
-				originStartPos = Mtm->status == MTM_RECOVERY && Mtm->donorNodeId == nodeId ? GetXLogInsertRecPtr() : InvalidXLogRecPtr;
+				originStartPos = (Mtm->status == MTM_RECOVERY && Mtm->donorNodeId == nodeId) ? GetXLogInsertRecPtr() : InvalidXLogRecPtr;
 				MTM_LOG1("Start logical receiver at position %lx from node %d", originStartPos, nodeId);
 			} else { 
 				if (Mtm->nodes[nodeId-1].restartLSN < originStartPos) { 
-					MTM_LOG2("[restartlsn] node %d: %lx -> %lx (pglogical_receiver_mains)", nodeId, Mtm->nodes[nodeId-1].restartLSN, originStartPos);
+					MTM_LOG1("Advance restartLSN for node %d: from %lx to %lx (pglogical_receiver_main)", nodeId, Mtm->nodes[nodeId-1].restartLSN, originStartPos);
 					Mtm->nodes[nodeId-1].restartLSN = originStartPos;
 				}
 				MTM_LOG1("Restart logical receiver at position %lx with origin=%d from node %d", originStartPos, originId, nodeId);
@@ -545,16 +545,17 @@ pglogical_receiver_main(Datum main_arg)
 					}
 					if (stmt[0] == 'Z' || (stmt[0] == 'M' && (stmt[1] == 'L' || stmt[1] == 'A' || stmt[1] == 'C'))) {
 						MTM_LOG3("Process '%c' message from %d", stmt[1], nodeId);
-						if ( stmt[1] == 'C') { /* concurrent DDL */
+						if (stmt[0] == 'M' && stmt[1] == 'C') { /* concurrent DDL should be executed by parallel workers */
 							MtmExecute(stmt, rc - hdr_len);
 						} else {
-							MtmExecutor(stmt, rc - hdr_len);
+							MtmExecutor(stmt, rc - hdr_len); /* all other messages can be processed by receiver itself */
 						}
 					} else { 
 						ByteBufferAppend(&buf, stmt, rc - hdr_len);
 						if (stmt[0] == 'C') /* commit */
 						{
-							if (!MtmFilterTransaction(stmt, rc - hdr_len)) { 
+							if (!MtmFilterTransaction(stmt, rc - hdr_len)) 
+							{ 
 								if (spill_file >= 0) { 
 									ByteBufferAppend(&buf, ")", 1);
 									pq_sendbyte(&spill_info, '(');
@@ -574,6 +575,7 @@ pglogical_receiver_main(Datum main_arg)
 											elog(WARNING, "Commit of prepared transaction takes %ld usec, flags=%x", stop - start, stmt[1]);
 										}
 									} else {
+										Assert(stmt[1] == PGLOGICAL_PREPARE || stmt[1] == PGLOGICAL_COMMIT); /* all other commits should be applied in place */
 										MtmExecute(buf.data, buf.used);
 									}
 								}