Drop replication slots on recovered nodes

Author: knizhnik

arbiter.c

@@ -362,7 +362,7 @@ static void MtmSendHeartbeat()
 					MTM_LOG2("Send heartbeat to node %d with timestamp %ld", i+1, now);    
 				}
 			} else { 
-				MTM_LOG1("Do not send hearbeat to node %d, busy mask %ld, status %d", i+1, busy_mask, Mtm->status);
+				MTM_LOG1("Do not send heartbeat to node %d, busy mask %ld, status %d", i+1, busy_mask, Mtm->status);
 			}
 		}
 	}
@@ -484,6 +484,8 @@ static int MtmConnectSocket(int node, int port, int timeout)
 	MtmCheckResponse(&resp);
 	MtmUnlock();
 
+	MtmOnNodeConnect(node+1);
+
 	busy_mask = save_mask;
 
 	return sd;
@@ -539,7 +541,7 @@ static bool MtmSendToNode(int node, void const* buf, int size)
 		}
 		if (sockets[node] < 0 || !MtmWriteSocket(sockets[node], buf, size)) { 
 			if (sockets[node] >= 0) { 
-				elog(WARNING, "Arbiter failed to write to node %d: %d", node+1, errno);
+				elog(WARNING, "Arbiter fail to write to node %d: %d", node+1, errno);
 				close(sockets[node]);
 				sockets[node] = -1;
 			}
@@ -549,7 +551,7 @@ static bool MtmSendToNode(int node, void const* buf, int size)
 				result = false;
 				break;
 			}
