Cleanup disconnected bit for recovered node

Author: knizhnik

contrib/mmts/multimaster.c

@@ -842,8 +842,9 @@ void MtmSendNotificationMessage(MtmTransState* ts, MtmMessageCode cmd)
 
 void MtmRecoveryCompleted(void)
 {
-	elog(WARNING, "Recevoery of node %d is completed", MtmNodeId);
+	elog(WARNING, "Recovery of node %d is completed", MtmNodeId);
 	Mtm->recoverySlot = 0;
+	BIT_CLEAR(Mtm->disabledNodeMask, MtmNodeId-1);
 	MtmSwitchClusterMode(MTM_ONLINE);
 }
 
@@ -1745,10 +1746,6 @@ MtmReplicationStartupHook(struct PGLogicalStartupHookArgs* args)
 			break;
 		}
 	}
-	if (isRecoverySession) { 
-		MTM_INFO("%d: PGLOGICAL startup hook\n", MyProcPid);
-		sleep(30);
-	}
 	MtmLock(LW_EXCLUSIVE);
 	if (isRecoverySession) {
 		elog(WARNING, "Node %d start recovery of node %d", MtmNodeId, MtmReplicationNodeId);

contrib/mmts/pglogical_receiver.c

@@ -53,7 +53,7 @@ static volatile sig_atomic_t got_sighup = false;
 
 /* GUC variables */
 static int receiver_idle_time = 0;
-static bool receiver_sync_mode = false;
+static bool receiver_sync_mode = true;
 
 /* Worker name */
 char worker_proc[BGW_MAXLEN];
@@ -292,7 +292,7 @@ pglogical_receiver_main(Datum main_arg)
 	}
 	CommitTransactionCommand();
 
-	appendPQExpBuffer(query, "START_REPLICATION SLOT \"%s\" LOGICAL %u/%u (\"startup_params_format\" '1', \"max_proto_version\" '%d',  \"min_proto_version\" '%d', \"mtm_replication_mode\" '%s')",
+	appendPQExpBuffer(query, "START_REPLICATION SLOT \"%s\" LOGICAL %u/%u (\"startup_params_format\" '1', \"max_proto_version\" '%d',  \"min_proto_version\" '%d', \"forward_changesets\" '1', \"mtm_replication_mode\" '%s')",
 					  args->receiver_slot,
 					  (uint32) (originStartPos >> 32),
 					  (uint32) originStartPos,

contrib/mmts/raftable.c

@@ -0,0 +1,38 @@
+#include <dlfcn.h>
+#include "postgres.h"
+#include "raftable.h"
+
+
+static raftable_get_t raftable_get_impl;
+static raftable_set_t raftable_set_impl;
+
+static void RaftableResolve()
+{
+	if (raftable_get_impl == NULL) { 
+		void* dll = dlopen(NULL, RTLD_NOW);
+		raftable_get_impl = dlsym(dll, "raftable_get");
+		raftable_set_impl = dlsym(dll, "raftable_set");
+		Assert(raftable_get_impl != NULL && raftable_set_impl != NULL);
+	}
+}
+
+/*
+ * Raftable function proxies
+ */
+void* RaftableGet(char const* key, int* size, RaftableTimestamp* ts, bool nowait)
+{
+	if (!MtmUseRaftable) { 
+		return NULL;
+	}
+	RaftableResolve();
+	return (*raftable_get_impl)(key, size, nowait ? 0 : -1);
+}
+
+
+void RaftableSet(char const* key, void const* value, int size, bool nowait)
+{
+	if (MtmUseRaftable) { 
+		RaftableResolve();
+		(*raftable_set_impl)(key, value, size, nowait ? 0 : -1);
+	}		
+}

src/backend/replication/walsender.c

@@ -992,6 +992,8 @@ StartLogicalReplication(StartReplicationCmd *cmd)
 	 * Initialize position to the last ack'ed one, then the xlog records begin
 	 * to be shipped from that position.
 	 */
+	elog(WARNING, "%d: StartLogicalReplication for slot %s startpoint %lx", MyProcPid, NameStr(MyReplicationSlot->data.name), cmd->startpoint);
+
 	logical_decoding_ctx = CreateDecodingContext(
 											   cmd->startpoint, cmd->options,
 												 logical_read_xlog_page,

src/include/miscadmin.h

@@ -114,7 +114,7 @@ do { \
 
 #define RESUME_INTERRUPTS() \
 do { \
-	Assert(InterruptHoldoffCount > 0); \
+	if (InterruptHoldoffCount <= 0) *(int*)0 = 0;	\
 	InterruptHoldoffCount--; \
 } while(0)