[Tarantool-patches] [PATCH 4/8] replication: make sync transactions wait quorum

[Vladislav Shpilevoy]

I added a new commmit with a small fix on top of
this one:

====================
    limbo: remove entry from limbo before txn completion
    
    [TO BE SQUASHED INTO THE PREVIOUS COMMIT]
    
    - txn_limbo_wait_complete() didn't remove the limbo entry from
      the limbo when the transaction appears to be completed right
      away. As a result, the limbo entry stayed in the limbo, and
      after the transaction termination turned into garbage, because
      was allocated on its region;
    
    - txn_limbo_ack() had the same problem - completed transaction
      should be removed from limbo.

diff --git a/src/box/txn_limbo.c b/src/box/txn_limbo.c
index 3b9aa6d40..f38639676 100644
--- a/src/box/txn_limbo.c
+++ b/src/box/txn_limbo.c
@@ -123,8 +123,10 @@ txn_limbo_wait_complete(struct txn_limbo *limbo, struct txn_limbo_entry *entry)
 	assert(entry->lsn > 0);
 	assert(!txn_has_flag(txn, TXN_IS_DONE));
 	assert(txn_has_flag(txn, TXN_WAIT_ACK));
-	if (txn_limbo_check_complete(limbo, entry))
+	if (txn_limbo_check_complete(limbo, entry)) {
+		txn_limbo_remove(limbo, entry);
 		return;
+	}
 	bool cancellable = fiber_set_cancellable(false);
 	while (!txn_limbo_entry_is_complete(entry))
 		fiber_yield();
@@ -144,8 +146,8 @@ txn_limbo_ack(struct txn_limbo *limbo, uint32_t replica_id, int64_t lsn)
 	assert(limbo->instance_id != REPLICA_ID_NIL);
 	int64_t prev_lsn = vclock_get(&limbo->vclock, replica_id);
 	vclock_follow(&limbo->vclock, replica_id, lsn);
-	struct txn_limbo_entry *e;
-	rlist_foreach_entry(e, &limbo->queue, in_queue) {
+	struct txn_limbo_entry *e, *tmp;
+	rlist_foreach_entry_safe(e, &limbo->queue, in_queue, tmp) {
 		if (e->lsn <= prev_lsn)
 			continue;
 		if (e->lsn > lsn)
@@ -158,6 +160,7 @@ txn_limbo_ack(struct txn_limbo *limbo, uint32_t replica_id, int64_t lsn)
 			// appliers will be supposed to wait for
 			// 'confirm' message.
 			e->is_commit = true;
+			rlist_del_entry(e, in_queue);
 			fiber_wakeup(e->txn->fiber);
 		}
 		assert(e->ack_count <= VCLOCK_MAX);