storage: fix moved buckets ref check

Gerold103 · Gerold103 · commit f316ad1b6daf · 2024-10-10T18:52:25.000+02:00
During the partial Map-Reduce the router might visit some storages
more than once. Happens when after a ref on storage-A another
storage-B reports A as having taken some buckets.

Then router would come back to A to confirm that. The storage
still must hold its previously created ref in order for such
checks to make any sense. Otherwise any of the previously
confirmed buckets could have had escaped by now.

Without the ref-checking the router could reach the Map stage and
send some Map requests even though could detect earlier, that not
all storages would succeed.

This wasn't strictly speaking a bug, but it was clearly suboptimal
behaviour leading to the requests being executed not on all the
needed storages while the others would report errors.

NO_DOC=internal
diff --git a/test/router-luatest/map_part_test.lua b/test/router-luatest/map_part_test.lua
@@ -267,6 +267,18 @@ g.test_map_part_ref_timeout = function(cg)
         end
     end, {{bid1, bid2, bid3, bid4}})
 
+    -- Count the map calls. The loss of ref must be detected before the
+    -- map-stage.
+    local _, err = vtest.cluster_exec_each_master(cg, function()
+        rawset(_G, 'old_do_map', _G.do_map)
+        rawset(_G, 'map_count', 0)
+        _G.do_map = function(...)
+            _G.map_count = _G.map_count + 1
+            return _G.old_do_map(...)
+        end
+    end)
+    t.assert_equals(err, nil)
+
     -- Send bucket so the router thinks:
     --     rs1: {b1, b2}, rs2: {b3, b4}
     -- and actually the state is:
@@ -324,11 +336,20 @@ g.test_map_part_ref_timeout = function(cg)
     t.assert_equals(res.err_uuid, cg.rs2_uuid)
 
     -- Make sure there are no references left.
-    local _, err = vtest.cluster_exec_each(cg, function()
+    _, err = vtest.cluster_exec_each(cg, function()
         ilt.assert_equals(require('vshard.storage.ref').count, 0)
     end)
     t.assert_equals(err, nil)
 
+    -- No maps had a chance to get executed.
+    _, err = vtest.cluster_exec_each_master(cg, function()
+        ilt.assert_equals(_G.map_count, 0)
+        _G.do_map = _G.old_do_map
+        _G.old_do_map = nil
+        _G.map_count = nil
+    end)
+    t.assert_equals(err, nil)
+
     -- Return the bucket back and re-enable discovery on the router.
     cg.replica_2_a:exec(function(bid, to)
         _G.bucket_send(bid, to)
diff --git a/test/storage-luatest/storage_1_test.lua b/test/storage-luatest/storage_1_test.lua
@@ -236,47 +236,49 @@ test_group.test_ref_with_buckets_basic = function(g)
 
         -- No buckets.
         res, err = ivshard.storage._call(
-            'storage_ref_with_buckets', rid, iwait_timeout, {})
+            'storage_ref_make_with_buckets', rid, iwait_timeout, {})
         ilt.assert_equals(err, nil)
         ilt.assert_equals(res, {moved = {}})
         ilt.assert_equals(lref.count, 0)
 
         -- Check for a single ok bucket.
         res, err = ivshard.storage._call(
-            'storage_ref_with_buckets', rid, iwait_timeout, {bids[1]})
+            'storage_ref_make_with_buckets', rid, iwait_timeout, {bids[1]})
         ilt.assert_equals(err, nil)
-        ilt.assert_equals(res, {rid = rid, moved = {}})
+        ilt.assert_equals(res, {is_done = true, moved = {}})
         ilt.assert_equals(lref.count, 1)
         _, err = ivshard.storage._call('storage_unref', rid)
         ilt.assert_equals(err, nil)
         ilt.assert_equals(lref.count, 0)
 
         -- Check for multiple ok buckets.
         res, err = ivshard.storage._call(
-            'storage_ref_with_buckets', rid, iwait_timeout, {bids[1], bids[2]})
+            'storage_ref_make_with_buckets', rid, iwait_timeout,
+            {bids[1], bids[2]})
         ilt.assert_equals(err, nil)
-        ilt.assert_equals(res, {rid = rid, moved = {}})
+        ilt.assert_equals(res, {is_done = true, moved = {}})
         _, err = ivshard.storage._call('storage_unref', rid)
         ilt.assert_equals(err, nil)
 
         -- Check for double referencing.
         res, err = ivshard.storage._call(
-            'storage_ref_with_buckets', rid, iwait_timeout, {bids[1], bids[1]})
+            'storage_ref_make_with_buckets', rid, iwait_timeout,
+            {bids[1], bids[1]})
         ilt.assert_equals(err, nil)
