Drop replication sets and slots when dropping node

PJMODOS · PJMODOS · commit ca53820f5974 · 2015-12-10T12:31:12.000+01:00
diff --git a/Makefile b/Makefile
@@ -18,7 +18,7 @@ SCRIPTS_built = pglogical_create_subscriber
 PG_CPPFLAGS = -I$(libpq_srcdir)
 SHLIB_LINK = $(libpq)
 
-REGRESS = init basic extended toasted replication_set add_table matview bidirectional
+REGRESS = init basic extended toasted replication_set add_table matview bidirectional drop
 
 ifdef PG94
 PG_CPPFLAGS += -Icompat
diff --git a/pglogical_functions.c b/pglogical_functions.c
@@ -37,6 +37,7 @@
 
 #include "replication/origin.h"
 #include "replication/reorderbuffer.h"
+#include "replication/slot.h"
 
 #include "storage/latch.h"
 #include "storage/proc.h"
@@ -140,7 +141,7 @@ pglogical_create_node(PG_FUNCTION_ARGS)
 /*
  * Drop the named node.
  *
- * TODO: cascade support
+ * TODO: support cascade (drop subscribers)
  */
 Datum
 pglogical_drop_node(PG_FUNCTION_ARGS)
@@ -165,13 +166,57 @@ pglogical_drop_node(PG_FUNCTION_ARGS)
 					 errmsg("cannot drop node \"%s\" because it still has subscriptions associated with it", node_name),
 					 errhint("drop the subscriptions first")));
 
-		/* Drop all the interfaces. */
-		drop_node_interfaces(node->id);
-
 		/* If the node is local node, drop the record as well. */
 		local_node = get_local_node(true);
 		if (local_node && local_node->node->id == node->id)
+		{
+			int		slotno;
+
+			/* Also drop all the slots associated with the node. */
+			for (slotno = 0; slotno < max_replication_slots; slotno++)
+			{
+				ReplicationSlot *slot = &ReplicationSlotCtl->replication_slots[slotno];
+
+				LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
+				SpinLockAcquire(&slot->mutex);
+				/*
+				 * Check the slot is in use and it's slot belonging to current
+				 * pglogical node.
+				 */
+				if (!slot->in_use ||
+					slot->data.database != MyDatabaseId ||
+					namestrcmp(&slot->data.plugin, "pglogical_output") != 0 ||
+					strncmp(NameStr(slot->data.name), "pgl_", 4) != 0)
+				{
+					SpinLockRelease(&slot->mutex);
+					LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
+					continue;
+				}
+
+				if (slot->active_pid != 0)
+				{
+					SpinLockRelease(&slot->mutex);
+					LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
+					ereport(ERROR,
+							(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+							 errmsg("cannot drop node \"%s\" because replication slot \"%s\" on the node is still active",
+									node_name, NameStr(slot->data.name)),
+							 errhint("drop the subscriptions first")));
+				}
+				SpinLockRelease(&slot->mutex);
+				LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
+
+				ReplicationSlotDrop(NameStr(slot->data.name));
+			}
+
 			drop_local_node();
+		}
+
+		/* Drop all the interfaces. */
+		drop_node_interfaces(node->id);
+
+		/* Drop replication sets associated with the node. */
+		drop_node_replication_sets(node->id);
 
 		/* Drop the node itself. */
 		drop_node(node->id);
diff --git a/pglogical_hooks.c b/pglogical_hooks.c
@@ -209,18 +209,35 @@ pglogical_row_filter_hook(struct PGLogicalRowFilterArgs *rowfilter_args)
 		 * We can use this to update our cached replication set info, without
 		 * having to deal with cache invalidation callbacks.
 		 */
-		HeapTuple		tup = &rowfilter_args->change->data.tp.newtuple->tuple;
+		HeapTuple			tup;
 		PGLogicalRepSet	   *replicated_set;
 		ListCell		   *plc;
 