-			MTM_LOG3("Arbiter restablished connection with node %d", node+1);
+			MTM_LOG3("Arbiter reestablish connection with node %d", node+1);
 		} else { 
 			result = true;
 			break;
@@ -1075,7 +1077,7 @@ static void MtmReceiver(Datum arg)
 				if (!MtmWatchdog(now)) { 
 					for (i = 0; i < nNodes; i++) { 
 						if (Mtm->nodes[i].lastHeartbeat != 0 && sockets[i] >= 0) {
-							MTM_LOG1("Last hearbeat from node %d received %ld microseconds ago", i+1, now - Mtm->nodes[i].lastHeartbeat);
+							MTM_LOG1("Last heartbeat from node %d received %ld microseconds ago", i+1, now - Mtm->nodes[i].lastHeartbeat);
 						}
 					}
 				}

multimaster.c

@@ -450,7 +450,7 @@ bool MtmXidInMVCCSnapshot(TransactionId xid, Snapshot snapshot)
 	for (i = 0; i < MAX_WAIT_LOOPS; i++)
     {
         MtmTransState* ts = (MtmTransState*)hash_search(MtmXid2State, &xid, HASH_FIND, NULL);
-        if (ts != NULL/* && ts->status != TRANSACTION_STATUS_IN_PROGRESS*/)
+        if (ts != NULL /*&& ts->status != TRANSACTION_STATUS_IN_PROGRESS*/)
         {
             if (ts->csn > MtmTx.snapshot) { 
                 MTM_LOG4("%d: tuple with xid=%d(csn=%ld) is invisibile in snapshot %ld",
@@ -1242,13 +1242,17 @@ void MtmAbortTransaction(MtmTransState* ts)
 {	
 	Assert(MtmLockCount != 0); /* should be invoked with exclsuive lock */
 	if (ts->status != TRANSACTION_STATUS_ABORTED) { 
-		MTM_LOG1("Rollback active transaction %d:%d (local xid %d)", ts->gtid.node, ts->gtid.xid, ts->xid);
-		ts->status = TRANSACTION_STATUS_ABORTED;
-		MtmAdjustSubtransactions(ts);
-		if (ts->isActive) {
-			ts->isActive = false;
-			Assert(Mtm->nActiveTransactions != 0);
-			Mtm->nActiveTransactions -= 1;
+		if (ts->status == TRANSACTION_STATUS_COMMITTED) { 
+			elog(WARNING, "Attempt to rollback already committed transaction %d (%s)", ts->xid, ts->gid);
+		} else { 
+			MTM_LOG1("Rollback active transaction %d:%d (local xid %d) status %d", ts->gtid.node, ts->gtid.xid, ts->xid, ts->status);
+			ts->status = TRANSACTION_STATUS_ABORTED;
+			MtmAdjustSubtransactions(ts);
+			if (ts->isActive) {
+				ts->isActive = false;
+				Assert(Mtm->nActiveTransactions != 0);
+				Mtm->nActiveTransactions -= 1;
+			}
 		}
 	}
 }
@@ -1320,6 +1324,7 @@ static void MtmDisableNode(int nodeId)
 	elog(WARNING, "Disable node %d at xlog position %lx, last status change time %d msec ago", nodeId, GetXLogInsertRecPtr(), 
 		 (int)USEC_TO_MSEC(now - Mtm->nodes[nodeId-1].lastStatusChangeTime));
 	BIT_SET(Mtm->disabledNodeMask, nodeId-1);
+	Mtm->nodes[nodeId-1].timeline += 1;
 	Mtm->nodes[nodeId-1].lastStatusChangeTime = now;
 	Mtm->nodes[nodeId-1].lastHeartbeat = 0; /* defuse watchdog until first heartbeat is received */
 	if (nodeId != MtmNodeId) { 
@@ -1343,8 +1348,8 @@ static void MtmEnableNode(int nodeId)
 void MtmRecoveryCompleted(void)
 {
 	int i;
-	MTM_LOG1("Recovery of node %d is completed, disabled mask=%lx, reconnect mask=%lx, live nodes=%d", 
-			 MtmNodeId, Mtm->disabledNodeMask, Mtm->reconnectMask, Mtm->nLiveNodes);
+	MTM_LOG1("Recovery of node %d is completed, disabled mask=%lx, connectivity mask=%lx, live nodes=%d", 
+			 MtmNodeId, Mtm->disabledNodeMask, Mtm->connectivityMask, Mtm->nLiveNodes);
 	MtmLock(LW_EXCLUSIVE);
 	Mtm->recoverySlot = 0;
 	BIT_CLEAR(Mtm->disabledNodeMask, MtmNodeId-1);
@@ -1902,6 +1907,7 @@ static void MtmInitialize()
 			Mtm->nodes[i].lastHeartbeat = 0;
 			Mtm->nodes[i].restartLsn = 0;
 			Mtm->nodes[i].originId = InvalidRepOriginId;
+			Mtm->nodes[i].timeline = 0;
 		}
 		PGSemaphoreCreate(&Mtm->sendSemaphore);
 		PGSemaphoreReset(&Mtm->sendSemaphore);

multimaster.h

@@ -184,6 +184,7 @@ typedef struct
 	csn_t       oldestSnapshot; /* Oldest snapshot used by active transactions at this node */	
 	XLogRecPtr  restartLsn;
 	RepOriginId originId;
+	int         timeline;
 } MtmNodeInfo;
 
 typedef struct MtmTransState

pglogical_apply.c

@@ -620,7 +620,7 @@ process_remote_commit(StringInfo in)
 			Assert(!TransactionIdIsValid(MtmGetCurrentTransactionId()));
 			csn = pq_getmsgint64(in); 
 			gid = pq_getmsgstring(in);
-			MTM_LOG2("PGLOGICAL_COMMIT_PREPARED commit: csn=%ld, gid=%s, lsn=%ld", csn, gid, end_lsn);
+			MTM_LOG1("PGLOGICAL_COMMIT_PREPARED commit: csn=%ld, gid=%s, lsn=%ld", csn, gid, end_lsn);
 			StartTransactionCommand();
 			MtmBeginSession();
 			MtmSetCurrentTransactionCSN(csn);
@@ -1057,7 +1057,7 @@ void MtmExecutor(int id, void* work, size_t size)
 		MemoryContextSwitchTo(oldcontext);
 		EmitErrorReport();
         FlushErrorState();
-		MTM_LOG2("%d: REMOTE begin abort transaction %d", MyProcPid, MtmGetCurrentTransactionId());
+		MTM_LOG1("%d: REMOTE begin abort transaction %d", MyProcPid, MtmGetCurrentTransactionId());
 		MtmEndSession(false);
         AbortCurrentTransaction();
 		MTM_LOG2("%d: REMOTE end abort transaction %d", MyProcPid, MtmGetCurrentTransactionId());

pglogical_proto.c

@@ -155,6 +155,8 @@ pglogical_write_commit(StringInfo out, PGLogicalOutputData *data,
 	else
     	Assert(false);
 
+	Assert(flags != PGLOGICAL_COMMIT_PREPARED || txn->xid < 1000 || MtmTransactionRecords != 1);
+
 	if (flags == PGLOGICAL_COMMIT || flags == PGLOGICAL_PREPARE) { 
 		if (MtmIsFilteredTxn) { 
 			Assert(MtmTransactionRecords == 0);
@@ -164,8 +166,7 @@ pglogical_write_commit(StringInfo out, PGLogicalOutputData *data,
 		csn_t csn = MtmTransactionSnapshot(txn->xid);
 		bool isRecovery = MtmIsRecoveredNode(MtmReplicationNodeId);
 
-
-		if (!isRecovery && txn->origin_id != InvalidRepOriginId) 
+		if (!isRecovery && csn == INVALID_CSN && (flags != PGLOGICAL_ABORT_PREPARED || txn->origin_id != InvalidRepOriginId))
 		{
 			if (flags == PGLOGICAL_ABORT_PREPARED) { 
 				MTM_LOG1("Skip ABORT_PREPARED for transaction %s to node %d", txn->gid, MtmReplicationNodeId);
@@ -182,6 +183,7 @@ pglogical_write_commit(StringInfo out, PGLogicalOutputData *data,
 			flags |= PGLOGICAL_CAUGHT_UP;
 		}
 	}
+
     pq_sendbyte(out, 'C');		/* sending COMMIT */
 
 	MTM_LOG2("PGLOGICAL_SEND commit: event=%d, gid=%s, commit_lsn=%lx, txn->end_lsn=%lx, xlog=%lx", flags, txn->gid, commit_lsn, txn->end_lsn, GetXLogInsertRecPtr());

pglogical_receiver.c

@@ -250,6 +250,7 @@ pglogical_receiver_main(Datum main_arg)
 		int  count;
 		ConnStatusType status;
 		XLogRecPtr originStartPos = InvalidXLogRecPtr;
+		int timeline;
 
 		/* 
 		 * Determine when and how we should open replication slot.
@@ -261,6 +262,7 @@ pglogical_receiver_main(Datum main_arg)
 		{ 
 			break;
 		}
+		timeline = Mtm->nodes[nodeId-1].timeline;
 		count = Mtm->recoveryCount;
 
 		/* Establish connection to remote server */
@@ -274,14 +276,13 @@ pglogical_receiver_main(Datum main_arg)
 		}
 
 		query = createPQExpBuffer();
-#if 0 /* Do we need to recreate slot ? */		
-		if (mode == REPLMODE_RECOVERED) { /* recreate slot */
+		if (mode == REPLMODE_NORMAL && timeline != Mtm->nodes[nodeId-1].timeline) { /* recreate slot */
 			appendPQExpBuffer(query, "DROP_REPLICATION_SLOT \"%s\"", slotName);
 			res = PQexec(conn, query->data);
 			PQclear(res);
 			resetPQExpBuffer(query);
+			timeline = Mtm->nodes[nodeId-1].timeline;
 		}
-#endif
 		/* My original assumption was that we can perfrom recovery only fromm existed slot, 
 		 * but unfortunately looks like slots can "disapear" together with WAL-sender.
 		 * So let's try to recreate slot always. */