Doubled buckets. Part 2 (stray tcp)

[mrForza]

Before this patch the rebalancer and recovery service could start
just right after master switch (by auto master detection or manual
reconfiguration) before the master had time to sync its vclock with
other replicas in replicaset. It could lead to doubled buckets according
to "Doubled buckets RFC".

To fix it we introduce a new storage service - on_master_enable
service. If master is changed in replicaset, this service is triggered
and waits until newly elected master syncs its vclock with other
replicas. Other storage services - rebalancer and recovery can't
start until on_master_enable set M.buckets_are_in_sync.

Closes #214

NO_TEST=bugfix
NO_DOC=bugfix

[Serpentian]

First part of review, mostly whining about naming and comments, the only serious one from my point of view is boostrap.

[Serpentian]

Nit: the commit "replicaset: refactoring of wait_masters_connect"

It was done in order to easier to use extra parameters in this function (e.g. - except in #214).

It sounds strange a little bit, maybe:

It was done to make it easier to use extra parameters in this function, such as except in gh-214.

[mrForza]

Yes, it sounds better, fixed

[Serpentian]

To hell with it, it's not critical, there are more important things worth spending time on. This one is already tested with more high level functions anyway

[Serpentian]

Nit:

that is it has the same vclock

I'd rather drop that part, it's misleading. We don't wait for vclock of the instances anymore, just the vclock of changes in _bucket.

[mrForza]

fixed

[Serpentian]

Nit: maybe are_buckets_in_sync or is_bucket_space_in_sync? The current one sounds a little bit awkward

[mrForza]

I like is_bucket_space_in_sync, but it might look bulky just a bit.

Vlad prefers the current name of this variable - #646 (comment).

[Gerold103]

The motivation is that we already have is_bucket_protected. And some other members refer to bucket space without mentioning "space" explicitly. Like bucket_on_replace, bucket_on_truncate.

[Serpentian]

Nit: should be vclock after a last transaction. It's not guaranteed, that txn over _bucket will have that vclock.

[mrForza]

fixed

[Serpentian]

Let's better initialize it during first vshard.storage.cfg and add an assertion here, that it's not nil. Current approach is way too error prone

[mrForza]

fixed

[Serpentian]

It's not a good idea to allow bootstrapping the cluster without any synchronization. Consider the case:

1. User bootstrapps the cluster, only on 1 replicaset buckets are created.
2. Master is changed in that replicaset. Old master still considers itself a master (due to misconfiguration), it considers itself syncrhonized, since it did that between 1 and 2.
3. User calls boostrap one more time, it passes, since the new leader doesn't have the buckets, created on the old one.
4. Old master considers itself in sync, sends the buckets to another replicaset
5. We have doubled buckets.

Instead, I propose to make map call over masters in cluster_bootstrap and wait there for masters to be in sync. After that we make map call one more time and create the buckets (with protection for is_sync). Though, the bootstrap can still fail, if master changes between these two map cals, though, it seems we're ok with it now, since partial boostrap is anyway possible now, if master changes and router doesn't know about that yet.

We should drop the must_be_synced from the wrapper.

[mrForza]

done

[Serpentian]

Nit: let's mention the _bucket in the name of the function

[mrForza]

fixed

[Serpentian]

See, you had pcalled it here, now you don't. This will lead to not calling the lsched.move_end in case of error, which is a bug

[mrForza]

fixed

[Serpentian]

Reminder

[Serpentian]

The remaining comments for storage: introduce on_master_enable service commit

[Serpentian]

Can we move bucket wrapper part to the separate commit? It's very difficult to parse the commit, imho

[Serpentian]

Maybe error instead of waiting? I understand, why do we wait in services, we cannot kill them. Though, we can do that for functions, which are called by these services: rebalancer_service_apply_routes_f, recovery_bucket_stat (you've actually already throw error in bucket stat)

[mrForza]

If I return an error from rebalancer_service_apply_routes_f, the calling node won't be able to receive it, since the function is invoked via pcall in a detached fiber. I consider to simply log the error and wait for the master to synchronize. A limiter is used to prevent log spam.

[Serpentian]

Hmm, let's won't introduce the limiter in that commit, we already have the ticket in order to properly limit the messages in that service, let's won't do that in the scope of this patchset, we don't have time for that: #611

Let's instead just throw error and that's it, as I proposed above.

[mrForza]

#646 (comment)

[Serpentian]

Maybe we can move that fiber creation to the master_on_enable and there create the fiber unconditionally? I'm really worried about these checks.

The problem with master_role_update is that it's the function, which is called several times even during single vshard.storage.cfg, it's not the case for master_role_update.

Or we should move the stopping of the fiber in master_role_update alternatively. We should make them consistent: either start/stop in master_role_update or start/stop in master_on_enable/disable.

Update: hmm, you already have it there, then why do you need it in master_role_update?

[mrForza]

Now we create on_master_enable fiber in master_on_enable and cancel it in master_on_disable.

We need it in master_role_update because otherwise the upgraded storage will not start on_master_enable service.

[Serpentian]

Let's add assertions for is_sync == false and M.fiber == nil

[mrForza]

done

[Serpentian]

Comments for the last 3 commits

[Serpentian]

Nit: not needed change

[Serpentian]

The except is useless here. You pass the rs_id, which is used in the wait_masters_connect, when the replicas_map_call requires replica_id in except

[Serpentian]

Nit: let's name it ERRINJ_FIRST_RECEIVE_DELAY then. Since it's activated only on the first call of the recv. To be consistent with ERRINJ_LAST_RECEIVE_DELAY.

[Serpentian]

Nit: let's log on the higher level, in the recovery_step_by_type. Let's try placing all logs in one place and not hide them inside other functions, when it's possible

[Serpentian]

Assertion here for remote_bucket.generation > generation

[Serpentian]

Why? If some tests doesn't work with sync, then only they should be disabled (of course with description, why they don't work)

[Serpentian]

We can majorly simplify the recovery process. Smth like that, though it's not debugged:

diff --git a/vshard/storage/init.lua b/vshard/storage/init.lua
index 4f875cb..66f1e39 100644
--- a/vshard/storage/init.lua
+++ b/vshard/storage/init.lua
@@ -94,6 +94,7 @@ if not M then
             ERRINJ_LAST_RECEIVE_DELAY = false,
             ERRINJ_LAST_SEND_DELAY = false,
             ERRINJ_RECEIVE_PARTIALLY = false,
+            ERRINJ_RECEIVE_DELAY = false,
             ERRINJ_RECOVERY_PAUSE = false,
             ERRINJ_DISCOVERY = false,
             ERRINJ_BUCKET_GC_PAUSE = false,
@@ -1005,6 +1006,38 @@ local function recover_bucket_to_state(recovered_buckets, bucket_id, state)
     recovered_buckets[state] = ids
 end

+local function recovery_search_missed_bucket(bucket_id)
+    local opts = {timeout = consts.RECOVERY_GET_STAT_TIMEOUT,
+                  except = M.this_replicaset.id}
+    local res, err, err_id = lreplicaset.cluster_map_master_call(
+        M.replicasets, 'vshard.storage._call',
+        {'recovery_bucket_stat', bucket_id}, opts)
+    if err then
+        err.master_id = err_id
+        return nil, err
+    end
+    local res_id, res_bucket_info
+    for rs_id, bucket_stat in pairs(res) do
+        local bucket_info, err = bucket_stat[1], bucket_stat[2]
+        if err then
+            err.master_id = err_id
+            return nil, err
+        end
+        -- Among all the masters we search those one which has the missed
+        -- remote bucket with higher generation.
+        if bucket_info and (not res_bucket_info or
+            res_bucket_info.generation < bucket_info.generation) then
+            res_id = rs_id
+            res_bucket_info = bucket_info
+        end
+    end
+    if res_bucket_info then
+        log.info('The bucket %s is found on replicaset %s',
+                 res_bucket_info.id, res_id)
+        return res_id, res_bucket_info
+    end
+end
+
 --
 -- Check status of each transferring bucket. Resolve status where
 -- possible.
@@ -1012,10 +1045,10 @@ end
 local function recovery_step_by_type(type, limiter)
     local _bucket = box.space._bucket
     local is_step_empty = true
-    local recovered = 0
     local total = 0
     local start_format = 'Starting %s buckets recovery step'
     local recovered_buckets = {}
+    local ok
     for _, bucket in _bucket.index.status:pairs(type) do
         lfiber.testcancel()
         total = total + 1
@@ -1073,30 +1106,55 @@ local function recovery_step_by_type(type, limiter)
             log.info(start_format, type)
         end
         lfiber.testcancel()
-        for state, checker in pairs(bucket_recoverability_checkers) do
-            if checker(bucket, remote_bucket) then
-                local _, err = recover_bucket_to_state(recovered_buckets,
-                                                       bucket_id, state)
-                if err then
-                    goto continue
+        local generation = bucket.opts.generation
+        if not remote_bucket then
+            err = lerror.make(string.format('No bucket %s on replicaset %s.',
+                                            bucket_id, peer_id))
+            limiter:log_warn(err, err.message .. ' Scanning all the masters')
+            peer_id, remote_bucket = recovery_search_missed_bucket(bucket_id)
+            if not peer_id and remote_bucket then
+                err = remote_bucket
+                limiter:log_error(err, 'Error during scanning the master ' ..
+                                  '%s: %s', err.master_id, err)
+                goto continue
+            end
+            assert(not remote_bucket or generation ~= remote_bucket.generation)
+        end
+        if remote_bucket and remote_bucket.generation > generation then
+            ok, err = recover_bucket_to_state(
+                recovered_buckets, bucket_id, BGARBAGE)
+        elseif remote_bucket and remote_bucket.generation < generation then
+            ok, err = recover_bucket_to_state(
+                recovered_buckets, bucket_id, BACTIVE)
+        else
+            assert(not remote_bucket or remote_bucket.generation == generation)
+            for state, checker in pairs(bucket_recoverability_checkers) do
+                if checker(bucket, remote_bucket) then
+                    ok, err = recover_bucket_to_state(recovered_buckets,
+                        bucket_id, state)
+                    if not ok then
+                        break
+                    end
+                    break
                 end
-                recovered = recovered + 1
-                break
             end
         end
-        if recovered == 0 and is_step_empty then
-            log.info('Bucket %s is %s local and %s on replicaset %s, waiting',
+        if not ok then
+            log.error(err)
+        end
+        if #recovered_buckets and is_step_empty then
+            log.info('Buckete%s is %s local and %s on replicaset %s, waiting',
                      bucket_id, bucket.status, remote_bucket.status, peer_id)
         end
         is_step_empty = false
 ::continue::
     end
-    if recovered > 0 then
+    if #recovered_buckets then
         log.info('Finish bucket recovery step, %d %s buckets are recovered '..
-                 'among %d. Recovered buckets: %s', recovered, type, total,
-                 json_encode(recovered_buckets))
+                 'among %d. Recovered buckets: %s', #recovered_buckets,
+                 type, total, json_encode(recovered_buckets))
     end
-    return total, recovered
+    return total, #recovered_buckets
 end

 --
@@ -1610,6 +1668,9 @@ local function bucket_recv_xc(bucket_id, from, data, opts)
         bucket_opts = {generation = opts.generation}
     end
     if not b then
+        while M.errinj.ERRINJ_RECEIVE_DELAY do
+            lfiber.sleep(0.01)
+        end
         if is_last then
             local msg = 'last message is received, but the bucket does not '..
                         'exist anymore'

[Gerold103]

You must be able to make vclock_lesseq() in storage/init.lua a one-line wrapper doing return vclock_compare(vc1, vc2, function(a, b) return a <= b end). I see no reason to keep 2 comparison functions, when one of them fully covers the other one. Hm?

[Gerold103]

In the commit's storage: introduce on_master_enable service message:

• received the _bucket space updats yes -> received the _bucket space updates yet.
• that they wouldn't failed -> that they wouldn't fail.

[Gerold103]

Lets not ignore errors, and also we must respect the timeout passed in opts so we don't wait infinitely even if the network is working fine.

[Gerold103]

Normally check means that it returns an error when the condition isn't true. This is the difference between check-functions and is-functions - the latter will just return true/false, while the former will throw or return nil, err when not true. Given that this function returns no error, it would be more appropriate to call it storage_is_bucket_sync().

But in this exact case I don't think you need this function at all. We have storage_service_info for that. Lets make it return one more key in its result-map. And name it is_master_sync. The name is_bucket_sync I think might be too low-level for not-fully-internal APIs.

[Gerold103]

I wouldn't say it is an error. It is a warning at worst. Also lets not create an error object for that, if we don't plan to return it.

Last thing - you mention here "during the start", but this check is never called just on start. You seem to always call it inside service loops, which means it might re-appear potentially long after start. Lets rephrase this to not mention "start".

[Gerold103]

I don't think users will know what on_master_enable means. If this is written to the logs, then lets make it more user-friendly. Like Master sync service stopped or similar. Or even remove this line entirely, because anyway 2 lines above we already say the result of this service.

[Gerold103]

You can't nullify on_master_enable_fiber unconditionally for the same reason why you can't nullify on_master_enable_service unconditionally. Because if this fiber was cancelled and another one was started, then you will remove the other fiber's ref here, not yours.

[Gerold103]

I understand that we named it 'on master enable', because it is started when master is enabled. But on_... is almost exclusively used by triggers, and it doesn't say what it actually does. For example, GC service and recovery are also started when master is enabled, but we can't call them 'on master enabled'. Lets please name this service appropriately, like master_sync service and master_sync_fiber.

[Gerold103]

It doesn't look right that the start and stop are scattered and duplicated across on-enable, on-disable, and role-update.

Isn't it enough to do this entire logic inside master_role_update()? Like all other master-services do? Both start and stop of this server + the flag reset.

[Gerold103]

This doesn't look right. This vclock is -- Vclock after a last transaction over the _bucket space. But here it gets violated, because the cfg might be done while there are bucket transactions in progress. I recommend to remove this from here, and instead patch storage_wait_vclock_persisted to assign M.bucket_latest_vclock = box.info.vclock if it was nil when we see 0 transactions, before return. Then it will be correct.