Fix using outdated space schema (#111)

The problem is that if schema is changed on storages, router still uses
outdated schema from vshard connection. It leads to errors like 
"non-existent space" when space is already created. 
The solution is quite simple - all such errors leads to `conn:reload_schema()` 
call and then crud retries. In case of error on the storage-side (e.g. in `*_object` 
functions when object was flattened using outdated space format) space schema
hash is returned. Router compares this hash with hash of schema that was used 
on router-side, and reloads schema if hashes are different.
 `conn:reload_schema()` is called by one fiber at time - if reloading is in progress,
other fibers just wait for it to end.

Author: dokshina

CHANGELOG.md

@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 
 ## [Unreleased]
 
+### Fixed
+
+* Using outdated schema on router-side
+
 ### Added
 
 * Support for UUID field types and UUID values

crud/common/call.lua

@@ -160,6 +160,8 @@ function call.rw_single(bucket_id, func_name, func_args, options)
             )
         end
 
+        err = errors.wrap(err)
+
         return nil, CallError:new(utils.format_replicaset_error(
              replicaset.uuid, "Function returned an error: %s", err
         ))

crud/common/const.lua

@@ -0,0 +1,6 @@
+local const = {}
+
+const.RELOAD_RETRIES_NUM = 1
+const.RELOAD_SCHEMA_TIMEOUT = 3 -- 3 seconds
+
+return const

crud/common/schema.lua

@@ -0,0 +1,170 @@
+local fiber = require('fiber')
+local msgpack = require('msgpack')
+local digest = require('digest')
+local vshard = require('vshard')
+local errors = require('errors')
+local log = require('log')
+
+local ReloadSchemaError = errors.new_class('ReloadSchemaError',  {capture_stack = false})
+
+local const = require('crud.common.const')
+
+local schema = {}
+
+local function table_len(t)
+    local len = 0
+    for _ in pairs(t) do
+        len = len + 1
+    end
+    return len
+end
+
+local function call_reload_schema_on_replicaset(replicaset, channel)
+    replicaset.master.conn:reload_schema()
+    channel:put(true)
+end
+
+local function call_reload_schema(replicasets)
+    local replicasets_num = table_len(replicasets)
+    local channel = fiber.channel(replicasets_num)
+
+    local fibers = {}
+    for _, replicaset in pairs(replicasets) do
+        local f = fiber.new(call_reload_schema_on_replicaset, replicaset, channel)
+        table.insert(fibers, f)
+    end
+
+    for _ = 1,replicasets_num do
+        if channel:get(const.RELOAD_SCHEMA_TIMEOUT) == nil then
+            for _, f in ipairs(fibers) do
+                if fiber:status() ~= 'dead' then
+                    f:cancel()
+                end
+            end
+            return nil, ReloadSchemaError:new("Reloading schema timed out")
+        end
+    end
+
+    return true
+end
+
+local reload_in_progress = false
+local reload_schema_cond = fiber.cond()
+
+local function reload_schema(replicasets)
+    if reload_in_progress then
+        if not reload_schema_cond:wait(const.RELOAD_SCHEMA_TIMEOUT) then
+            return nil, ReloadSchemaError:new('Waiting for schema to be reloaded is timed out')
+        end
+    else
+        reload_in_progress = true
+
+        local ok, err = call_reload_schema(replicasets)
+        if not ok then
+            return nil, err
+        end
+
+        reload_schema_cond:broadcast()
+        reload_in_progress = false
+    end
+
+    return true
+end
+
+-- schema.wrap_func_reload calls func with specified arguments.
+-- func should return `res, err, need_reload`
+-- If function returned error and `need_reload` is true,
+-- then schema is reloaded and one more attempt is performed
+-- (but no more than RELOAD_RETRIES_NUM).
+-- This wrapper is used for functions that can fail if router uses outdated
+-- space schema. In case of such errors these functions returns `need_reload`
+-- for schema-dependent errors.
+function schema.wrap_func_reload(func, ...)
+    local i = 0
+
+    local res, err, need_reload
+    while true do
+        res, err, need_reload = func(...)
+
+        if err == nil or not need_reload then
+            break
+        end
+
+        local ok, reload_schema_err = reload_schema(vshard.router.routeall())
+        if not ok then
+            log.warn("Failed to reload schema: %s", reload_schema_err)
+            break
+        end
+
+        i = i + 1
+        if i > const.RELOAD_RETRIES_NUM then
+            break
+        end
+    end
+
+    return res, err
+end
+
+local function get_space_schema_hash(space)
+    if space == nil then
+        return ''
+    end
+
+    local indexes_info = {}
+    for i = 0, table.maxn(space.index) do
+        local index = space.index[i]
+        if index ~= nil then
+            indexes_info[i] = {
+                unique = index.unique,
+                parts = index.parts,
+                id = index.id,
+                type = index.type,
+                name = index.name,
+                path = index.path,
+            }
+        end
+    end
+
+    local space_info = {
+        format = space:format(),
+        indexes = indexes_info,
+    }
+
+    return digest.murmur(msgpack.encode(space_info))
+end
+
+-- schema.wrap_box_space_func_result pcalls some box.space function
+-- and returns its result as a table
+-- `{res = ..., err = ..., space_schema_hash = ...}`
+-- space_schema_hash is computed if function failed and
+-- `add_space_schema_hash` is true
+function schema.wrap_box_space_func_result(add_space_schema_hash, space, func_name, ...)
+    local result = {}
+
+    local ok, func_res = pcall(space[func_name], space, ...)
+    if not ok then
+        result.err = func_res
+        if add_space_schema_hash then
+            result.space_schema_hash = get_space_schema_hash(space)
+        end
+    else
+        result.res = func_res
+    end
+
+    return result
+end
+
+-- schema.result_needs_reload checks that schema reload can
+-- be helpful to avoid storage error.
+-- It checks if space_schema_hash returned by storage
+-- is the same as hash of space used on router.
+-- Note, that storage returns `space_schema_hash = nil`
+-- if reloading space format can't avoid the error.
+function schema.result_needs_reload(space, result)
+    if result.space_schema_hash == nil then
+        return false
+    end
+    return result.space_schema_hash ~= get_space_schema_hash(space)
+end
+
+return schema

