Doubled buckets

[Serpentian]

• Main reviewer: @Gerold103
• Second reviewer: @mrForza
• Team Lead: @sergepetrenko
• CTO: @sergos

Associated tickets

• Doubled buckets after master switch #576
• Stray TCP message with big delay may duplicate a bucket #214

Changelog

1. First iteration fixes

• Fixed the solution summary, which didn't comply with the description;
• Nullable opts field instead of the generation in the _bucket;
• The fiber, which synchronizes the _bucket after master switch
  now doesn't require to be cancelled synchronously;
• Renamed the M.rebalance_allowed to the M.bucket_are_in_sync;
• Take buckets without refs, the remainings are with random number of RW refs.

2. Second iteration fixes

• Added to considered alternatives bucket vclock and generation for discovery;
• Renamed generation to send_generation.

3. Third iteration fixes

• Renamed send_generation to generation back)

4. Fourth iteration fixes

• Renamed PREPARED to READONLY to comply with another RFC.
• Added Allow to skip disabled nodes when syncing part to alternatives.

Disclaimer for users

Currently during rebalance it's unsafe to make vshard requests, if a master change is possible. Read-only requests may return invalid data, and write requests can be lost even after a user has received a write confirmation.

Thus, users of vshard must disable failover prior to moving any bucket before this bunch of problems is fixed in vshard.

P.S. Note, that vshard is supposed to be used via vshard.router.call API, not vshard.router.route*() + rs:call*, which is unsafe.

Here's the list of tickets, which must be closed for safe requests during rebalancing when using clean vshard:

• RW refs break replication when master changes #573
• Doubled buckets after master switch #576
• Stray TCP message with big delay may duplicate a bucket #214

This RFC focuses on the last two problems.

---

