[Tarantool-patches] [PATCH 4/5] [tosquash] replication: rework how local transactions wait sync

[Vladislav Shpilevoy]

>> diff --git a/src/box/txn_limbo.c b/src/box/txn_limbo.c
>> index 387cfd337..44a0c7273 100644
>> --- a/src/box/txn_limbo.c
>> +++ b/src/box/txn_limbo.c
>> @@ -316,7 +320,7 @@ txn_limbo_read_rollback(struct txn_limbo *limbo, int64_t lsn)
>>       assert(limbo->instance_id != REPLICA_ID_NIL);
>>       struct txn_limbo_entry *e, *tmp;
>>       rlist_foreach_entry_safe_reverse(e, &limbo->queue, in_queue, tmp) {
>> -        if (e->lsn <= lsn)
>> +        if (e->lsn <= lsn && txn_has_flag(e->txn, TXN_WAIT_ACK))
>>               break;
> 
> Are you rolling back the async transactions that are before the last sync
> transaction to be rolled back? Why?
> Shouldn't this condition stay the same?

Да, похоже на баг. Оставить без изменений не выйдет - у асинхронных
транзакций в лимбе лсн -1. Так что на первой асинхронной транзакции
роллбек остановится, даже если надо идти дальше. Я сделал как в
confirm в итоге. Сначала находим докуда откат. Потом делаем откат.

Тест пока не придумал. Мастер откатывает сейчас либо все, либо ничего.
Потому как откат только по таймауту, а значит всегда с самой первой.

====================
--- a/src/box/txn_limbo.c
+++ b/src/box/txn_limbo.c
@@ -319,9 +319,17 @@ txn_limbo_read_rollback(struct txn_limbo *limbo, int64_t lsn)
 {
 	assert(limbo->instance_id != REPLICA_ID_NIL);
 	struct txn_limbo_entry *e, *tmp;
-	rlist_foreach_entry_safe_reverse(e, &limbo->queue, in_queue, tmp) {
-		if (e->lsn <= lsn && txn_has_flag(e->txn, TXN_WAIT_ACK))
+	struct txn_limbo_entry *last_rollback = NULL;
+	rlist_foreach_entry_reverse(e, &limbo->queue, in_queue) {
+		if (!txn_has_flag(e->txn, TXN_WAIT_ACK))
+			continue;
+		if (e->lsn <= lsn)
 			break;
+		last_rollback = e;
+	}
+	if (last_rollback == NULL)
+		return;
+	rlist_foreach_entry_safe_reverse(e, &limbo->queue, in_queue, tmp) {
 		e->is_rollback = true;
 		txn_limbo_pop(limbo, e);
 		txn_clear_flag(e->txn, TXN_WAIT_SYNC);
@@ -342,6 +350,8 @@ txn_limbo_read_rollback(struct txn_limbo *limbo, int64_t lsn)
 			txn_complete(e->txn);
 			e->txn->fiber = fiber;
 		}
+		if (e == last_rollback)
+			break;
 	}
 }