Add mtm.resume_node function

Author: knizhnik

multimaster--1.0.sql

@@ -18,19 +18,28 @@ CREATE FUNCTION mtm.stop_replication() RETURNS void
 AS 'MODULE_PATHNAME','mtm_stop_replication'
 LANGUAGE C;
 
+-- Stop replication to the node. Node is didsabled, If drop_slot is true, then replication slot is dropped and node can be recovered using basebackup and recover_node function.
+-- If drop_slot is false and limit for maximal slot gap was not reached, then node can be restarted using resume_node function.
 CREATE FUNCTION mtm.stop_node(node integer, drop_slot bool default false) RETURNS void
 AS 'MODULE_PATHNAME','mtm_stop_node'
 LANGUAGE C;
 
+-- Add new node to the cluster. Number of nodes should not exeed maximal number of nodes in the cluster.
 CREATE FUNCTION mtm.add_node(conn_str text) RETURNS void
 AS 'MODULE_PATHNAME','mtm_add_node'
 LANGUAGE C;
 
--- Create replication slot for the node which was previously stopped 
+-- Create replication slot for the node which was previously stalled (its replicatoin slot was deleted)
 CREATE FUNCTION mtm.recover_node(node integer) RETURNS void
 AS 'MODULE_PATHNAME','mtm_recover_node'
 LANGUAGE C;
 
+-- Resume previously stopped node with live replication slot. If node was not stopped, this function has no effect.
+-- It doesn't create slot and returns error if node is stalled (slot eas dropped)
+CREATE FUNCTION mtm.resume_node(node integer) RETURNS void
+AS 'MODULE_PATHNAME','mtm_resume_node'
+LANGUAGE C;
+
 
 CREATE FUNCTION mtm.get_snapshot() RETURNS bigint
 AS 'MODULE_PATHNAME','mtm_get_snapshot'
@@ -63,11 +72,11 @@ CREATE FUNCTION mtm.get_trans_by_xid(xid bigint) RETURNS mtm.trans_state
 AS 'MODULE_PATHNAME','mtm_get_trans_by_xid'
 LANGUAGE C;
 
-CREATE FUNCTION mtm.get_cluster_state() RETURNS mtm.cluster_state 
+CREATE FUNCTION mtm.get_cluster_state() RETURNS mtm.cluster_state
 AS 'MODULE_PATHNAME','mtm_get_cluster_state'
 LANGUAGE C;
 
-CREATE FUNCTION mtm.collect_cluster_info() RETURNS SETOF mtm.cluster_state 
+CREATE FUNCTION mtm.collect_cluster_info() RETURNS SETOF mtm.cluster_state
 AS 'MODULE_PATHNAME','mtm_collect_cluster_info'
 LANGUAGE C;
 
@@ -105,7 +114,7 @@ LANGUAGE C;
 
 CREATE TABLE IF NOT EXISTS mtm.local_tables(rel_schema name, rel_name name, primary key(rel_schema, rel_name));
 
-CREATE OR REPLACE FUNCTION mtm.alter_sequences() RETURNS boolean AS 
+CREATE OR REPLACE FUNCTION mtm.alter_sequences() RETURNS boolean AS
 $$
 DECLARE
     seq_class record;

multimaster.c

@@ -117,6 +117,7 @@ PG_FUNCTION_INFO_V1(mtm_stop_node);
 PG_FUNCTION_INFO_V1(mtm_add_node);
 PG_FUNCTION_INFO_V1(mtm_poll_node);
 PG_FUNCTION_INFO_V1(mtm_recover_node);
+PG_FUNCTION_INFO_V1(mtm_resume_node);
 PG_FUNCTION_INFO_V1(mtm_get_snapshot);
 PG_FUNCTION_INFO_V1(mtm_get_csn);
 PG_FUNCTION_INFO_V1(mtm_get_trans_by_gid);
