Add some log and fix COPY BOTH mode

Andrei Krichinin · Andrei Krichinin · commit 5fe8013782e2 · 2022-10-19T12:15:30.000+05:00
diff --git a/src/dmq.c b/src/dmq.c
@@ -809,7 +809,7 @@ dmq_sender_main(Datum main_arg)
 			switch (conns[conn_id].state)
 			{
 				case Idle:
-					Assert(false);
+//					Assert(false);
 					break;
 
 					/*
@@ -908,10 +908,23 @@ dmq_sender_main(Datum main_arg)
 					if (!PQisBusy(conns[conn_id].pgconn))
 					{
 						int8 mask_pos = conns[conn_id].mask_pos;
+						PGresult *result = PQgetResult(conns[conn_id].pgconn);
 
-						/*
-						 * XXX check here that dmq_receiver_loop not failed?
-						 */
+						if (!result) {
+							mtm_log(ERROR, "[DMQ] PQgetResult returned NULL");
+						}
+
+						mtm_log(DmqStateIntermediate, "PQgetResult status: %d", PQresultStatus(result));
+
+						if (PQresultStatus(result) != PGRES_COPY_BOTH) {
+							mtm_log(DmqStateFinal, "[DMQ] wrong response from dmq receiver '%s': '%s'; '%s'",
+									conns[conn_id].receiver_name,
+									PQresStatus(PQresultStatus(result)),
+									PQerrorMessage(conns[conn_id].pgconn));
+							conns[conn_id].state = Idle;
+							dmq_state->sconn_cnt[conns[conn_id].mask_pos] = DMQSCONN_DEAD;
+							break;
+						}
 
 						conns[conn_id].state = Active;
 						DeleteWaitEvent(set, event.pos);
diff --git a/src/multimaster.c b/src/multimaster.c
@@ -52,6 +52,8 @@
 
 #include "compat.h"
 
+#include "time.h"
+
 typedef enum
 {
 	MTM_STATE_LOCK_ID
@@ -2040,14 +2042,27 @@ gather(nodemask_t participants,
 	   gather_hook_t msg_ok, Datum msg_ok_arg,
 	   int *sendconn_cnt, uint64 gen_num)
 {
+	time_t start;
 	*msg_count = 0;
+   
+	start = time(NULL);
+
+	//elog(LOG, "----> gather 1");
 	while (participants != 0)
 	{
 		bool		ret;
 		int8 sender_mask_pos;
 		StringInfoData msg;
 		int			rc;
 		bool		wait;
+		time_t current;
+	   
+		current = time(NULL);
+
+		if (current - start > 5) {
+			elog(LOG, "----> gather timeout");
+//			return false;
+		}
 
 		ret = dmq_pop_nb(&sender_mask_pos, &msg, participants, &wait);
 		if (ret)
@@ -2102,6 +2117,7 @@ gather(nodemask_t participants,
 		}
 
 	}
+	//elog(LOG, "----> gather 2");
 	return true;
 }
 
diff --git a/src/pglogical_proto.c b/src/pglogical_proto.c
@@ -175,6 +175,7 @@ pglogical_write_begin(StringInfo out, PGLogicalOutputData *data,
 	MtmLastRelId = InvalidOid;
 	MtmCurrentXid = txn->xid;
 	DDLInProgress = false;
+	elog(LOG, "---> pglogical_write_begin: false");
 
 	pq_sendbyte(out, 'B');		/* BEGIN */
 	pq_sendint(out, hooks_data->cfg->my_node_id, 4);
@@ -253,6 +254,7 @@ pglogical_write_message(StringInfo out, LogicalDecodingContext *ctx,
 
 		case 'D':
 			DDLInProgress = true;
+			elog(LOG, "---> pglogical_write_message: true %d: %s",hooks_data->receiver_node_id, message);
 			mtm_log(ProtoTraceMessage, "Sent tx DDL message to node %d: %s",
 					hooks_data->receiver_node_id, message);
 			break;
@@ -266,6 +268,7 @@ pglogical_write_message(StringInfo out, LogicalDecodingContext *ctx,
 
 		case 'E':
 			DDLInProgress = false;
+			elog(LOG, "---> pglogical_write_message: false");
 
 			/*
 			 * we use End message only as indicator of DDL transaction finish,
