restart walsender on recevery_count increment

Author: kelvich

src/include/pglogical_output.h

@@ -47,6 +47,7 @@ typedef struct MtmDecoderPrivate
 {
 	bool	is_recovery;
 	bool	recovery_done;
+	int		recovery_count;
 	int64	session_id;
 	MtmConfig *cfg;
 } MtmDecoderPrivate;

src/pglogical_output.c

@@ -38,6 +38,8 @@
 
 #include "storage/ipc.h"
 
+#include "tcop/tcopprot.h"
+
 #include "utils/builtins.h"
 #include "utils/catcache.h"
 #include "utils/guc.h"
@@ -53,6 +55,7 @@
 
 #include "multimaster.h"
 #include "logger.h"
+#include "state.h"
 
 extern void		_PG_output_plugin_init(OutputPluginCallbacks *cb);
 
@@ -464,6 +467,17 @@ pg_decode_caughtup(LogicalDecodingContext *ctx)
 	PGLogicalOutputData *data = (PGLogicalOutputData *) ctx->output_plugin_private;
 	MtmDecoderPrivate *hooks_data = (MtmDecoderPrivate *) data->hooks.hooks_private_data;
 
+
+	if (hooks_data->recovery_count != MtmGetRecoveryCount())
+	{
+		mtm_log(LOG, "exiting due to disabled status");
+		if (whereToSendOutput == DestRemote)
+			whereToSendOutput = DestNone;
+
+		proc_exit(0);
+		abort();					/* keep the compiler quiet */
+	}
+
 	/*
 	 * MtmOutputPluginPrepareWrite send some bytes to downstream,
 	 * so we must avoid calling it in normal (non-recovery) situation.

src/pglogical_proto.c

@@ -721,6 +721,7 @@ MtmReplicationStartupHook(struct PGLogicalStartupHookArgs* args)
 	hooks_data->recovery_done = false;
 	hooks_data->is_recovery = false;
 	hooks_data->cfg = MtmLoadConfig();
+	hooks_data->recovery_count = MtmGetRecoveryCount();
 
 	foreach(param, args->in_params)
 	{

src/state.c

@@ -394,7 +394,7 @@ MtmStateProcessEvent(MtmEvent ev, bool locked)
 	{
 		case MTM_CLIQUE_DISABLE:
 			BIT_CLEAR(mtm_state->enabled_mask, Mtm->my_node_id - 1);
-			mtm_state->recovery_count++; /* this will restart replication connection */
+			// mtm_state->recovery_count++; /* this will restart replication connection */
 			break;
 
 		case MTM_REMOTE_DISABLE:
@@ -413,7 +413,7 @@ MtmStateProcessEvent(MtmEvent ev, bool locked)
 		case MTM_RECOVERY_FINISH2:
 			{
 				mtm_state->recovered = true;
-				mtm_state->recovery_count++; /* this will restart replication connection */
+				// mtm_state->recovery_count++; /* this will restart replication connection */
 				mtm_state->recovery_slot = 0;
 			}
 			break;