However, not only vshard is unsafe, but also high-level products (such as crud (tarantool/crud#448) or tdg (https://github.com/tarantool/tdg2/issues/2175)) use vshard's API unsafely, which leads to even higher chance of broken requests.

Projects which use crud or tdg (not all requests are unsafe here, see the issue in the tdg) cannot rely on RW or RO requests during rebalancing at all, the load on the cluster should be stopped during rebalancing. Requests can read incorrect data, write requests can be lost. After rebalancing end the bucket cache on all of the routers must be emptied.

Here's the list of tickets, which must be closed for safe requests during rebalancing in high-level projects:

• Implement router.map_callro #333 (required for tdg)
• Full map_callrw with split args #559 (required for tdg)
• Allow to split data in map_callrw #539 (required for crud and tdg)

These vshard issues are already designed and wait to be implemented. The following issues should be resolved by the project themselves:

• Calls bypass vshard's checks  crud#448
• https://github.com/tarantool/tdg2/issues/2175

Disclaimer for reviewers and developers

I'm using TLA+ specification of the storage to proof, that these problems with doubled buckets are real and the changes, I'm proposing in this RFC, solves the problem. You can find the TLA+ spec and the solutions implemented here.

However, note, that sophisticated cases of doubled buckets require big depth of states (e.g. stray TCP issue's depth is 15-19), which in case of general specification (e.g. proofs/tla/test/storage/DoubledBucketsTest.tla) requires hours of execution on my hardware at constant 100% on all cores to reach that depth. And even this spec is very limited (no more that 1 reorder or drop of messages in the network, the maximum number of bucket sends is 2, connection from one instance to another doesn't have more than 3 unprocessed messages, refs are disabled completely, 1 instance in each of the 3 replicasets).

We'll have to test general specifications on stronger hardware, but even then we cannot be sure, that doubled bucket problem is completely solved, since it may be reproduced outside of constraints, we'll set, or it may be missing from TLA spec, since it's just an abstraction of the real code.

But anyway, some testing of the algorithm is anyway better, than nothing. We should move towards developing of the TLA+ spec, since any error in the algorithm of bucket rebalancing causes data loss and inconsistencies.

1. Problem overview

The doubled bucket problem is when there're two or more ACTIVE buckets with the same bucket_id on the masters of different shards. This is crucial problem, since routers doesn't know, which of these buckets is the real one and will send requests to the random one, this leads to incorrect reads and lost writes.

During incident it was discovered, that it's very easy to achieve doubled buckets in the cluster, if failover is enabled and changes masters (#576). Here's what happened:

1. Rebalancing is in progress. Due to high load on storages during rebalancing, they don't respond to cartridge failover's pings, failover constantly changes masters in the cluster.
2. Instance 1 is the master of rs1, it sends bucket N to another replicaset, makes it SENT, on another shard the bucket is ACTIVE. Another Instance 2 in the replicaset doesn't have that change yet, bucket is ACTIVE there. Master switch happens.
3. Instance 2 is the new master, it has replication lag from the old master. It starts to process RW on top of bucket N, then the update of the bucket N state comes from the old master.
4. The replication becomes stopped due to RW refs break replication when master changes #573. SENT bucket cannot have RW refs. The bucket remains in the ACTIVE state.
5. From now on there're two ACTIVE buckets with number N in the cluster. All of the updates for the bucket in the rs1 will be lost as soon as replication is restored and the bucket becomes deleted.

If Instance 2 doesn't have RW refs in the 3rd step, then the data is lost even without breaking the replication. But this will happen for a small amount of time.

This problem has really small depth (6-7) in the TLA spec (see MasterDoubledTest.tla or DoubledBucketsSmallTest.tla) and consequently can be reproduced very easily. Very dangerous bug. It probably happens in a cluster pretty often, but the consistency of requests just wasn't checked.

---

There's other way doubled buckets can happen (#214), which is not related to master switch:

1. rs1 sends a bucket to rs2. The bucket becomes SENDING on rs1, but the message, with is supposed to add RECEIVING to rs2 is lost.
2. rs1 timeouts on sending bucket, recreates the connection, checks bucket on the rs2, makes the bucket ACTIVE.
3. rs1 sends the same bucket to rs3.
4. Lost message to rs2 arrives, makes the bucket on rs2 RECEIVING.
5. rs2's recovery goes to rs1, doesn't find the bucket there (or finds it there as SENT or GARBAGE) and makes the bucket ACTIVE.
6. The bucket is ACTIVE on rs2 and rs3.

The important words here is "recreate the connection", since TCP guarantees the order of messages in the scope of one connection, but after recreation the new request (checking for recovery) may come earlier, than the older one (bucket_recv call).

The depth of the problem is 15-19 in the spec (see StrayTCPDoubledTest.tla and DoubledBucketsTest.tla). It's very difficult to encounter such situation in real life, but it's still possible and we should fix it too.

2. Solution

2.0 Summary

From now on rebalancing of replicaset won't work, if there're unavailable nodes in the replicaset. Even if a replica is manually disabled, we cannot allow rebalancing in the replicaset, it's way too unsafe, since user may enable the instance back at any time. We expect user to have fully working cluster. Rebalancing happens very rarely and we can require that from user, IMHO.

rebalance_apply_routes process works as follows (please, note, that another RFC for #573 also adds READONLY state of the bucket):

1. Prefer buckets without refs, if there're not enough buckets without refs, take the remainings from the first buckets sorted from the primary index. Set rw_lock on them, no new RW refs locally. Solution for Rebalancer should firstly try to pick buckets which already have no refs #351.
2. Take this bunch of buckets, wait for 0 RW refs on them locally, make the buckets SENDING and increment their generation version (persisted), wait for vclock of all connected nodes to reach the master's vclock. Solution for Rebalancer should firstly try to pick buckets which already have no refs #351.
3. The buckets can now be sent alongside with generation version to bucket_recv().

After a node becomes a master:

1. A new master synces with the replicaset to have latest state of the _bucket space. Needed for Stray TCP message with big delay may duplicate a bucket #214.
2. The rebalancer and recovery service can start their work only after sync. The node also doesn't respond to rebalancer_apply_routes or recovery_bucket_stat before these checks are completed.

Solution for #214. Recovery uses bucket generation from now on. So, firstly, the recovery goes to the sender:

• If there's a bucket with any state and greater generation, local one is GARBAGE;
• If bucket generation is equal to the local one, we use the old logic;
• If the bucket is missing from remote node, then fullscan all masters of the cluster.

The format of the _bucket space looks as follows, opts is the new field, it's map and it's nullable for backward compatibility:

{bucket_id, state, destination, opts}

2.1 Doubled buckets after master switch (#576)

Solution summary: after making the bunch of buckets SENDING we synchronize with our replicaset and do not send, until all of the nodes confirm the update.

In order to reproduce that issue, the new master must have replication lag from the old master and do not get the update of the sent bucket statuses. This way we get the situation, when there's two ACTIVE buckets until the new master synchronizes with the old one.

The simplest solution here is to make the current master synchronize with all instances in the replicaset as soon as it makes the bucket SENDING. The master cannot make the bucket SENT until all of the nodes confirmed, that they have the bucket in SENDING.

But in order to to fortify the checks I propose not to start sending bucket to another replicaset at all until all the nodes confirm the SENDING bucket state. This also decreases the load on network, if some nodes in the replicaset are unavailable: syncing with replicaset is way cheaper, than sending a data from a bucket and should be done first.

We don't restore the state of bucket in the bucket_send(), if we didn't manage to sync with replicasets during timeout. Recovery service will do that for us in the future, just don't forget to wakeup it in case of error.

Rejected alternatives for #576

Alternative 1. Synchro _bucket space

In case of synchro _bucket and election failover it's enough to make sure that the majority of them got the change of the SENDING bucket. The node, which didn't get that change won't be able to become a master in the future, so the new master will always have a consistent state of the _bucket.

The only problem here is that, Tarantool is not ready right now to become synchronous only DB and we cannot make the _bucket sync, while there're async spaces. Otherwise, user will start to encounter SplitBrain replication errors.

Alternative 2. Allow to skip disabled nodes when syncing

It was proposed by @sergepetrenko to allow rebalancing, when there're disabled nodes in the replicaset, and skip them, when we need to sync the SENDING or READONLY state. However, this cannot be done, because it's way too unsafe. Even allowing excluded from the configuration nodes to be skipped is unsafe (described in the #623, leave as is for now).

Our own above-standing projects may disable/enable instances, when they want to (now cartridge disables instance, when the roles wasn't applied (applying is not done yet or error during role applying) and the storage may remain in this disabled state as long as it's needed, at any time the node may become enabled again).

Now we consider the situation:

1. A role on one storage cannot be applied, a user doesn't care about that and start rebalancing. The bucket is sent, replication is broken (or lags) on the node, which cannot apply a role.
2. Then somehow the applying role ends and node becomes enabled, failover changes master to that node
3. We have doubled buckets again.

Why expelling is better? Because expelling deletes the node from the _cluster space, which makes the node to rejoin (that guarantees us, that the node will have correct bucket state after rejoin is done) 1. We have the same logic in Tarantool 3, but it's not enabled by default 2.

2.2 Doubled buckets after stray TCP (#214)

Solution summary: recovery uses bucket generations and fullscans the masters of a cluster. Recovery and rebalancer are not started until the new master synchronzies with other instances in the replicaset.

We have a problem with clean stray TCP, described in the "Problem overview" part. The simplest way, which may seem to solve it, is to make the recovery service fullscan the cluster for buckets. But unfortunately, it doesn't work (see alternatives), so we must move towards bucket versioning here.

From now on _bucket space will look like that:

{bucket_id, state, destination, opts}

The new field is added: opts, which has the map type. This is done in order to simplify adding new fields in the future. The opts is nullable for backward compatibility. There may be only one option for now: opts.generation(below I refer to it as generation instead of opts.generation. We want generation to be persistent, so that after restart recovery could still work.

When bucket is sent, its generation is incremented (we make the bucket SENDING and increment generation in one transaction) and is sent alongside the data of the bucket to the bucket_recv, the receiver side persists that generation in the _bucket.

Recovery uses that generation in order to distinguish, which bucket is more recent, if it cannot find a bucket on the sender node. So, firstly, the node goes to the sender, if there's a bucket with any state and greater generation, local one is GARBAGE, we don't care about the status here. If bucket generation is equal to the local one, we use the old logic, if the bucket is missing from remote node, then fullscan all masters of the cluster. When all of the nodes replied, if there exists higher generation, the local is GARBAGE, ACTIVE otherwise.

Why do we scan only for masters, you may ask? The following can happen:

1. The rs1 sends the bucket to rs2, message is lost, bucket is recovered. Generation is 2 on rs1 and rs2.
2. The bucket is sent from the rs1 to the rs3. Generation is 3 on rs1 and rs3.
3. Master switches in the rs3, new node doesn't have the bucket at all.
4. rs2 gets the message, makes the bucket RECEIVING with generation 2, scans masters only, doesn't find the bucket at all and recovers it to ACTIVE.
5. New master of rs3 gets the change from the old master, two ACTIVE buckets.

In order to fight with this we make the new master sync the _bucket space. And before this sync happens, the node doesn't start rebalancer or recovery services, it also doesn't respond to the requests for these services from another nodes (e.g. rebalance_apply_routes or recovery_bucket_stat.

Here is either fullscan of ALL replicas in the cluster, or new master always synces with its replicaset. And I like the latter more, since the number of replicas may be 2 - 5 times the number of masters (it really can, I've never seen more that 3, but I've heard of them).

This becomes a new background service on_master_enable_service, which will be started, when instance becomes RW in auto mode, and during reconfiguration in case of "manual" master mode. It's not a loop (which is common among other services), but a function, which is executed in background once the node becomes master and then exits.

It's a service, since we should not block the reconfiguration for that. It's crucial that before starting the new service the old one already cannot interfere in our flags, so we use testcancel before any change of the M.bucket_are_in_sync (see below).

Right before starting the new service old one is cancelled, the flag M.bucket_are_in_sync will be set to false and the rebalancer and recovery services won't be able to start until it becomes true, the node also doesn't respond to requests from these services.

This service will wait for the current node to reach vclock of all instances in the replicaset in order to get the latest updates from the _bucket space. As soon as these conditions are satisified, the M.bucket_are_in_sync is set to true, rebalancing is allowed, the service dies, now recovery and rebalancer can do their stuff.

Rejected alternatives for #214

Alternative 1. Recovery fullscan and that's it

The simplest solution here may seem to make recovery service fullscan the cluster, if it finds the bucket in SENT, GARBAGE state or missing on the sender. But, unfortunately, it won't work:

1. rs1 sends the bucket to rs2, message is lost, rs1 recovers the bucket from SENDING to ACTIVE.
2. rs2 sends the bucket to rs3, makes the bucket SENT, the message, which is supposed to make the bucket ACTIVE on rs3 is lost (final = true in code).
3. Now recovery on rs2 and rs3 are supposed to decide, which of their RECEIVING buckets is correct, but they won't be able to do that without knowing, which of them came later.

The recovery should make the bucket on rs3 ACTIVE, since it may have data, which was written by the rs1, while the bucket was ACTIVE between bucket sends. The bucket on rs2 consequently should become GARBAGE.

Fullscan won't work, to my deepest regret. We need to apply more sophisitcated solution here.

Alternative 2. Never delete the bucket info from space, even for transferred buckets.

This is the alternative solution for fullscan of the cluster, when the bucket is not found.

I know clients with tens (or even hunderds) of millions buckets, and that solution requires storing the versions of all buckets in the cluster (at least when the cluster worked long enough), and I'm afraid that it may become costly to store.

Rebalancing happens once a year or smth like that, it happens, when load on the cluster is disabled or minimal. It's better to increase the load on the network once a year than make all users to pay with RAM for rebalancing.

Alternative 3. generation field instead of the opts in the _bucket

The initial proposition was to use the following format:

{bucket_id, state, destination, generation}

What if in the future we'll want to persist some new info about the buckets, then we'll have to add new fields to the bucket again and it'll become similar to the _func space:

- [{'name': 'id', 'type': 'unsigned'}, {'name': 'owner', 'type': 'unsigned'}, {'name': 'name',
    'type': 'string'}, {'name': 'setuid', 'type': 'unsigned'}, {'name': 'language',
    'type': 'string'}, {'name': 'body', 'type': 'string'}, {'name': 'routine_type',
    'type': 'string'}, {'name': 'param_list', 'type': 'array'}, {'name': 'returns',
    'type': 'string'}, {'name': 'aggregate', 'type': 'string'}, {'name': 'sql_data_access',
    'type': 'string'}, {'name': 'is_deterministic', 'type': 'boolean'}, {'name': 'is_sandboxed',
    'type': 'boolean'}, {'name': 'is_null_call', 'type': 'boolean'}, {'name': 'exports',
    'type': 'array'}, {'name': 'opts', 'type': 'map'}, {'name': 'comment', 'type': 'string'},
  {'name': 'created', 'type': 'string'}, {'name': 'last_altered', 'type': 'string'},
  {'name': 'trigger', 'type': 'array'}]

I propose not to add the generation field but instead introduce the new opts field, which will have the map type. It'd be great to place the destination there too, but I'm afraid we might break too many above standing code (in our products and also in the client's code).

We could build indexes other the map, when we'll need them in the future: link

Alternative 4. Try using the generation for making discovery cheaper

It'd be great, if we could use generation in the future in order to make the discovery cheaper: link. Unfortunately, the generation which is proposed in the RFC, won't help us to solve the issue. It's proposed to use prev_generation_of_this_bucket + 1, which won't allow us to make the discovery cheaper: rs1 sends bucket 1 to rs2, the generation on rs1 and rs2 becomes 1 (maximum over generations), then rs1 sends bucket 2 to rs2, the generation remains 1 on both storages, router won't be able to notice the change, even though it should.

So, assigning new generation as max_local_generation + 1 is required for that to work. But this will break the logic of doubled bucket generation:

1. rs1 sends bucket with generation 1 to rs2, message lost
2. rs2 restores the bucket, generation is 2, sends it to rs3, generation is 3
3. rs3 gets bucket, the last message is lost, generation is 3, state - RECEIVING.
4. rs2 gets bucket, it locally has generation 100, assigns to received bucet generation 101

The generation which is needed for making discovery process cheaper, must have absolutely another logic from the generation, which is used for deudplication of the buckets.

These two different generations can be merged into one (see bucket vclock below), but this is way too costly memory-wise, and we decided to add new generation to opts for making discovery cheaper, when it'll be needed.

Vclock for buckets

Drop generation, let's save vclock for every bucket instead! The vclock looks as follows:

{<replicaset_id>: <number of WHATEVER->SENDING transitions}

In the RFC for doubled buckets all checks are done only for destination field: e.g. if bucket is not found on the sender, then we fullscan the cluster and check only the destination (sender id) component of the vclock. It'll work the same: if found vclock[sender] > local vclock[sender], then GARBAGE, ACTIVE otherwise.

Now to the discovery process:

1. Storage

When storage starts, it initializes bucket vclock as {<r_id> = max over all vclock[r_id]}. It updates the maximums in that bucket vclock from the on_replace trigger on _bucket. There're N indexes, built over every replicaset from vclock on _bucket. When replicaset is deleted, index is dropped, when added - new index is created. All of the fields in vclocks are nullable, null = 0.

2. Router

When router starts, it has a table {rs_id = <vclock>}, all vclocks are initialized with 0s. It goes to every replicaset with this vclock, if the keys in vclock (replicaset_ids) from the router and storage doesn't match, storage returns error, incorrect configuration. Then storage goes over vclock from the router, for every replicaset it sends buckets with bsn > router's. The storage also returns the maximum vclock of sent buckets, so that router can iteratively update the map.

---

There're still problems with redownloading of the map on reconnect/storage restart, if the max bucket vclock is not persisted. But speaking of Can't download removal of sent and deleted buckets part, I don't think it's a problem: yeah, most of the storages will just delete the buckets, they will send nothing in response to discovery of the router, but one of replicasets will get a lot of buckets, which will be sent to router and it will update the route map, as needed.

[Gerold103]

Thanks, amazing work. I especially like how you leveraged TLA+ for this.

My comments below.

1. "Sort all buckets in ascending order by the number of RW refs on them, take the first N buckets, set rw_lock on them" - this won't be necessary. Number of refs tells not much about when a bucket will become ref-free. A single ref on a bucket might be held for seconds, 1000 refs on another bucket - for a millisecond. The other bucket would then be easier to wait on. In such cases it is easier to select them randomly, or select just a sequence of buckets sorted by ID from _bucket's primary index (this way we can also preserve some sort of "bucket ranges" which don't have a practical use now, but might be in the future. What might also be helpful is to immediately take the buckets having no refs. And then random ones having any number of refs.

2. "This requires changing of the _bucket space schema, but I don't see any problems with it" - this is fine. Just make the new field nullable. Then the old versions will boot ok without having to change anything.

3. "check out the alternatives, I want to discuss opts field" - yes. A map of options is always better for compatibility. Even the "destination" field should have been in options from the beginning, but we didn't think much about extensions back then. Also we have JSON indexes now, which makes root fields relatively pointless even performance-wise (at least in all imaginable _bucket extensions).

4. "For that we must synchronously (right during reconfiguration or from the instance_watch_f) cancel the fiber and wait for its end" - this shouldn't be necessary, if the service-fiber would do fiber.testcancel() before doing any changes to the instance's state.

5. "M.rebalance_allowed" - I suggest to consider a more generic name. It is not only about rebalancer itself, but also about manual bucket_send() calls and about consequences of rebalancing after it is already done, like GC and recovery. Sometime like M.are_buckets_in_sync or alike.

6. "Bucket generations will allow us to make the discovery cheaper in the future" - the ticket Make discovery almost free and continuous after all is discovered #238 talks about replicaset-local bucket generation. It is not a generation of one bucket, but a generation of _bucket space state on the whole. Assuming you are going to have bucket generations bucket-local (so the increments are done for each bucket individually, and don't bump the generations of other buckets), the discovery speed up seems unlikely to become easier. Or do I miss something?

[Gerold103]

All good. Just lets perhaps keep the old name. It was quite concise and clear. And send_generation will prevent us from bumping it for transitions like active -> readonly, or active -> pinned, which I think might be in the future handy for something.

[Serpentian]

active -> readonly, or active -> pinned

We can indeed do that, but it seems, that we can increase the generation only when it transitions to SENDING or from ACTIVE. Any other transitions will break the logic of recovery, this is why I wanted to narrow the name

P.S. renamed back

[Gerold103]

Hm. Wait wait. This looks strange. Why would it break when it gets incremented on ACTIVE -> PINNED for example?

[Serpentian]

Incrementing generation on the following transitions don't break the proposed logic of recovery:

ACTIVE <-> PINNED
ACTIVE <-> READONLY
READONLY -> SENDING (this one is the only, we use now)

Any other transitions require changing the design of the recovery. The bucket, which is sent later, must have higher generation for recovery to figure out, what to do.

For example, incrementing generation on NULL -> RECEIVING can be done, but it requires changing the RFC: old behavior, when on sender generation is current - 1, GARBAGE when greater than current - 1 and so on. Incrementing on RECEIVING -> GARBAGE cannot be done at all, since then we cannot figure out, which bucket was sent later.

[Gerold103]

Ah, but this is fine then. Other states aren't completely usable anyway. Like for example NULL -> RECEIVING increment wouldn't make sense, this we already discussed. Anything to GARBAGE or SENT is also pointless. Bucket in this state is basically unrecoverable trash now. The only left transactions are the ones you listed above, and bumping on them is ok.

[sergepetrenko]

@Gerold103 if I read you correctly, you're fine with this solution for now:

Could we settle with the original RFC proposal? Simply sync with everyone in configuration. No extra steps on configuration change or on a newly added node.S

[Serpentian]

Added the discussion on allowing disabled nodes to alternatives for #576. Created ticket #623, because allowing expelled nodes is also unsafe

[Gerold103]

Yes, I am ok with the current proposal. Plus I would like us to have a ticket for double-buckets automatic detection as I described above.

[sergepetrenko]

Plus I would like us to have a ticket for double-buckets automatic detection as I described above.

I think @Serpentian has mentioned this in #623

[Serpentian]

I did