Implement quorum-based synchronous replication

[sergos]

The quorum-based synchronous replication has to address:

• protocol backward compatibility to enable cluster upgrade w/o downtime
• consistency of data on replica and leader
• switch from leader to replica without data loss
• up to date replicas to run read-only requests
• ability to switch async replicas into sync ones
• guarantee of rollback on leader and sync replicas
• simplicity of cluster orchestration

The RFC is available here

Major subtasks:

• Introduce a flag in WAL op entry to enable tx required quorum #4843 Introduce a flag in WAL op entry to enable tx required quorum
• Develop quorum collection mechanism #4844 Develop quorum collection mechanism
• Rework the request answering to wait until quorum achieved #4845 Rework the request answering to wait until quorum achieved
• Implement IPROTO message to report quorum error from replica #4846 Implement IPROTO message to report quorum error from replica
• Implement WAL op entry for quorum confirmation #4847 Implement WAL op entry for quorum
• Implement WAL op entry for rollback in case of failed quorum #4848 Implement WAL op entry for rollback in case of failed quorum
• Replica appointed to SR Leader has to issue a rollback #4849 Replica appointed to SR Leader has to issue a rollback
• Delay memtx snapshot until quorum for the op is achieved #4850 Delay memtx snapshot until quorum for the op is achieved
• Recovery needs rollback in case WAL had a tx without appropriate quorum #4851 Recovery needs rollback in case WAL had a tx without appropriate quorum
• track in _cluster table hostname, vclock and time of last login/logout of replica #3363 Track in _cluster table hostname, vclock and time of last login/logout of replica
• Provide interfaces to support cluster management CLI #4852 Provide interfaces to support cluster CLI
• Quorum-based replication configuration options #5073 Quorum-based replication configuration options

[Gerold103]

I started implementation on this branch: gerold103/gh-4842-sync-replication. Since the tickets are depending on each other, we probably should push our intermediate results more often and inter-merge our branches sometimes.

[Gerold103]

I pushed 'some kind of sync replication' on the branch.
https://github.com/tarantool/tarantool/tree/ad983366d97f6a9f50c0edfb48cb66918c42deaf
Basically it is just wait_lsn, but right inside txn_commit() in C. However there is a plan how to develop it next. Probably other issues may depend on that, so I decided to describe it here.

1. I tried to solve a problem where to store transactions, which wait for a quorum, so as relays could reach to them and confirm. For that I introduced a structure txn_limbo. It is a list of transactions originated from the master, and a vclock. Every component of the vclock is how replicas see master's LSN. So it is 32 versions of master LSN, how it is seen on all replicas.

2. I added replication_sync_quorum option. It is a global variable which means how many instances should apply the transaction before its commit succeeds.

3. Every relay thread on the master in tx_status_update() tells the last vclock of the relay to the txn_limb, and increments ack_count of newly acked transactions.

4. When transaction's fiber wakes up and sees it has reached the quorum, the commit succeeds.

The key thing here is txn_limbo. It is supposed to be the key data structure in all these confirmation/rollback things. It is a channel between replication module and txn module.

[Gerold103]

I pushed a more or less finished version of the things I described above on the same branch: gerold103/gh-4842-sync-replication.

[kostja]

* Vladislav Shpilevoy <notifications@github.com> [20/04/30 08:17]:

I pushed 'some kind of sync replication' on the branch. https://github.com/tarantool/tarantool/tree/ad983366d97f6a9f50c0edfb48cb66918c42deaf Basically it is just wait_lsn, but right inside `txn_commit()` in C. However there is a plan how to develop it next. Probably other issues may depend on that, so I decided to describe it here. 1) I tried to solve a problem where to store transactions, which wait for a quorum, so as relays could reach to them and confirm. For that I introduced a structure `txn_limb`. It is a list of transactions originated from the master, and a vclock. Every component of the vclock is how replicas see master's LSN. So it is 32 versions of master LSN, how it is seen on all replicas. 2) I added replication_sync_quorum option. It is a global variable which means how many instances should apply the transaction before its commit succeeds.

There is no global option in the spec, all options are per space.

…

3) Every relay thread on the master in `tx_status_update()` tells the last vclock of the relay to the txn_limb, and increments ack_count of newly acked transactions. 4) When transaction's fiber wakes up and sees it has reached the quorum, the commit succeeds. The key thing here is txn_limb. It is supposed to be the key data structure in all these confirmation/rollback things. It is a channel between replication module and txn module.

-- Konstantin Osipov, Moscow, Russia

[Gerold103]

So if we add space option quorum, then all transactions will have their own quorum value? And the needed quorum is the MAX among quorums of affected sync spaces, right?

[kostja]

I mentioned earlier on the thread that I don't think we should let spaces with different quorum requirements mix in the same transaction and provided an example how it leads either to availability or consistency issues. Please read my reply to Sergey's spec.

The more I think of it the more I believe that we need to explicitily specify the replication group the space is part of, not even replication factor. and the replication group has to be explicitly defined from uuids of its participants. Otherwise we have 6 replicas, rf=3, two transactions get quorum=2 but from 2 different replicas -> consistency violation.

[kostja]

In other words, I suggest to introduce new ddl:
box.schema.group.create('groupname')
box.schema.group.add('gropuname', server_uuid)
box.schema.group.drop()
And then in space properties:
box.schema.space.create('spacename', {group = groupname})
It's a good idea to reserve the first thousand groups for the system.

[kostja]

System spaces should be part of a system group = all, which includes all nodes in _cluster