eXplicitely set restart LSN

Author: knizhnik

contrib/mmts/arbiter.c

@@ -912,7 +912,14 @@ static void MtmTransReceiver(Datum arg)
 						switch (msg->code) { 
 						  case MSG_READY:
 							MTM_TXTRACE(ts, "MtmTransReceiver got MSG_READY");
+							if (ts->status == TRANSACTION_STATUS_COMMITTED) { 
+								elog(WARNING, "Receive READY response for already committed transaction %d from node %d",
+									 ts->xid, msg->node);
+								continue;
+							}
 							if (ts->nVotes >= Mtm->nLiveNodes) {
+								elog(WARNING, "Receive deteriorated READY response for transaction %d (%s) from node %d",
+									 ts->xid, ts->gid, msg->node);
 								MtmAbortTransaction(ts);
 								MtmWakeUpBackend(ts);
 							} else { 
@@ -956,6 +963,8 @@ static void MtmTransReceiver(Datum arg)
 						  case MSG_PREPARED:
 							MTM_TXTRACE(ts, "MtmTransReceiver got MSG_PREPARED");
 							if (ts->nVotes >= Mtm->nLiveNodes) {					
+								elog(WARNING, "Receive deteriorated PREPARED response for transaction %d (%s) from node %d",
+									 ts->xid, ts->gid, msg->node);
 								MtmAbortTransaction(ts);
 								MtmWakeUpBackend(ts);
 							} else { 

contrib/mmts/multimaster.c

@@ -388,16 +388,17 @@ MtmInitializeSequence(int64* start, int64* step)
 
 csn_t MtmTransactionSnapshot(TransactionId xid)
 {
-	MtmTransState* ts;
 	csn_t snapshot = INVALID_CSN;
-
+	
 	MtmLock(LW_SHARED);
-    ts = hash_search(MtmXid2State, &xid, HASH_FIND, NULL);
-    if (ts != NULL && !ts->isLocal) { 
-		snapshot = ts->snapshot;
+	if (Mtm->status == MTM_ONLINE) {
+		MtmTransState* ts = hash_search(MtmXid2State, &xid, HASH_FIND, NULL);
+		if (ts != NULL && !ts->isLocal) { 
+			snapshot = ts->snapshot;
+			Assert(ts->gtid.node == MtmNodeId || MtmIsRecoverySession); 		
+		}
 	}
 	MtmUnlock();
-
     return snapshot;
 }
 
@@ -1007,7 +1008,7 @@ MtmEndTransaction(MtmCurrentTrans* x, bool commit)
 			}
 		}
 		if (!commit && x->isReplicated && TransactionIdIsValid(x->gtid.xid)) { 
-			Assert(Mtm->status != MTM_RECOVERY);
+			Assert(Mtm->status != MTM_RECOVERY || Mtm->recoverySlot != MtmNodeId);
 			/* 
 			 * Send notification only if ABORT happens during transaction processing at replicas, 
 			 * do not send notification if ABORT is received from master 
@@ -2455,29 +2456,32 @@ MtmReplicationMode MtmGetReplicationMode(int nodeId, sig_atomic_t volatile* shut
 			return REPLMODE_EXIT;
 		}
 		MTM_LOG2("%d: receiver slot mode %s", MyProcPid, MtmNodeStatusMnem[Mtm->status]);
+		MtmLock(LW_EXCLUSIVE);
 		if (Mtm->status == MTM_RECOVERY) { 
 			recovery = true;
 			if (Mtm->recoverySlot == 0 || Mtm->recoverySlot == nodeId) { 
 				/* Choose for recovery first available slot */
-				MTM_LOG1("Start recovery from node %d", nodeId);
+				elog(WARNING, "Process %d starts recovery from node %d", MyProcPid, nodeId);
 				Mtm->recoverySlot = nodeId;
 				Mtm->nReceivers = 0;
 				Mtm->recoveryCount += 1;
 				Mtm->pglogicalNodeMask = 0;
-				FinishAllPreparedTransactions(false);
 				for (i = 0; i < Mtm->nAllNodes; i++) { 
-					Mtm->nodes[i].restartLsn = 0;
+					Mtm->nodes[i].restartLsn = InvalidXLogRecPtr;
 				}
+				MtmUnlock();
+				FinishAllPreparedTransactions(false);
 				return REPLMODE_RECOVERY;
 			}
 		}
+		MtmUnlock();
 		/* delay opening of other slots until recovery is completed */
 		MtmSleep(STATUS_POLL_DELAY);
 	}
 	if (recovery) { 
-		MTM_LOG1("Recreate replication slot for node %d after end of recovery", nodeId);
+		MTM_LOG1("%d: Restart replication for node %d after end of recovery", MyProcPid, nodeId);
 	} else { 
-		MTM_LOG2("%d: Reuse replication slot for node %d", MyProcPid, nodeId);
+		MTM_LOG1("%d: Continue replication slot for node %d", MyProcPid, nodeId);
 	}
 	/* After recovery completion we need to drop all other slots to avoid receive of redundant data */
 	return recovery ? REPLMODE_RECOVERED : REPLMODE_NORMAL;

