Fix XactLogAbortRecord

Author: knizhnik

contrib/mmts/multimaster.c

@@ -3320,6 +3320,8 @@ MtmReplicationStartupHook(struct PGLogicalStartupHookArgs* args)
 				MTM_LOG1("Recovered position of node %d is %lx", MtmReplicationNodeId, recoveredLSN); 
 				if (Mtm->nodes[MtmReplicationNodeId-1].restartLSN < recoveredLSN) { 
 					MTM_LOG2("[restartlsn] node %d: %lx -> %lx (MtmReplicationStartupHook)", MtmReplicationNodeId, Mtm->nodes[MtmReplicationNodeId-1].restartLSN, recoveredLSN);
+					Assert(Mtm->nodes[MtmReplicationNodeId-1].restartLSN == InvalidXLogRecPtr 
+						   || recoveredLSN < Mtm->nodes[MtmReplicationNodeId-1].restartLSN + MtmMaxRecoveryLag);
 					Mtm->nodes[MtmReplicationNodeId-1].restartLSN = recoveredLSN;
 				}
 			} else { 
@@ -3533,7 +3535,8 @@ bool MtmFilterTransaction(char* record, int size)
 	}
 	restart_lsn = origin_node == MtmReplicationNodeId ? end_lsn : origin_lsn;
     if (Mtm->nodes[origin_node-1].restartLSN < restart_lsn) {
-		Assert(restart_lsn != InvalidXLogRecPtr);
+		Assert(Mtm->nodes[origin_node-1].restartLSN == InvalidXLogRecPtr 
+			   || restart_lsn < Mtm->nodes[origin_node-1].restartLSN + MtmMaxRecoveryLag);
 		MTM_LOG2("[restartlsn] node %d: %lx -> %lx (MtmFilterTransaction)", MtmReplicationNodeId, Mtm->nodes[MtmReplicationNodeId-1].restartLSN, restart_lsn);
 		Mtm->nodes[origin_node-1].restartLSN = restart_lsn;
     } else {

contrib/mmts/tests2/test_regression.py

@@ -15,7 +15,7 @@ def setUpClass(self):
     @classmethod
     def tearDownClass(self):
         print('tearDown')
-        #subprocess.check_call(['docker-compose','down'])
+        subprocess.check_call(['docker-compose','down'])
 
     def test_regression(self):
         # XXX: make smth clever here

src/backend/access/rmgrdesc/xactdesc.c

@@ -252,7 +252,7 @@ xact_desc_commit(StringInfo buf, uint8 info, xl_xact_commit *xlrec, RepOriginId
 }
 
 static void
-xact_desc_abort(StringInfo buf, uint8 info, xl_xact_abort *xlrec)
+xact_desc_abort(StringInfo buf, uint8 info, xl_xact_abort *xlrec, RepOriginId origin_id)
 {
 	xl_xact_parsed_abort parsed;
 	int			i;
@@ -282,6 +282,14 @@ xact_desc_abort(StringInfo buf, uint8 info, xl_xact_abort *xlrec)
 		for (i = 0; i < parsed.nsubxacts; i++)
 			appendStringInfo(buf, " %u", parsed.subxacts[i]);
 	}
+	if (parsed.xinfo & XACT_XINFO_HAS_ORIGIN)
+	{
+		appendStringInfo(buf, "; origin: node %u, lsn %X/%X, at %s",
+						 origin_id,
+						 (uint32) (parsed.origin_lsn >> 32),
+						 (uint32) parsed.origin_lsn,
+						 timestamptz_to_str(parsed.origin_timestamp));
+	}
 }
 
 static void
@@ -312,7 +320,8 @@ xact_desc(StringInfo buf, XLogReaderState *record)
 	{
 		xl_xact_abort *xlrec = (xl_xact_abort *) rec;
 
-		xact_desc_abort(buf, XLogRecGetInfo(record), xlrec);
+		xact_desc_abort(buf, XLogRecGetInfo(record), xlrec,
+						XLogRecGetOrigin(record));
 	}
 	else if (info == XLOG_XACT_ASSIGNMENT)
 	{

src/backend/access/transam/xact.c

@@ -5358,9 +5358,6 @@ XactLogAbortRecord(TimestampTz abort_time,
 		XLogRegisterData((char *) twophase_gid, xl_twophase.gidlen);
 	}
 
-	if (xl_xinfo.xinfo & XACT_XINFO_HAS_DBINFO)
-		XLogRegisterData((char *) (&xl_dbinfo), sizeof(xl_dbinfo));
-
 	if (xl_xinfo.xinfo & XACT_XINFO_HAS_ORIGIN)
 		XLogRegisterData((char *) (&xl_origin), sizeof(xl_xact_origin));