Recvoery bug fixing

knizhnik · kelvich · commit 50a052c319d2 · 2017-11-13T00:15:22.000+03:00
diff --git a/arbiter.c b/arbiter.c
@@ -386,7 +386,7 @@ static void MtmOpenConnections()
 static bool MtmSendToNode(int node, void const* buf, int size)
 {
 	while (sockets[node] < 0 || !MtmWriteSocket(sockets[node], buf, size)) { 
-		elog(WARNING, "Arbiter failed to write socket: %d", errno);
+		elog(WARNING, "Arbiter failed to write to node %d: %d", node+1, errno);
 		if (sockets[node] >= 0) { 
 			close(sockets[node]);
 		}
@@ -395,6 +395,7 @@ static bool MtmSendToNode(int node, void const* buf, int size)
 			MtmOnNodeDisconnect(node+1);
 			return false;
 		}
+		elog(NOTICE, "Arbiter restablish connection with node %d", node+1);
 	}
 	return true;
 }
diff --git a/multimaster.c b/multimaster.c
@@ -672,7 +672,7 @@ MtmPrePrepareTransaction(MtmCurrentTrans* x)
 
 	if (Mtm->disabledNodeMask != 0) { 
 		MtmRefreshClusterStatus(true);
-		if (Mtm->status != MTM_ONLINE) { 
+		if (!IsBackgroundWorker && Mtm->status != MTM_ONLINE) { 
 			elog(ERROR, "Abort current transaction because this cluster node is not online");			
 		}
 	}
@@ -682,7 +682,9 @@ MtmPrePrepareTransaction(MtmCurrentTrans* x)
 	/*
 	 * Check if there is global multimaster lock preventing new transaction from commit to make a chance to wal-senders to catch-up
 	 */
-	MtmCheckClusterLock();
+	if (!x->isReplicated) { 
+		MtmCheckClusterLock();
+	}
 
 	ts = hash_search(MtmXid2State, &x->xid, HASH_ENTER, NULL);
 	ts->status = TRANSACTION_STATUS_IN_PROGRESS;
@@ -994,21 +996,23 @@ bool MtmRecoveryCaughtUp(int nodeId, XLogRecPtr slotLSN)
 	if (MtmIsRecoveredNode(nodeId)) { 
 		XLogRecPtr walLSN = GetXLogInsertRecPtr();
 		MtmLock(LW_EXCLUSIVE);
+#if 0
 		if (slotLSN == walLSN) {
 			if (BIT_CHECK(Mtm->nodeLockerMask, nodeId-1)) { 
 				elog(WARNING,"Node %d is caught-up", nodeId);	
-				BIT_CLEAR(Mtm->disabledNodeMask, nodeId-1);
 				BIT_CLEAR(Mtm->walSenderLockerMask, MyWalSnd - WalSndCtl->walsnds);
 				BIT_CLEAR(Mtm->nodeLockerMask, nodeId-1);
 				Mtm->nLockers -= 1;
 			} else { 
 				elog(WARNING,"Node %d is caugth-up without locking cluster", nodeId);	
 				/* We are lucky: caugth-up without locking cluster! */
-				Mtm->nNodes += 1;
-				BIT_CLEAR(Mtm->disabledNodeMask, nodeId-1);
 			}
+			BIT_CLEAR(Mtm->disabledNodeMask, nodeId-1);
+			Mtm->nNodes += 1;
 			caughtUp = true;
-		} else if (!BIT_CHECK(Mtm->nodeLockerMask, nodeId-1)
+		} else 
+#endif
+		if (!BIT_CHECK(Mtm->nodeLockerMask, nodeId-1)
 				   && slotLSN + MtmMinRecoveryLag > walLSN) 
 		{ 
 			/*
@@ -2248,7 +2252,12 @@ MtmExecutorFinish(QueryDesc *queryDesc)
 
 void MtmExecute(void* work, int size)
 {
-    BgwPoolExecute(&Mtm->pool, work, size);
+	if (Mtm->status == MTM_RECOVERY) { 
+		/* During recovery apply changes sequentially to preserve commit order */
+		MtmExecutor(0, work, size);
+	} else { 
+		BgwPoolExecute(&Mtm->pool, work, size);
+	}
 }
     
 static BgwPool* 
diff --git a/pglogical_apply.c b/pglogical_apply.c
@@ -497,12 +497,12 @@ process_remote_commit(StringInfo in)
 	uint8 		flags;
 	csn_t       csn;
 	const char *gid = NULL;	
-	bool        caughtUp;
+	bool        caughtUp = false;
 
 	/* read flags */
 	flags = pq_getmsgbyte(in);
 	MtmReplicationNode = pq_getmsgbyte(in);
-	caughtUp = pq_getmsgbyte(in) != 0;
+	/*caughtUp = pq_getmsgbyte(in) != 0;*/
 
 	/* read fields */
 	replorigin_session_origin_lsn = pq_getmsgint64(in); /* commit_lsn */
diff --git a/pglogical_proto.c b/pglogical_proto.c
@@ -147,6 +147,7 @@ pglogical_write_commit(StringInfo out, PGLogicalOutputData *data,
 			return;
 		}
 	}