contrib/mmts/pglogical_apply.c

@@ -59,7 +59,6 @@ typedef struct TupleData
 	bool		changed[MaxTupleAttributeNumber];
 } TupleData;
 
-static int MtmTransactionRecords;
 static bool inside_tx = false;
 
 static Relation read_rel(StringInfo s, LOCKMODE mode);
@@ -529,6 +528,8 @@ MtmEndSession(bool unlock)
 	if (replorigin_session_origin != InvalidRepOriginId) { 
 		MTM_LOG2("%d: Begin reset replorigin session for node %d: %d, progress %lx", MyProcPid, MtmReplicationNodeId, replorigin_session_origin, replorigin_session_get_progress(false));
 		replorigin_session_origin = InvalidRepOriginId;
+		replorigin_session_origin_lsn = InvalidXLogRecPtr;
+		replorigin_session_origin_timestamp = 0;
 		replorigin_session_reset();
 		if (unlock) { 
 			MtmUnlockNode(MtmReplicationNodeId);
@@ -540,42 +541,25 @@ MtmEndSession(bool unlock)
 static void
 process_remote_commit(StringInfo in)
 {
-	int         i;
 	uint8 		flags;
 	csn_t       csn;
 	const char *gid = NULL;	
 	XLogRecPtr  end_lsn;
 	XLogRecPtr  origin_lsn;
-	RepOriginId originId;
-	int         n_records;
+	int         origin_node;
 	/* read flags */
 	flags = pq_getmsgbyte(in);
 	MtmReplicationNodeId = pq_getmsgbyte(in);
 
-	n_records = pq_getmsgint(in, 4);
-	if (MtmTransactionRecords != n_records) { 
-		elog(ERROR, "Transaction %d flags %d contains %d records instead of %d", MtmGetCurrentTransactionId(), flags, MtmTransactionRecords, n_records);
-	}
-
 	/* read fields */
 	replorigin_session_origin_lsn = pq_getmsgint64(in); /* commit_lsn */
 	end_lsn = pq_getmsgint64(in); /* end_lsn */
 	replorigin_session_origin_timestamp = pq_getmsgint64(in); /* commit_time */
 
-	originId = (RepOriginId)pq_getmsgint(in, 2);
+	origin_node = pq_getmsgbyte(in);
 	origin_lsn = pq_getmsgint64(in);
+	Mtm->nodes[origin_node-1].restartLsn = origin_lsn;
 
-	if (originId != InvalidRepOriginId) { 
-		for (i = 0; i < Mtm->nAllNodes; i++) { 
-			if (Mtm->nodes[i].originId == originId) { 
-				Mtm->nodes[i].restartLsn = origin_lsn;
-				break;
-			} 
-		}
-		if (i == Mtm->nAllNodes) { 
-			elog(WARNING, "Failed to map origin %d", originId);
-		}
-	}
 	Assert(replorigin_session_origin == InvalidRepOriginId);
 
 	switch(PGLOGICAL_XACT_EVENT(flags))
@@ -677,8 +661,6 @@ process_remote_insert(StringInfo s, Relation rel)
 	ScanKey	*index_keys;
 	int	i;
 
-	MtmTransactionRecords += 1;
-
 	estate = create_rel_estate(rel);
 	newslot = ExecInitExtraTupleSlot(estate);
 	oldslot = ExecInitExtraTupleSlot(estate);
@@ -777,8 +759,6 @@ process_remote_update(StringInfo s, Relation rel)
 	ScanKeyData skey[INDEX_MAX_KEYS];
 	HeapTuple	remote_tuple = NULL;
 
-	MtmTransactionRecords += 1;
-
 	action = pq_getmsgbyte(s);
 
 	/* old key present, identifying key changed */
@@ -896,8 +876,6 @@ process_remote_delete(StringInfo s, Relation rel)
 	ScanKeyData skey[INDEX_MAX_KEYS];
 	bool		found_old;
 
-	MtmTransactionRecords += 1;
-
 	estate = create_rel_estate(rel);
 	oldslot = ExecInitExtraTupleSlot(estate);
 	ExecSetSlotDescriptor(oldslot, RelationGetDescr(rel));
@@ -985,7 +963,6 @@ void MtmExecutor(int id, void* work, size_t size)
     }
     MemoryContextSwitchTo(ApplyContext);
 	replorigin_session_origin = InvalidRepOriginId;
-	MtmTransactionRecords = 0;
     PG_TRY();
     {    
         while (true) { 

contrib/mmts/pglogical_proto.c

@@ -186,14 +186,25 @@ pglogical_write_commit(StringInfo out, PGLogicalOutputData *data,
     pq_sendbyte(out, MtmNodeId);
 
 	Assert(txn->xact_action != XLOG_XACT_PREPARE || txn->xid < 1000 || MtmTransactionRecords >= 2);
-	pq_sendint(out, MtmTransactionRecords, 4);
-	
+
     /* send fixed fields */
     pq_sendint64(out, commit_lsn);
     pq_sendint64(out, txn->end_lsn);
     pq_sendint64(out, txn->commit_time);
 
-	pq_sendint(out, txn->origin_id, 2);
+	if (txn->origin_id != InvalidRepOriginId) { 
+		int i;
+		for (i = 0; i < Mtm->nAllNodes && Mtm->nodes[i].originId != txn->origin_id; i++);
+		if (i == Mtm->nAllNodes) { 
+			elog(WARNING, "Failed to map origin %d", txn->origin_id);
+			i = MtmNodeId-1;
+		} else { 
+			//Assert(i == MtmNodeId-1 || txn->origin_lsn != InvalidXLogRecPtr);
+		}
+		pq_sendbyte(out, i+1);
+	} else { 
+		pq_sendbyte(out, MtmNodeId);
+	}
 	pq_sendint64(out, txn->origin_lsn);
 
 	if (txn->xact_action == XLOG_XACT_COMMIT_PREPARED) { 

contrib/mmts/pglogical_receiver.c

@@ -247,7 +247,7 @@ pglogical_receiver_main(Datum main_arg)
 	{ 
 		int  count;
 		ConnStatusType status;
-		XLogRecPtr originStartPos = 0;
+		XLogRecPtr originStartPos = InvalidXLogRecPtr;
 
 		/* 
 		 * Determine when and how we should open replication slot.
@@ -306,8 +306,9 @@ pglogical_receiver_main(Datum main_arg)
 		/* Start logical replication at specified position */
 		if (mode == REPLMODE_RECOVERED) {
 			originStartPos = Mtm->nodes[nodeId-1].restartLsn;
+			MTM_LOG1("Restart replication from node %d from position %lx", nodeId, originStartPos);
 		} 
-		if (originStartPos == 0) { 
+		if (originStartPos == InvalidXLogRecPtr) { 
 			StartTransactionCommand();
 			originName = psprintf(MULTIMASTER_SLOT_PATTERN, nodeId);
 			originId = replorigin_by_name(originName, true);

contrib/mmts/tests2/docker-entrypoint.sh

@@ -61,7 +61,7 @@ if [ "$1" = 'postgres' ]; then
 
 		############################################################################
 
-		CONNSTRS='dbname=postgres host=node1, dbname=postgres host=node2, dbname=postgres host=node3'
+		CONNSTRS='dbname=postgres user=postgres host=node1, dbname=postgres user=postgres host=node2, dbname=postgres user=postgres host=node3'
 		RAFT_PEERS='1:node1:6666, 2:node2:6666, 3:node3:6666'
 
 		cat <<-EOF >> $PGDATA/postgresql.conf