Recvoery bug fixing

Author: knizhnik

contrib/mmts/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;
 }

contrib/mmts/multimaster.c

@@ -673,7 +673,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");			
 		}
 	}
@@ -683,7 +683,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;
@@ -995,21 +997,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) 
 		{ 
 			/*
@@ -2250,7 +2254,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* 

contrib/mmts/pglogical_apply.c

@@ -499,12 +499,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 */

contrib/mmts/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);

src/backend/replication/logical/decode.c

@@ -488,7 +488,7 @@ DecodeCommit(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
 {
 	XLogRecPtr	origin_lsn = InvalidXLogRecPtr;
 	TimestampTz	commit_time = parsed->xact_time;
-	XLogRecPtr	origin_id = XLogRecGetOrigin(buf->record);
+	RepOriginId	origin_id = XLogRecGetOrigin(buf->record);
 	int			i;
 
 	if (parsed->xinfo & XACT_XINFO_HAS_ORIGIN)
@@ -541,6 +541,9 @@ DecodeCommit(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
 		(parsed->dbId != InvalidOid && parsed->dbId != ctx->slot->data.database) ||
 		FilterByOrigin(ctx, origin_id))
 	{
+		elog(WARNING, "%d: WAL-SENDER ignore record %lx with origin %d: SnapBuildXactNeedsSkip=%d, FilterByOrigin=%d", 
+			 getpid(), buf->origptr, origin_id, 
+			 SnapBuildXactNeedsSkip(ctx->snapshot_builder, buf->origptr), FilterByOrigin(ctx, origin_id));
 		for (i = 0; i < parsed->nsubxacts; i++)
 		{
 			ReorderBufferForget(ctx->reorder, parsed->subxacts[i], buf->origptr);