+	MtmRecoveryCaughtUp(MtmReplicationNodeId, txn->end_lsn);
 
     pq_sendbyte(out, 'C');		/* sending COMMIT */
 
@@ -155,7 +156,7 @@ pglogical_write_commit(StringInfo out, PGLogicalOutputData *data,
     /* send the flags field */
     pq_sendbyte(out, flags);
     pq_sendbyte(out, MtmNodeId);
-    pq_sendbyte(out, MtmRecoveryCaughtUp(MtmReplicationNodeId, txn->end_lsn));
+    /*pq_sendbyte(out, MtmRecoveryCaughtUp(MtmReplicationNodeId, txn->end_lsn));*/
 
     /* send fixed fields */
     pq_sendint64(out, commit_lsn);

Original file line number	Diff line number	Diff line change
`@@ -386,7 +386,7 @@ static void MtmOpenConnections()`
`386`	`386`	`static bool MtmSendToNode(int node, void const* buf, int size)`
`387`	`387`	`{`
`388`	`388`	`while (sockets[node] < 0 \|\| !MtmWriteSocket(sockets[node], buf, size)) {`
`389`		`- elog(WARNING, "Arbiter failed to write socket: %d", errno);`
	`389`	`+ elog(WARNING, "Arbiter failed to write to node %d: %d", node+1, errno);`
`390`	`390`	`if (sockets[node] >= 0) {`
`391`	`391`	`close(sockets[node]);`
`392`	`392`	`}`
`@@ -395,6 +395,7 @@ static bool MtmSendToNode(int node, void const* buf, int size)`
`395`	`395`	`MtmOnNodeDisconnect(node+1);`
`396`	`396`	`return false;`
`397`	`397`	`}`
	`398`	`+ elog(NOTICE, "Arbiter restablish connection with node %d", node+1);`
`398`	`399`	`}`
`399`	`400`	`return true;`
`400`	`401`	`}`
Original file line number	Diff line number	Diff line change
`@@ -147,6 +147,7 @@ pglogical_write_commit(StringInfo out, PGLogicalOutputData *data,`
`147`	`147`	`return;`
`148`	`148`	`}`
`149`	`149`	`}`
	`150`	`+ MtmRecoveryCaughtUp(MtmReplicationNodeId, txn->end_lsn);`
`150`	`151`
`151`	`152`	`pq_sendbyte(out, 'C'); /* sending COMMIT */`
`152`	`153`
`@@ -155,7 +156,7 @@ pglogical_write_commit(StringInfo out, PGLogicalOutputData *data,`
`155`	`156`	`/* send the flags field */`
`156`	`157`	`pq_sendbyte(out, flags);`
`157`	`158`	`pq_sendbyte(out, MtmNodeId);`
`158`		`- pq_sendbyte(out, MtmRecoveryCaughtUp(MtmReplicationNodeId, txn->end_lsn));`
	`159`	`+ /pq_sendbyte(out, MtmRecoveryCaughtUp(MtmReplicationNodeId, txn->end_lsn));/`
`159`	`160`
`160`	`161`	`/* send fixed fields */`
`161`	`162`	`pq_sendint64(out, commit_lsn);`