restart receivers on node disables too

Author: kelvich

Cluster.pm

@@ -16,7 +16,7 @@ sub new
 
 	my $self = {
 		nodes => \@nodes,
-		recv_timeout => 3
+		recv_timeout => 6
 	};
 
 	bless $self, $class;

src/include/state.h

@@ -62,6 +62,7 @@ extern nodemask_t MtmGetDisabledNodeMask(void);
 extern nodemask_t MtmGetConnectedNodeMask(void);
 extern nodemask_t MtmGetEnabledNodeMask(void);
 extern int MtmGetRecoveryCount(void);
+extern int MtmGetNodeDisableCount(int node_id);
 
 extern nodemask_t MtmGetDisabledNodeMask(void);
 

src/pglogical_receiver.c

@@ -564,7 +564,8 @@ pglogical_receiver_main(Datum main_arg)
 	 */
 	for(;;)
 	{
-		int	 count;
+		int			count,
+					counterpart_disable_count;
 		XLogRecPtr		remote_start = InvalidXLogRecPtr;
 		Syncpoint	   *spvector = NULL;
 		HTAB		   *filter_map = NULL;
@@ -586,11 +587,8 @@ pglogical_receiver_main(Datum main_arg)
 		Mtm->peers[nodeId - 1].receiver_mode = mode;
 		LWLockRelease(Mtm->lock);
 
-		// XXX: delete unnecessary modes
-		Assert(mode == REPLMODE_RECOVERY || mode == REPLMODE_RECOVERED);
-
-		// XXX: avoid that
 		count = MtmGetRecoveryCount();
+		counterpart_disable_count = MtmGetNodeDisableCount(nodeId);
 
 		/* Establish connection to remote server */
 		conn = receiver_connect(MtmNodeById(receiver_mtm_cfg, nodeId)->conninfo);
@@ -715,12 +713,17 @@ pglogical_receiver_main(Datum main_arg)
 				proc_exit(1);
 			}
 
-
 			if (count != MtmGetRecoveryCount()) {
 				ereport(LOG, (MTM_ERRMSG("%s: restart WAL receiver because node was recovered", worker_proc)));
 				goto OnError;
 			}
 
+			if (counterpart_disable_count != MtmGetNodeDisableCount(nodeId))
+			{
+				ereport(LOG, (MTM_ERRMSG("%s: restart WAL receiver because counterpart was disabled", worker_proc)));
+				goto OnError;
+			}
+
 			/*
 			 * Receive data.
 			 */

src/state.c

@@ -71,6 +71,7 @@ struct MtmState
 	int			recovery_slot;
 
 	int			recovery_count;
+	int			node_recovery_count[MTM_MAX_NODES];
 
 	LWLock	   *lock;
 
@@ -449,6 +450,7 @@ MtmDisableNode(int node_id)
 	mtm_log(MtmStateMessage, "[STATE] Node %i: disabled", node_id);
 
 	BIT_CLEAR(mtm_state->enabled_mask, node_id - 1);
+	mtm_state->node_recovery_count[node_id - 1] += 1;
 
 	// XXX my_node_id
 	if (mtm_state->status == MTM_ONLINE)
@@ -1077,6 +1079,18 @@ MtmGetRecoveryCount()
 	return mtm_state->recovery_count;
 }
 
+int
+MtmGetNodeDisableCount(int node_id)
+{
+	int count;
+
+	LWLockAcquire(mtm_state->lock, LW_SHARED);
+	count = mtm_state->node_recovery_count[node_id - 1];
+	LWLockRelease(mtm_state->lock);
+
+	return count;
+}
+
 // XXX: During evaluation of (mtm.node_info(id)).* this function called
 // once each columnt for every row. So may be just rewrite to SRF.
 Datum