@@ -2002,7 +2003,7 @@ static void MtmDisableNode(int nodeId)
  */
 static void MtmEnableNode(int nodeId)
 {
-	if (BIT_SET(Mtm->disabledNodeMask, nodeId-1)) {
+	if (BIT_CHECK(Mtm->disabledNodeMask, nodeId-1)) {
 		BIT_CLEAR(Mtm->disabledNodeMask, nodeId-1);
 		BIT_CLEAR(Mtm->reconnectMask, nodeId-1);
 		BIT_SET(Mtm->recoveredNodeMask, nodeId-1);
@@ -3678,7 +3679,7 @@ void MtmRecoverNode(int nodeId)
 		MTM_ELOG(ERROR, "NodeID %d is out of range [1,%d]", nodeId, Mtm->nAllNodes);
 	}
 	MtmLock(LW_EXCLUSIVE);
-	if (BIT_SET(Mtm->stoppedNodeMask, nodeId-1))
+	if (BIT_CHECK(Mtm->stoppedNodeMask, nodeId-1))
 	{
 		Assert(BIT_CHECK(Mtm->disabledNodeMask, nodeId-1));
 		BIT_CLEAR(Mtm->stoppedNodeMask, nodeId-1);
@@ -3693,6 +3694,36 @@ void MtmRecoverNode(int nodeId)
 	}
 }
 
+/*
+ * Resume previosly stopped node.
+ * This function creates logical replication slot for the node which will collect
+ * all changes which should be sent to this node from this moment.
+ */
+void MtmResumeNode(int nodeId)
+{
+	if (nodeId <= 0 || nodeId > Mtm->nAllNodes)
+	{
+		MTM_ELOG(ERROR, "NodeID %d is out of range [1,%d]", nodeId, Mtm->nAllNodes);
+	}
+	MtmLock(LW_EXCLUSIVE);
+	if (BIT_CHECK(Mtm->stalledNodeMask, nodeId-1))
+	{
+		MtmUnlock();
+		MTM_ELOG(ERROR, "Node %d can not be resumed because it's replication slot is dropped", nodeId);
+	}
+	if (BIT_CHECK(Mtm->stoppedNodeMask, nodeId-1))
+	{
+		Assert(BIT_CHECK(Mtm->disabledNodeMask, nodeId-1));
+		BIT_CLEAR(Mtm->stoppedNodeMask, nodeId-1);
+	}
+	MtmUnlock();
+
+	if (!MtmIsBroadcast())
+	{
+		MtmBroadcastUtilityStmt(psprintf("select mtm.resume_node(%d)", nodeId), true);
+	}
+}
+
 /*
  * Permanently exclude node from the cluster. Node will not participate in voting and can not be automatically recovered
  * until MtmRecoverNode is invoked.
@@ -3785,6 +3816,12 @@ MtmReplicationStartupHook(struct PGLogicalStartupHookArgs* args)
 	}
 	MTM_LOG1("Startup of logical replication to node %d", MtmReplicationNodeId);
 	MtmLock(LW_EXCLUSIVE);
+
+	if (BIT_CHECK(Mtm->stalledNodeMask, MtmReplicationNodeId-1)) {
+		MtmUnlock();
+		MTM_ELOG(ERROR, "Stalled node %d tries to initiate recovery", MtmReplicationNodeId);
+	}
+
 	if (BIT_CHECK(Mtm->stoppedNodeMask, MtmReplicationNodeId-1)) {
 		MTM_ELOG(WARNING, "Stopped node %d tries to initiate recovery", MtmReplicationNodeId);
 		do {
@@ -4171,6 +4208,14 @@ mtm_recover_node(PG_FUNCTION_ARGS)
 	PG_RETURN_VOID();
 }
 
+Datum
+mtm_resume_node(PG_FUNCTION_ARGS)
+{
+	int nodeId = PG_GETARG_INT32(0);
+	MtmResumeNode(nodeId);
+	PG_RETURN_VOID();
+}
+
 Datum
 mtm_get_snapshot(PG_FUNCTION_ARGS)
 {

multimaster.h

@@ -400,6 +400,7 @@ extern void  MtmUnlockNode(int nodeId);
 extern void  MtmStopNode(int nodeId, bool dropSlot);
 extern void  MtmReconnectNode(int nodeId);
 extern void  MtmRecoverNode(int nodeId);
+extern void  MtmResumeNode(int nodeId);
 extern void  MtmOnNodeDisconnect(int nodeId);
 extern void  MtmOnNodeConnect(int nodeId);
 extern void  MtmWakeUpBackend(MtmTransState* ts);