+		if (rowfilter_args->change_type == REORDER_BUFFER_CHANGE_UPDATE)
+			 tup = &rowfilter_args->change->data.tp.newtuple->tuple;
+		else if (rowfilter_args->change_type == REORDER_BUFFER_CHANGE_DELETE)
+			 tup = &rowfilter_args->change->data.tp.oldtuple->tuple;
+		else
+			return false;
+
 		replicated_set = replication_set_from_tuple(tup);
 		foreach (plc, private->replication_sets)
 		{
 			PGLogicalRepSet	   *rs = lfirst(plc);
 
-			/* Our cached replication set was updated, update local cache. */
+			/* Check if the changed repset is used by us. */
 			if (rs->id == replicated_set->id)
 			{
+				/*
+				 * In case this was delete, somebody deleted one of our
+				 * rep sets, bail here and let reconnect logic handle any
+				 * potential issues.
+				 */
+				if (rowfilter_args->change_type == REORDER_BUFFER_CHANGE_DELETE)
+					elog(ERROR, "replication set \"%s\" used by this connection was deleted, existing",
+						 rs->name);
+
+				/* This was update of our repset, update the cache. */
 				rs->replicate_insert = replicated_set->replicate_insert;
 				rs->replicate_update = replicated_set->replicate_update;
 				rs->replicate_delete = replicated_set->replicate_delete;
diff --git a/pglogical_node.c b/pglogical_node.c
@@ -332,7 +332,7 @@ drop_local_node(void)
 	HeapTuple		tuple;
 
 	rv = makeRangeVar(EXTENSION_NAME, CATALOG_LOCAL_NODE, -1);
-	rel = heap_openrv(rv, RowExclusiveLock);
+	rel = heap_openrv(rv, AccessExclusiveLock);
 
 	/* Find the local node tuple. */
 	scan = systable_beginscan(rel, 0, true, NULL, 0, NULL);
@@ -349,7 +349,7 @@ drop_local_node(void)
 
 	/* Cleanup. */
 	systable_endscan(scan);
-	heap_close(rel, RowExclusiveLock);
+	heap_close(rel, NoLock);
 
 	CommandCounterIncrement();
 }
diff --git a/pglogical_repset.c b/pglogical_repset.c
@@ -792,6 +792,35 @@ drop_replication_set(Oid setid)
 	heap_close(rel, RowExclusiveLock);
 }
 
+void
+drop_node_replication_sets(Oid nodeid)
+{
+	RangeVar	   *rv;
+	Relation		rel;
+	SysScanDesc		scan;
+	HeapTuple		tuple;
+	ScanKeyData		key[1];
+
+	Assert(IsTransactionState());
+
+	rv = makeRangeVar(EXTENSION_NAME, CATALOG_REPSET, -1);
+	rel = heap_openrv(rv, RowExclusiveLock);
+
+	ScanKeyInit(&key[0],
+				Anum_repset_nodeid,
+				BTEqualStrategyNumber, F_OIDEQ,
+				ObjectIdGetDatum(nodeid));
+
+	scan = systable_beginscan(rel, 0, true, NULL, 1, key);
+
+	/* Remove matching tuples. */
+	while (HeapTupleIsValid(tuple = systable_getnext(scan)))
+		simple_heap_delete(rel, &tuple->t_self);
+
+	systable_endscan(scan);
+	heap_close(rel, RowExclusiveLock);
+}
+
 /*
  * Insert new replication set / relation mapping.
  *
diff --git a/pglogical_repset.h b/pglogical_repset.h
@@ -67,6 +67,7 @@ extern bool relation_is_replicated(Relation rel, Oid nodeid,
 extern void create_replication_set(PGLogicalRepSet *repset);
 extern void alter_replication_set(PGLogicalRepSet *repset);
 extern void drop_replication_set(Oid setid);
+extern void drop_node_replication_sets(Oid nodeid);
 
 extern void replication_set_add_table(Oid setid, Oid reloid);
 extern void replication_set_remove_table(Oid setid, Oid reloid,
