Keep data for disabled nodes

Author: knizhnik

arbiter.c

@@ -361,9 +361,14 @@ static void MtmAcceptOneConnection()
 			close(fd);
 		} else{ 			
 			Assert(msg.node > 0 && msg.node <= MtmNodes && msg.node != MtmNodeId);
-			elog(NOTICE, "Arbiter established connection with node %d", msg.node); 
-			MtmRegisterSocket(fd, msg.node-1);
-			sockets[msg.node-1] = fd;
+			if (BIT_SET(ds->disabledNodeMask, msg.node-1)) { 
+				elog(WARNING, "Reject attempt to reconnect from disabled node %d", msg.node); 
+				close(fd);
+			} else { 
+				elog(NOTICE, "Arbiter established connection with node %d", msg.node); 
+				MtmRegisterSocket(fd, msg.node-1);
+				sockets[msg.node-1] = fd;
+			}
 		}
 	}
 }

multimaster.c

@@ -902,7 +902,7 @@ void MtmDropNode(int nodeId, bool dropSlot)
 		}
 		if (dropSlot) 
 		{
-			ReplicationSlotDrop(psprintf("mtm_slot_%d", nodeId));
+			ReplicationSlotDrop(psprintf(MULTIMASTER_SLOT_PATTERN, nodeId));
 		}		
 	}
 }

multimaster.h

@@ -17,6 +17,7 @@
 #define MULTIMASTER_NAME           "mtm"
 #define MULTIMASTER_SCHEMA_NAME    "mtm"
 #define MULTIMASTER_DDL_TABLE      "ddl_log"
+#define MULTIMASTER_SLOT_PATTERN   "mtm_slot_%d"
 
 #define Natts_mtm_ddl_log 2
 #define Anum_mtm_ddl_log_issued		1

pglogical_proto.c

@@ -39,6 +39,8 @@
 typedef struct PGLogicalProtoMM
 {
     PGLogicalProtoAPI api;
+	MtmState* state;
+	int  nodeId;
     bool isLocal;
 } PGLogicalProtoMM;
 
@@ -106,7 +108,7 @@ pglogical_write_begin(StringInfo out, PGLogicalOutputData *data,
     PGLogicalProtoMM* mm = (PGLogicalProtoMM*)data->api;
 	csn_t csn = MtmTransactionSnapshot(txn->xid);
 	MTM_TRACE("pglogical_write_begin %d CSN=%ld\n", txn->xid, csn);
-    if (csn == INVALID_CSN) {
+    if (csn == INVALID_CSN || BIT_SET(mm->state->disabledNodeMask, mm->nodeId-1)) {
         mm->isLocal = true;
     } else { 
         mm->isLocal = false;        
@@ -377,6 +379,8 @@ pglogical_init_api(PGLogicalProtoType typ)
 	PGLogicalProtoMM* pmm = palloc0(sizeof(PGLogicalProtoMM));
     PGLogicalProtoAPI* res = &pmm->api;
     pmm->isLocal = false;
+	pmm->state = MtmGetState();
+	sscanf(MyReplicationSlot->data.name.data, MULTIMASTER_SLOT_PATTERN, &pmm->nodeId);
     res->write_rel = pglogical_write_rel;
     res->write_begin = pglogical_write_begin;
     res->write_commit = pglogical_write_commit;