handle BucketRefError for single operations

Author: Satbek

crud/common/call.lua

@@ -8,6 +8,7 @@ local sharding_utils = require('crud.common.sharding.utils')
 local fiber = require('fiber')
 local const = require('crud.common.const')
 local rebalance = require('crud.common.rebalance')
+local bucket_ref_unref = require('crud.common.sharding.bucket_ref_unref')
 
 local BaseIterator = require('crud.common.map_call_cases.base_iter')
 local BasePostprocessor = require('crud.common.map_call_cases.base_postprocessor')
@@ -149,6 +150,7 @@ local function wrap_vshard_err(vshard_router, err, func_name, replicaset_id, buc
     ))
 end
 
+-- todo: rename it
 local function retry_call_with_master_discovery(vshard_router, replicaset,
                                                 method, func_name, func_args,
                                                 call_opts, mode, bucket_ids)
@@ -161,19 +163,23 @@ local function retry_call_with_master_discovery(vshard_router, replicaset,
     if err == nil then
         return resp, err
     end
-
+    
     -- This is a partial copy of error handling from vshard.router.router_call_impl()
     -- It is much simpler mostly because bucket_set() can't be accessed from outside vshard.
-    if err.name == 'WRONG_BUCKET' or
-            err.name == 'BUCKET_IS_LOCKED' or
-            err.name == 'TRANSFER_IS_IN_PROGRESS' then
-        vshard_router:_bucket_reset(err.bucket_id)
-
-        -- Substitute replicaset only for single bucket_id calls.
-        if err.destination and vshard_router.replicasets[err.destination] and #bucket_ids == 1 then
-            replicaset = vshard_router.replicasets[err.destination]
-        else
-            return nil, err
+    -- may be we cant get this -- todo: check
+    if err.class_name == bucket_ref_unref.BucketRefError.name then
+        for _, bucket_ref_err in pairs(err.bucket_ref_errs) do
+                local bucket_id = bucket_ref_err.bucket_id
+                local vshard_err = bucket_ref_err.vshard_err
+                if vshard_err.name == 'WRONG_BUCKET' or
+                    vshard_err.name == 'BUCKET_IS_LOCKED' or
+                    vshard_err.name == 'TRANSFER_IS_IN_PROGRESS' then
+                vshard_router:_bucket_reset(bucket_id)
+
+                if vshard_err.destination and vshard_router.replicasets[vshard_err.destination] then
+                    replicaset = vshard_router.replicasets[vshard_err.destination]
+                end
+            end
         end
     elseif err.name == 'MISSING_MASTER' and replicaset.locate_master ~= nil then
         replicaset:locate_master()

crud/common/sharding/bucket_ref_unref.lua

@@ -9,8 +9,22 @@
 --- safe_mode change during execution.
 
 local vshard = require('vshard')
+local errors = require('errors')
 
-local M = {}
+local M = {
+    BucketRefError = errors.new_class('bucket_ref_error', {capture_stack = false})
+}
+
+local function make_bucket_ref_err(bucket_id, vshard_ref_err)
+    local err = M.BucketRefError:new(M.BucketRefError:new("failed bucket_ref: %s, bucket_id: %s", vshard_ref_err.name, bucket_id))
+    err.bucket_ref_errs = {
+        {
+            bucket_id = bucket_id,
+            vshard_err = vshard_ref_err,
+        }
+    }
+    return err
+end
 
 --- on module initialization safe_mode_status func must be set
 --- it's rebalance.safe_mode_status
@@ -31,7 +45,13 @@ function M.bucket_refrw(bucket_id)
     if not M.safe_mode_status() then
         return true
     end
-    return vshard.storage.bucket_refrw(bucket_id)
+
+    local ref_ok, vshard_ref_err = vshard.storage.bucket_refrw(bucket_id)
+    if not ref_ok then
+        return false, make_bucket_ref_err(bucket_id, vshard_ref_err)
+    end
+
+    return true
 end
 
 --- bucket_unrefrw calls vshard.storage.bucket_unrefrw if safe mode enabled
@@ -41,6 +61,7 @@ function M.bucket_unrefrw(bucket_id)
     if not M.safe_mode_status() then
         return true
     end
+
     return vshard.storage.bucket_unrefrw(bucket_id)
 end
 
@@ -50,7 +71,13 @@ function M.bucket_refro(bucket_id)
     if not M.safe_mode_status() then
         return true
     end
-    return vshard.storage.bucket_refro(bucket_id)
+
+    local ref_ok, vshard_ref_err = vshard.storage.bucket_refro(bucket_id)
+    if not ref_ok then
+        return false, make_bucket_ref_err(bucket_id, vshard_ref_err)
+    end
+
+    return true
 end
 
 --- bucket_unrefro calls vshard.storage.bucket_unrefro if safe mode enabled
@@ -62,28 +89,4 @@ function M.bucket_unrefro(bucket_id)
     return vshard.storage.bucket_unrefro(bucket_id)
 end
 
---- bucket_unrefro calls vshard.storage.bucket_unrefrw for every bucket if safe mode enabled
---- otherwise returns true
---- @param bucket_ids table<number, true>
-function M.bucket_unrefrw_many(bucket_ids)
-    if not M.safe_mode_status() then
-        return true
-    end
-
-    local unref_all_ok = true
-    local unref_last_err
-    for reffed_bucket_id in pairs(bucket_ids) do
-        local unref_ok, unref_err = M.bucket_unrefrw(reffed_bucket_id)
-        if not unref_ok then
-            unref_all_ok = nil
-            unref_last_err = unref_err
-        end
-    end
-
-    if not unref_all_ok then
-        return nil, unref_last_err
-    end
-    return true
-end
-
 return M

crud/delete.lua

@@ -45,9 +45,9 @@ local function delete_on_storage(space_name, key, field_names, opts)
     end
 
     local function make_delete()
-        local ref_ok, err_ref = bucket_ref_unref.bucket_refrw(opts.bucket_id)
+        local ref_ok, bucket_ref_err = bucket_ref_unref.bucket_refrw(opts.bucket_id)
         if not ref_ok then
-            return nil, err_ref
+            return nil, bucket_ref_err
         end
 
         -- add_space_schema_hash is false because

crud/get.lua

@@ -43,9 +43,9 @@ local function get_on_storage(space_name, key, field_names, opts)
         return nil, err
     end
 
-    local ref_ok, err_ref = bucket_ref_unref.bucket_refro(opts.bucket_id)
+    local ref_ok, bucket_ref_err = bucket_ref_unref.bucket_refro(opts.bucket_id)
     if not ref_ok then
-        return nil, err_ref
+        return nil, bucket_ref_err
     end
     -- add_space_schema_hash is false because
     -- reloading space format on router can't avoid get error on storage

crud/insert.lua

@@ -45,9 +45,10 @@ local function insert_on_storage(space_name, tuple, opts)
 
     local function make_insert()
         local bucket_id = tuple[utils.get_bucket_id_fieldno(space)]
-        local ref_ok, err_ref = bucket_ref_unref.bucket_refrw(bucket_id)
+        local ref_ok, bucket_ref_err = bucket_ref_unref.bucket_refrw(bucket_id)
+
         if not ref_ok then
-            return nil, err_ref
+            return nil, bucket_ref_err
         end
 
         -- add_space_schema_hash is true only in case of insert_object

crud/replace.lua

@@ -45,9 +45,9 @@ local function replace_on_storage(space_name, tuple, opts)
 
     local function make_replace()
         local bucket_id = tuple[utils.get_bucket_id_fieldno(space)]
-        local ref_ok, err_ref = bucket_ref_unref.bucket_refrw(bucket_id)
+        local ref_ok, bucket_ref_err = bucket_ref_unref.bucket_refrw(bucket_id)
         if not ref_ok then
-            return nil, err_ref
+            return nil, bucket_ref_err
         end
         -- add_space_schema_hash is true only in case of replace_object
         -- the only one case when reloading schema can avoid insert error

crud/update.lua

@@ -45,9 +45,9 @@ local function update_on_storage(space_name, key, operations, field_names, opts)
     end
 
     local function make_update()
-        local ref_ok, err_ref = bucket_ref_unref.bucket_refrw(opts.bucket_id)
+        local ref_ok, bucket_ref_err = bucket_ref_unref.bucket_refrw(opts.bucket_id)
         if not ref_ok then
-            return nil, err_ref
+            return nil, bucket_ref_err
         end
 
         -- add_space_schema_hash is false because

crud/upsert.lua

@@ -44,9 +44,9 @@ local function upsert_on_storage(space_name, tuple, operations, opts)
 
     local function make_upsert()
         local bucket_id = tuple[utils.get_bucket_id_fieldno(space)]
-        local ref_ok, err_ref = bucket_ref_unref.bucket_refrw(bucket_id)
+        local ref_ok, bucket_ref_err = bucket_ref_unref.bucket_refrw(bucket_id)
         if not ref_ok then
-            return nil, err_ref
+            return nil, bucket_ref_err
         end
 
         -- add_space_schema_hash is true only in case of upsert_object