[PATCH] replication: unregister replica with gc if deleted from cluster

[Vladimir Davydov <vdavydov.dev@gmail.com>]

When a replica is removed from the cluster table, the corresponding
replica struct isn't destroyed unless both the relay and the applier
attached to it are stopped, see replica_clear_id(). Since replica struct
is a holder of the garbage collection state, this means that in case an
evicted replica has an applier or a relay that fails to exit for some
reason, garbage collection will hang.

A relay thread stops as soon as the replica it was started for receives
a row that tries to delete it from the cluster table (because this isn't
allowed by the cluster space trigger, see on_replace_dd_cluster()).
If a replica isn't running, the corresponding relay can't run as well,
because writing to a closed socket isn't allowed. That said, a relay
can't block garbage collection.

An applier, however, is deleted only when replication is reconfigured.
So if a replica that was evicted from the cluster was configured as a
master, its replica struct will hang around blocking garbage collection
for as long as the replica remains in box.cfg.replication. This is what
happens in #3546.

Fix this issue by forcefully unregistering a replica with the garbage
collector when it is deleted from the cluster table. This is OK as it
won't be able to resubscribe and so we don't need to keep WALs for it
any longer. Note, the relay thread may still be running when a replica
is deleted from the cluster table, in which case we can't unregister it
with the garbage collector right away, because the relay may need to
access the garbage collection state. In such a case, leave the job to
replica_clear_relay, which is called as soon as the relay thread exits.

Closes #3546
---
https://github.com/tarantool/tarantool/issues/3546
https://github.com/tarantool/tarantool/commits/dv/gh-3546-fix-replica-gc

 src/box/replication.cc       | 22 ++++++++++++
 test/replication/gc.result   | 83 ++++++++++++++++++++++++++++++++++++++++++++
 test/replication/gc.test.lua | 36 +++++++++++++++++++
 3 files changed, 141 insertions(+)

diff --git a/src/box/replication.cc b/src/box/replication.cc
index c752006e..528fe445 100644
--- a/src/box/replication.cc
+++ b/src/box/replication.cc
@@ -197,6 +197,18 @@ replica_clear_id(struct replica *replica)
 	 */
 	replicaset.replica_by_id[replica->id] = NULL;
 	replica->id = REPLICA_ID_NIL;
+	/*
+	 * The replica will never resubscribe so we don't need to keep
+	 * WALs for it anymore. Unregister it with the garbage collector
+	 * if the relay thread is stopped. In case the relay thread is
+	 * still running, it may need to access replica->gc so leave the
+	 * job to replica_clear_relay, which will be called as soon as
+	 * the relay thread exits.
+	 */
+	if (replica->gc != NULL && replica->relay == NULL) {
+		gc_consumer_unregister(replica->gc);
+		replica->gc = NULL;
+	}
 	if (replica_is_orphan(replica)) {
 		replica_hash_remove(&replicaset.hash, replica);
 		replica_delete(replica);
@@ -702,6 +714,16 @@ replica_clear_relay(struct replica *replica)
 {
 	assert(replica->relay != NULL);
 	replica->relay = NULL;
+	/*
+	 * If the replica was evicted from the cluster, we don't
+	 * need to keep WALs for it anymore. Unregister it with
+	 * the garbage collector then. See also replica_clear_id.
+	 */
+	assert(replica->gc != NULL);
+	if (replica->id == REPLICA_ID_NIL) {
+		gc_consumer_unregister(replica->gc);
+		replica->gc = NULL;
+	}
 	if (replica_is_orphan(replica)) {
 		replica_hash_remove(&replicaset.hash, replica);
 		replica_delete(replica);
diff --git a/test/replication/gc.result b/test/replication/gc.result
index fec9d819..e2b8f47f 100644
--- a/test/replication/gc.result
+++ b/test/replication/gc.result
@@ -327,6 +327,89 @@ replica_set.wait_all(test_run)
 replica_set.drop_all(test_run)
 ---
 ...
+--
+-- Check that once a replica is removed from the cluster table,
+-- all xlogs kept for it are removed even if it is configured as
+-- a replication master (gh-3546).
+--
+fio = require('fio')
+---
+...
+fiber = require('fiber')
+---
+...
+-- Start a replica and set it up as a master for this instance.
+test_run:cmd("start server replica")
+---
+- true
+...
+replica_port = test_run:eval('replica', 'return box.cfg.listen')[1]
+---
+...
+replica_port ~= nil
+---
+- true
+...
+box.cfg{replication = replica_port}
+---
+...
+-- Stop the replica and write a few WALs.
+test_run:cmd("stop server replica")
+---
+- true
+...
+test_run:cmd("cleanup server replica")
+---
+- true
+...
+_ = s:auto_increment{}
+---
+...
+box.snapshot()
+---
+- ok
+...
+_ = s:auto_increment{}
+---
+...
+box.snapshot()
+---
+- ok
+...
+_ = s:auto_increment{}
+---
+...
+box.snapshot()
+---
+- ok
+...
+#fio.glob('./master/*.xlog') == 4 or fio.listdir('./master')
+---
+- true
+...
+-- Delete the replica from the cluster table and check that
+-- all xlog files are removed.
+test_run:cleanup_cluster()
+---
+...
+box.snapshot()
+---
+- ok
+...
+t = fiber.time()
+---
+...
+while #fio.glob('./master/*xlog') > 0 and fiber.time() - t < 10 do fiber.sleep(0.01) end
+---
+...
+#fio.glob('./master/*.xlog') == 0 or fio.listdir('./master')
+---
+- true
+...
+-- Restore the config.
+box.cfg{replication = {}}
+---
+...
 -- Cleanup.
 s:drop()
 ---
diff --git a/test/replication/gc.test.lua b/test/replication/gc.test.lua
index 8d5a36dd..2f1b2e7c 100644
--- a/test/replication/gc.test.lua
+++ b/test/replication/gc.test.lua
@@ -157,6 +157,42 @@ replica_set.start_all(test_run)
 replica_set.wait_all(test_run)
 replica_set.drop_all(test_run)
 
+--
+-- Check that once a replica is removed from the cluster table,
+-- all xlogs kept for it are removed even if it is configured as
+-- a replication master (gh-3546).
+--
+fio = require('fio')
+fiber = require('fiber')
+
+-- Start a replica and set it up as a master for this instance.
+test_run:cmd("start server replica")
+replica_port = test_run:eval('replica', 'return box.cfg.listen')[1]
+replica_port ~= nil
+box.cfg{replication = replica_port}
+
+-- Stop the replica and write a few WALs.
+test_run:cmd("stop server replica")
+test_run:cmd("cleanup server replica")
+_ = s:auto_increment{}
+box.snapshot()
+_ = s:auto_increment{}
+box.snapshot()
+_ = s:auto_increment{}
+box.snapshot()
+#fio.glob('./master/*.xlog') == 4 or fio.listdir('./master')
+
+-- Delete the replica from the cluster table and check that
+-- all xlog files are removed.
+test_run:cleanup_cluster()
+box.snapshot()
+t = fiber.time()
+while #fio.glob('./master/*xlog') > 0 and fiber.time() - t < 10 do fiber.sleep(0.01) end
+#fio.glob('./master/*.xlog') == 0 or fio.listdir('./master')
+
+-- Restore the config.
+box.cfg{replication = {}}
+
 -- Cleanup.
 s:drop()
 box.error.injection.set("ERRINJ_RELAY_REPORT_INTERVAL", 0)
-- 
2.11.0