disable node on broken dmq sender connetion

Author: kelvich

src/dmq.c

@@ -136,8 +136,10 @@ static volatile sig_atomic_t got_SIGHUP = false;
 
 static shmem_startup_hook_type PreviousShmemStartupHook;
 
-dmq_receiver_hook_type dmq_receiver_start_hook;
-dmq_receiver_hook_type dmq_receiver_stop_hook;
+dmq_hook_type dmq_receiver_start_hook;
+dmq_hook_type dmq_receiver_stop_hook;
+dmq_hook_type dmq_sender_connect_hook;
+dmq_hook_type dmq_sender_disconnect_hook;
 
 void dmq_sender_main(Datum main_arg);
 
@@ -461,6 +463,8 @@ dmq_sender_main(Datum main_arg)
 								"[DMQ] failed to send message to %s: %s",
 								conns[conn_id].receiver_name,
 								PQerrorMessage(conns[conn_id].pgconn));
+
+						dmq_sender_disconnect_hook(conns[conn_id].receiver_name);
 					}
 					else
 					{
@@ -577,6 +581,8 @@ dmq_sender_main(Datum main_arg)
 								"[DMQ] failed to send heartbeat to %s: %s",
 								conns[conn_id].receiver_name,
 								PQerrorMessage(conns[conn_id].pgconn));
+
+						dmq_sender_disconnect_hook(conns[conn_id].receiver_name);
 					}
 				}
 			}
@@ -681,6 +687,8 @@ dmq_sender_main(Datum main_arg)
 						mtm_log(DmqStateFinal,
 								"[DMQ] Connected to %s",
 								conns[conn_id].receiver_name);
+
+						dmq_sender_connect_hook(conns[conn_id].receiver_name);
 					}
 					break;
 
@@ -695,6 +703,8 @@ dmq_sender_main(Datum main_arg)
 								"[DMQ] connection error with %s: %s",
 								conns[conn_id].receiver_name,
 								PQerrorMessage(conns[conn_id].pgconn));
+
+						dmq_sender_disconnect_hook(conns[conn_id].receiver_name);
 					}
 					break;
 			}

src/include/dmq.h

@@ -29,8 +29,10 @@ extern bool dmq_pop_nb(DmqSenderId *sender_id, StringInfo msg, uint64 mask);
 extern void dmq_push(DmqDestinationId dest_id, char *stream_name, char *msg);
 extern void dmq_push_buffer(DmqDestinationId dest_id, char *stream_name, const void *buffer, size_t len);
 
-typedef void (*dmq_receiver_hook_type) (char *);
-extern dmq_receiver_hook_type dmq_receiver_start_hook;
-extern dmq_receiver_hook_type dmq_receiver_stop_hook;
+typedef void (*dmq_hook_type) (char *);
+extern dmq_hook_type dmq_receiver_start_hook;
+extern dmq_hook_type dmq_receiver_stop_hook;
+extern dmq_hook_type dmq_sender_connect_hook;
+extern dmq_hook_type dmq_sender_disconnect_hook;
 
 #endif

src/include/state.h

@@ -48,6 +48,8 @@ extern void MtmStateProcessEvent(MtmEvent ev, bool locked);
 
 extern void MtmOnNodeDisconnect(char *node_name);
 extern void MtmOnNodeConnect(char *node_name);
+extern void MtmOnDmqSenderConnect(char *node_name);
+extern void MtmOnDmqSenderDisconnect(char *node_name);
 
 extern void MtmRefreshClusterStatus(void);
 

src/multimaster.c

@@ -505,6 +505,8 @@ _PG_init(void)
 	dmq_init();
 	dmq_receiver_start_hook = MtmOnNodeConnect;
 	dmq_receiver_stop_hook = MtmOnNodeDisconnect;
+	dmq_sender_connect_hook = MtmOnDmqSenderConnect;
+	dmq_sender_disconnect_hook = MtmOnDmqSenderDisconnect;
 
 	ResolverInit();
 

src/state.c

@@ -58,9 +58,10 @@ struct MtmState
 {
 	nodemask_t	configured_mask;
 	nodemask_t	connected_mask;
-	nodemask_t	enabled_mask;
+	nodemask_t	dmq_senders_mask;
 	nodemask_t	receivers_mask;
 	nodemask_t	senders_mask;
+	nodemask_t	enabled_mask;
 	nodemask_t	clique;
 
 	bool		referee_grant;
@@ -155,6 +156,7 @@ MtmStateFill(MtmConfig *cfg)
 		mtm_state->recovery_slot = cfg->backup_node_id;
 
 	BIT_SET(mtm_state->connected_mask, cfg->my_node_id - 1);
+	BIT_SET(mtm_state->dmq_senders_mask, cfg->my_node_id - 1);
 
 	/* re-create configured_mask */
 	mtm_state->configured_mask = 0;
@@ -217,7 +219,7 @@ MtmCheckState(void)
 	bool isEnabledState;
 	MtmNodeStatus old_status;
 	int nEnabled    = popcount(mtm_state->enabled_mask);
-	int nConnected  = popcount(mtm_state->connected_mask);
+	int nConnected  = popcount(mtm_state->connected_mask & mtm_state->dmq_senders_mask);
 	int nReceivers  = popcount(mtm_state->receivers_mask);
 	int nSenders    = popcount(mtm_state->senders_mask);
 	int nConfigured = popcount(mtm_state->configured_mask);
@@ -227,9 +229,10 @@ MtmCheckState(void)
 	old_status = mtm_state->status;
 
 	mtm_log(MtmStateMessage,
-		"[STATE]   Status = (enabled=%s, connected=%s, clique=%s, receivers=%s, senders=%s, total=%i, referee_grant=%d)",
+		"[STATE]   Status = (enabled=%s, connected=%s, dmq_senders=%s, clique=%s, receivers=%s, senders=%s, total=%i, referee_grant=%d)",
 		maskToString(mtm_state->enabled_mask),
 		maskToString(mtm_state->connected_mask),
+		maskToString(mtm_state->dmq_senders_mask),
 		maskToString(mtm_state->clique),
 		maskToString(mtm_state->receivers_mask),
 		maskToString(mtm_state->senders_mask),
@@ -493,6 +496,31 @@ MtmOnNodeDisconnect(char *node_name)
 	LWLockRelease(mtm_state->lock);
 }
 
+void
+MtmOnDmqSenderConnect(char *node_name)
+{
+	int			node_id;
+	sscanf(node_name, MTM_DMQNAME_FMT, &node_id);
+
+	LWLockAcquire(mtm_state->lock, LW_EXCLUSIVE);
+	BIT_SET(mtm_state->dmq_senders_mask, node_id - 1);
+	MtmCheckState();
+	LWLockRelease(mtm_state->lock);
+}
+
+void
+MtmOnDmqSenderDisconnect(char *node_name)
+{
+	int			node_id;
+	sscanf(node_name, MTM_DMQNAME_FMT, &node_id);
+
+	LWLockAcquire(mtm_state->lock, LW_EXCLUSIVE);
+	BIT_CLEAR(mtm_state->dmq_senders_mask, node_id - 1);
+	MtmDisableNode(node_id);
+	MtmCheckState();
+	LWLockRelease(mtm_state->lock);
+}
+
 // XXXX: make that event too
 void MtmOnNodeConnect(char *node_name)
 {