crud/common/utils.lua

@@ -1,12 +1,15 @@
 local errors = require('errors')
 local ffi = require('ffi')
+local vshard = require('vshard')
 
+local schema = require('crud.common.schema')
 local dev_checks = require('crud.common.dev_checks')
 
 local FlattenError = errors.new_class("FlattenError", {capture_stack = false})
 local UnflattenError = errors.new_class("UnflattenError", {capture_stack = false})
 local ParseOperationsError = errors.new_class('ParseOperationsError',  {capture_stack = false})
 local ShardingError = errors.new_class('ShardingError',  {capture_stack = false})
+local GetSpaceFormatError = errors.new_class('GetSpaceFormatError',  {capture_stack = false})
 
 local utils = {}
 
@@ -34,30 +37,52 @@ end
 function utils.get_space(space_name, replicasets)
     local replicaset = select(2, next(replicasets))
     local space = replicaset.master.conn.space[space_name]
+
     return space
 end
 
+function utils.get_space_format(space_name, replicasets)
+    local space = utils.get_space(space_name, replicasets)
+    if space == nil then
+        return nil, GetSpaceFormatError:new("Space %q doesn't exist", space_name)
+    end
+
+    local space_format = space:format()
+
+    return space_format
+end
+
 local system_fields = { bucket_id = true }
 
 function utils.flatten(object, space_format, bucket_id)
     if object == nil then return nil end
 
     local tuple = {}
 
+    local fieldnames = {}
+
     for fieldno, field_format in ipairs(space_format) do
-        local value = object[field_format.name]
+        local fieldname = field_format.name
+        local value = object[fieldname]
 
-        if not system_fields[field_format.name] then
+        if not system_fields[fieldname] then
             if not field_format.is_nullable and value == nil then
-                return nil, FlattenError:new("Field %q isn't nullable", field_format.name)
+                return nil, FlattenError:new("Field %q isn't nullable", fieldname)
             end
         end
 
-        if bucket_id ~= nil and field_format.name == 'bucket_id' then
+        if bucket_id ~= nil and fieldname == 'bucket_id' then
             value = bucket_id
         end
 
         tuple[fieldno] = value
+        fieldnames[fieldname] = true
+    end
+
+    for fieldname in pairs(object) do
+        if not fieldnames[fieldname] then
+            return nil, FlattenError:new("Unknown field %q is specified", fieldname)
+        end
     end
 
     return tuple
@@ -124,7 +149,7 @@ local function determine_enabled_features()
     enabled_tarantool_features.uuids = major >= 2 and (minor > 4 or minor == 4 and patch >= 1)
 end
 
-local function tarantool_supports_fieldpaths()
+function utils.tarantool_supports_fieldpaths()
     if enabled_tarantool_features.fieldpaths == nil then
         determine_enabled_features()
     end
@@ -141,7 +166,7 @@ function utils.tarantool_supports_uuids()
 end
 
 function utils.convert_operations(user_operations, space_format)
-    if tarantool_supports_fieldpaths() then
+    if utils.tarantool_supports_fieldpaths() then
         return user_operations
     end
 
@@ -226,4 +251,29 @@ function utils.is_uuid(value)
     return ffi.istype(uuid_t, value)
 end
 
+function utils.format_result(rows, space)
+    return {
+        metadata = table.copy(space:format()),
+        rows = rows,
+    }
+end
+
+local function flatten_obj(space_name, obj)
+    local space_format, err = utils.get_space_format(space_name, vshard.router.routeall())
+    if err ~= nil then
+        return nil, FlattenError:new("Failed to get space format: %s", err), true
+    end
+
+    local tuple, err = utils.flatten(obj, space_format)
+    if err ~= nil then
+        return nil, FlattenError:new("Object is specified in bad format: %s", err), true
+    end
+
+    return tuple
+end
+
+function utils.flatten_obj_reload(space_name, obj)
+    return schema.wrap_func_reload(flatten_obj, space_name, obj)
+end
+
 return utils