More recovery fixes

Author: knizhnik

arbiter.c

@@ -1003,7 +1003,7 @@ static void MtmTransReceiver(Datum arg)
 				}
 			}
 		}
-		if (Mtm->status != MTM_RECOVERY) { 
+		if (Mtm->status == MTM_ONLINE) { 
 			now = MtmGetSystemTime();
 			if (now > lastHeartbeatCheck + MSEC_TO_USEC(MtmHeartbeatRecvTimeout)) { 
 				if (!MtmWatchdog(stopPolling)) { 

multimaster.c

@@ -40,7 +40,6 @@
 #include "utils/builtins.h"
 #include "utils/memutils.h"
 #include "commands/dbcommands.h"
-#include "miscadmin.h"
 #include "postmaster/autovacuum.h"
 #include "storage/pmsignal.h"
 #include "storage/proc.h"
@@ -1214,12 +1213,16 @@ 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);
 	MtmLock(LW_EXCLUSIVE);
 	Mtm->recoverySlot = 0;
 	Mtm->nodes[MtmNodeId-1].lastStatusChangeTime = MtmGetSystemTime();
 	BIT_CLEAR(Mtm->disabledNodeMask, MtmNodeId-1);
+	for (i = 0; i < Mtm->nAllNodes; i++) { 
+		Mtm->nodes[i].lastHeartbeat = 0; /* defuse watchdog until first heartbeat is received */
+	}
 	/* Mode will be changed to online once all logical reciever are connected */
 	MtmSwitchClusterMode(MTM_CONNECTED);
 	MtmUnlock();
@@ -2297,7 +2300,7 @@ void MtmReceiverStarted(int nodeId)
  * Druing recovery we need to open only one replication slot from which node should receive all transactions.
  * Slots at other nodes should be removed 
  */
-MtmSlotMode MtmReceiverSlotMode(int nodeId)
+MtmReplicationMode MtmGetReplicationMode(int nodeId)
 {
 	bool recovery = false;
 	while (Mtm->status != MTM_CONNECTED && Mtm->status != MTM_ONLINE) { 		
@@ -2312,7 +2315,7 @@ MtmSlotMode MtmReceiverSlotMode(int nodeId)
 				Mtm->recoveryCount += 1;
 				Mtm->pglogicalNodeMask = 0;
 				FinishAllPreparedTransactions(false);
-				return SLOT_OPEN_EXISTED;
+				return REPLMODE_RECOVERY;
 			}
 		}
 		/* delay opening of other slots until recovery is completed */
@@ -2324,7 +2327,7 @@ MtmSlotMode MtmReceiverSlotMode(int nodeId)
 		MTM_LOG2("%d: Reuse 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 ? SLOT_CREATE_NEW : SLOT_OPEN_ALWAYS;
+	return recovery ? REPLMODE_RECOVERED : REPLMODE_NORMAL;
 }
 
 static bool MtmIsBroadcast() 

multimaster.h

@@ -117,10 +117,10 @@ typedef enum
 
 typedef enum
 {
-	SLOT_CREATE_NEW,   /* create new slot (drop existed) */
-	SLOT_OPEN_EXISTED, /* open existed slot */
-	SLOT_OPEN_ALWAYS,  /* open existed slot or create new if not exists */
-} MtmSlotMode;
+	REPLMODE_RECOVERED,  /* recovery of node is completed so drop old slot and restart replication from the current position in WAL */
+	REPLMODE_RECOVERY,   /* perform recorvery of the node by applying all data from the slot from specified point */
+	REPLMODE_NORMAL      /* normal mode: use existed slot or create new one and start receiving data from it from the specified position */
+} MtmReplicationMode;
 
 typedef struct
 {
@@ -244,7 +244,7 @@ extern csn_t MtmAssignCSN(void);
 extern csn_t MtmSyncClock(csn_t csn);
 extern void  MtmJoinTransaction(GlobalTransactionId* gtid, csn_t snapshot);
 extern void  MtmReceiverStarted(int nodeId);
-extern MtmSlotMode MtmReceiverSlotMode(int nodeId);
+extern MtmReplicationMode MtmGetReplicationMode(int nodeId);
 extern void  MtmExecute(void* work, int size);
 extern void  MtmExecutor(int id, void* work, size_t size);
 extern void  MtmSendNotificationMessage(MtmTransState* ts, MtmMessageCode cmd);

pglogical_output.c

@@ -47,6 +47,7 @@
 #include "utils/relcache.h"
 #include "utils/syscache.h"
 #include "utils/typcache.h"
+#include "miscadmin.h"
 
 extern void		_PG_output_plugin_init(OutputPluginCallbacks *cb);
 
@@ -155,6 +156,8 @@ pg_decode_startup(LogicalDecodingContext * ctx, OutputPluginOptions *opt,
 {
 	PGLogicalOutputData  *data = palloc0(sizeof(PGLogicalOutputData));
 
+	elog(LOG, "%d: pg_decode_startup is_init=%d", MyProcPid, is_init);
+
 	data->context = AllocSetContextCreate(TopMemoryContext,
 										  "pglogical conversion context",
 										  ALLOCSET_DEFAULT_MINSIZE,

pglogical_proto.c

@@ -433,8 +433,7 @@ decide_datum_transfer(Form_pg_attribute att, Form_pg_type typclass,
 PGLogicalProtoAPI *
 pglogical_init_api(PGLogicalProtoType typ)
 {
-    PGLogicalProtoAPI* res = malloc(sizeof(PGLogicalProtoAPI));
-	MemSet(res, 0, sizeof(PGLogicalProtoAPI));
+    PGLogicalProtoAPI* res = palloc0(sizeof(PGLogicalProtoAPI));
 	sscanf(MyReplicationSlot->data.name.data, MULTIMASTER_SLOT_PATTERN, &MtmReplicationNodeId);
 	MTM_LOG1("%d: PRGLOGICAL init API for slot %s node %d", MyProcPid, MyReplicationSlot->data.name.data, MtmReplicationNodeId);
     res->write_rel = pglogical_write_rel;

pglogical_receiver.c

@@ -193,9 +193,9 @@ feTimestampDifference(int64 start_time, int64 stop_time,
 
 static char const* const MtmReplicationModeName[] = 
 {
-	"recovered", /* SLOT_CREATE_NEW: recovery of node is completed so drop old slot and restart replication from the current position in WAL */
-	"recovery",  /* SLOT_OPEN_EXISTED: perform recorvery of the node by applying all data from theslot from specified point */
-	"normal"     /* SLOT_OPEN_ALWAYS: normal mode: use existed slot or create new one and start receiving data from it from the specified position */
+	"recovered", /* recovery of node is completed so drop old slot and restart replication from the current position in WAL */
+	"recovery",  /* perform recorvery of the node by applying all data from theslot from specified point */
+	"normal"     /* normal mode: use existed slot or create new one and start receiving data from it from the specified position */
 };
 
 static void
@@ -206,7 +206,7 @@ pglogical_receiver_main(Datum main_arg)
 	PQExpBuffer query;
 	PGconn *conn;
 	PGresult *res;
-	MtmSlotMode mode;
+	MtmReplicationMode mode;
 
     ByteBuffer buf;
 	XLogRecPtr originStartPos = 0;
@@ -251,7 +251,7 @@ pglogical_receiver_main(Datum main_arg)
 		 * Druing recovery we need to open only one replication slot from which node should receive all transactions.
 		 * Slots at other nodes should be removed 
 		 */
-		mode = MtmReceiverSlotMode(nodeId);	
+		mode = MtmGetReplicationMode(nodeId);	
 		count = Mtm->recoveryCount;
 
 		/* Establish connection to remote server */
@@ -264,14 +264,19 @@ pglogical_receiver_main(Datum main_arg)
 		}
 
 		query = createPQExpBuffer();
-		
-		if (mode == SLOT_CREATE_NEW) {
+#if 1 /* Do we need to recretate slot ? */		
+		if (mode == REPLMODE_RECOVERED) { /* recreate slot */
 			appendPQExpBuffer(query, "DROP_REPLICATION_SLOT \"%s\"", slotName);
 			res = PQexec(conn, query->data);
 			PQclear(res);
 			resetPQExpBuffer(query);
 		}
-		if (mode != SLOT_OPEN_EXISTED) { 
+#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. */
+		/* if (mode != REPLMODE_REPLICATION) */
+		{ 
 			appendPQExpBuffer(query, "CREATE_REPLICATION_SLOT \"%s\" LOGICAL \"%s\"", slotName, MULTIMASTER_NAME);
 			res = PQexec(conn, query->data);
 			if (PQresultStatus(res) != PGRES_TUPLES_OK)
@@ -291,24 +296,28 @@ pglogical_receiver_main(Datum main_arg)
 		}
 
 		/* Start logical replication at specified position */
-		StartTransactionCommand();
-		originName = psprintf(MULTIMASTER_SLOT_PATTERN, nodeId);
-		originId = replorigin_by_name(originName, true);
-		if (originId == InvalidRepOriginId) { 
-			originId = replorigin_create(originName);
-			/* 
-			 * We are just creating new replication slot.
-			 * It is assumed that state of local and remote nodes is the same at this moment.
-			 * Them are either empty, either new node is synchronized using base_backup.
-			 * So we assume that LSNs are the same for local and remote node
-			 */
-			originStartPos = Mtm->status == MTM_RECOVERY ? GetXLogInsertRecPtr() : 0;
-			MTM_LOG1("Start logical receiver at position %lx from node %d", originStartPos, nodeId);
+		if (mode == REPLMODE_RECOVERED) {
+			originStartPos = 0;
 		} else { 
-			originStartPos = replorigin_get_progress(originId, false);
-			MTM_LOG1("Restart logical receiver at position %lx with origin=%d from node %d", originStartPos, originId, nodeId);
+			StartTransactionCommand();
+			originName = psprintf(MULTIMASTER_SLOT_PATTERN, nodeId);
+			originId = replorigin_by_name(originName, true);
+			if (originId == InvalidRepOriginId) { 
+				originId = replorigin_create(originName);
+				/* 
+				 * We are just creating new replication slot.
+				 * It is assumed that state of local and remote nodes is the same at this moment.
+				 * Them are either empty, either new node is synchronized using base_backup.
+				 * So we assume that LSNs are the same for local and remote node
+				 */
+				originStartPos = Mtm->status == MTM_RECOVERY ? GetXLogInsertRecPtr() : 0;
+				MTM_LOG1("Start logical receiver at position %lx from node %d", originStartPos, nodeId);
+			} else { 
+				originStartPos = replorigin_get_progress(originId, false);
+				MTM_LOG1("Restart logical receiver at position %lx with origin=%d from node %d", originStartPos, originId, nodeId);
+			}
+			CommitTransactionCommand();
 		}
-		CommitTransactionCommand();
 
 		appendPQExpBuffer(query, "START_REPLICATION SLOT \"%s\" LOGICAL %x/%x (\"startup_params_format\" '1', \"max_proto_version\" '%d',  \"min_proto_version\" '%d', \"forward_changesets\" '1', \"mtm_replication_mode\" '%s')",
 						  slotName,