-        ilt.assert_equals(res, {rid = rid, moved = {}})
+        ilt.assert_equals(res, {is_done = true, moved = {}})
         ilt.assert_equals(lref.count, 1)
         _, err = ivshard.storage._call('storage_unref', rid)
         ilt.assert_equals(err, nil)
         ilt.assert_equals(lref.count, 0)
 
         -- Bucket mix.
         res, err = ivshard.storage._call(
-            'storage_ref_with_buckets', rid, iwait_timeout,
+            'storage_ref_make_with_buckets', rid, iwait_timeout,
             {bucket_count + 1, bids[1], bucket_count + 2, bids[2],
              bucket_count + 3})
         ilt.assert_equals(err, nil)
         ilt.assert_equals(res, {
-            rid = rid,
+            is_done = true,
             moved = {
                 {id = bucket_count + 1},
                 {id = bucket_count + 2},
@@ -288,7 +290,7 @@ test_group.test_ref_with_buckets_basic = function(g)
 
         -- No ref when all buckets are missing.
         res, err = ivshard.storage._call(
-            'storage_ref_with_buckets',
+            'storage_ref_make_with_buckets',
             rid,
             iwait_timeout,
             {bucket_count + 1, bucket_count + 2}
@@ -315,7 +317,7 @@ test_group.test_ref_with_buckets_timeout = function(g)
         box.space._bucket:update(
             {bids[1]}, {{'=', 2, ivconst.BUCKET.SENDING}})
         local res, err = ivshard.storage._call(
-            'storage_ref_with_buckets', rid, 0.01, {bids[2]})
+            'storage_ref_make_with_buckets', rid, 0.01, {bids[2]})
         box.space._bucket:update(
             {bids[1]}, {{'=', 2, ivconst.BUCKET.ACTIVE}})
         t.assert_str_contains(err.message, 'Timeout exceeded')
@@ -341,7 +343,7 @@ test_group.test_ref_with_buckets_return_last_known_dst = function(g)
         box.space._bucket:update(
             {bid}, {{'=', 2, ivconst.BUCKET.SENT}})
         local res, err = ivshard.storage._call(
-            'storage_ref_with_buckets', rid, iwait_timeout, {bid})
+            'storage_ref_make_with_buckets', rid, iwait_timeout, {bid})
         ilt.assert_equals(err, nil)
         ilt.assert_equals(res, {moved = {{
             id = bid,
@@ -378,7 +380,7 @@ test_group.test_ref_with_buckets_move_part_while_referencing = function(g)
         local session_id
         local f = ifiber.new(function()
             session_id = box.session.id()
-            return ivshard.storage._call('storage_ref_with_buckets', rid,
+            return ivshard.storage._call('storage_ref_make_with_buckets', rid,
                                          iwait_timeout, {bids[1], bids[2]})
         end)
         f:set_joinable(true)
@@ -395,7 +397,7 @@ test_group.test_ref_with_buckets_move_part_while_referencing = function(g)
         t.assert_equals(err, nil)
         ilt.assert_equals(res, {
             moved = {{id = bids[2], dst = id}},
-            rid = rid,
+            is_done = true,
         })
         -- Ref was done, because at least one bucket was ok.
         ilt.assert_equals(lref.count, 1)
@@ -428,7 +430,7 @@ test_group.test_ref_with_buckets_move_all_while_referencing = function(g)
             {bids[3]}, {{'=', 2, ivconst.BUCKET.SENDING}, {'=', 3, id}})
         -- Start referencing.
         local f = ifiber.new(function()
-            return ivshard.storage._call('storage_ref_with_buckets', rid,
+            return ivshard.storage._call('storage_ref_make_with_buckets', rid,
                                          iwait_timeout, {bids[1], bids[2]})
         end)
         f:set_joinable(true)
@@ -536,10 +538,8 @@ test_group.test_moved_buckets_various_statuses = function(g)
             {{'=', 2, ivconst.BUCKET.GARBAGE}})
         _bucket:delete({bid_404})
 
-        local res, err = ivshard.storage._call('moved_buckets', bids)
-        ilt.assert_equals(err, nil)
-        ilt.assert(res and res.moved)
-        ilt.assert_items_equals(res.moved, {
+        local moved = ivshard.storage.internal.bucket_get_moved(bids)
+        ilt.assert_items_equals(moved, {
             {
                 id = bid_sent,
                 dst = id_sent,
diff --git a/test/upgrade/upgrade.result b/test/upgrade/upgrade.result
@@ -174,13 +174,13 @@ vshard.storage._call('test_api', 1, 2, 3)
  | - - bucket_recv
  |   - bucket_test_gc
  |   - info
- |   - moved_buckets
  |   - rebalancer_apply_routes
  |   - rebalancer_request_state
  |   - recovery_bucket_stat
  |   - storage_map
  |   - storage_ref
- |   - storage_ref_with_buckets
+ |   - storage_ref_check_with_buckets
+ |   - storage_ref_make_with_buckets
  |   - storage_unref
  |   - test_api
  | - 1
diff --git a/vshard/router/init.lua b/vshard/router/init.lua
@@ -919,9 +919,11 @@ local function router_ref_storage_by_buckets(router, bucket_ids, timeout)
             if replicasets_to_map[id] then
                 -- Replicaset is already referenced on a previous iteration.
                 -- Simply get the moved buckets without double referencing.
-                args_ref = {'moved_buckets', buckets}
+                args_ref = {
+                    'storage_ref_check_with_buckets', rid, buckets}
             else
-                args_ref = {'storage_ref_with_buckets', rid, timeout, buckets}
+                args_ref = {
+                    'storage_ref_make_with_buckets', rid, timeout, buckets}
             end
             res, err = replicasets_all[id]:callrw('vshard.storage._call',
                                                   args_ref, opts_async)
@@ -941,7 +943,7 @@ local function router_ref_storage_by_buckets(router, bucket_ids, timeout)
                 err_id = id
                 goto fail
             end
-            -- Ref returns nil,err or {rid, moved}.
+            -- Ref returns nil,err or {is_done, moved}.
             res, err = res[1], res[2]
             if res == nil then
                 err_id = id
@@ -959,7 +961,7 @@ local function router_ref_storage_by_buckets(router, bucket_ids, timeout)
                 end
                 table.insert(bucket_ids, bid)
             end
-            if res.rid then
+            if res.is_done then
                 assert(not replicasets_to_map[id])
                 -- If there are no buckets on the replicaset, it would not be
                 -- referenced.
diff --git a/vshard/storage/init.lua b/vshard/storage/init.lua
@@ -3184,9 +3184,9 @@ end
 -- Lookup buckets which are definitely not going to recover into ACTIVE state
 -- under any circumstances.
 --
-local function storage_moved_buckets(bucket_ids)
+local function bucket_get_moved(bucket_ids)
     local allstatus = consts.BUCKET
-    local moved_buckets = {}
+    local res = {}
     for _, bucket_id in pairs(bucket_ids) do
         local bucket = box.space._bucket:get{bucket_id}
         local is_moved
@@ -3197,14 +3197,14 @@ local function storage_moved_buckets(bucket_ids)
             is_moved = status == allstatus.GARBAGE or status == allstatus.SENT
         end
         if is_moved then
-            table.insert(moved_buckets, {
+            table.insert(res, {
                 id = bucket_id,
                 dst = bucket and bucket.destination or M.route_map[bucket_id],
                 status = bucket and bucket.status,
             })
         end
     end
-    return { moved = moved_buckets }
+    return res
 end
 
 --
@@ -3228,12 +3228,12 @@ end
 --         are absent, the reference is not created and a nil reference id
 --         with the list of absent buckets is returned.
 --
-local function storage_ref_with_buckets(rid, timeout, bucket_ids)
-    local moved = storage_moved_buckets(bucket_ids).moved
+local function storage_ref_make_with_buckets(rid, timeout, bucket_ids)
+    local moved = bucket_get_moved(bucket_ids)
     if #moved == #bucket_ids then
         -- If all the passed buckets are absent, there is no need to create a
         -- ref.
-        return {rid = nil, moved = moved}
+        return {moved = moved}
     end
     local bucket_generation = M.bucket_generation
     local ok, err = storage_ref(rid, timeout)
@@ -3243,13 +3243,31 @@ local function storage_ref_with_buckets(rid, timeout, bucket_ids)
     if M.bucket_generation ~= bucket_generation then
         -- Need to redo it. Otherwise there is a risk that some buckets were
         -- moved while waiting for the ref.
-        moved = storage_moved_buckets(bucket_ids).moved
+        moved = bucket_get_moved(bucket_ids)
         if #moved == #bucket_ids then
             storage_unref(rid)
-            rid = nil
+            return {moved = moved}
         end
     end
-    return {rid = rid, moved = moved}
+    return {is_done = true, moved = moved}
+end
+
+--
+-- Check which buckets from the given list are moved out of this storage, while
+-- also making sure that the storage-ref is still in place.
+--
+-- Partial Map-Reduce makes a ref on the storages having any of the needed
+-- buckets, but then can come back if other storages report the already visited
+-- ones as having the needed buckets. Only makes sense to check, if this storage
+-- still holds the ref. Otherwise its previous guarantees given during the ref
+-- creation are all gone.
+--
+local function storage_ref_check_with_buckets(rid, bucket_ids)
+    local ok, err = lref.check(rid, box.session.id())
+    if not ok then
+        return nil, err
+    end
+    return {moved = bucket_get_moved(bucket_ids)}
 end
 
 --
@@ -3298,9 +3316,9 @@ service_call_api = setmetatable({
     rebalancer_apply_routes = rebalancer_apply_routes,
     rebalancer_request_state = rebalancer_request_state,
     recovery_bucket_stat = recovery_bucket_stat,
-    moved_buckets = storage_moved_buckets,
     storage_ref = storage_ref,
-    storage_ref_with_buckets = storage_ref_with_buckets,
+    storage_ref_make_with_buckets = storage_ref_make_with_buckets,
+    storage_ref_check_with_buckets = storage_ref_check_with_buckets,
     storage_unref = storage_unref,
     storage_map = storage_map,
     info = storage_service_info,
@@ -4182,6 +4200,7 @@ M.bucket_state_edges = bucket_state_edges
 
 M.bucket_are_all_rw = bucket_are_all_rw_public
 M.bucket_generation_wait = bucket_generation_wait
+M.bucket_get_moved = bucket_get_moved
 lregistry.storage = M
 
 --
