call: support vshard's request_timeout in read calls

vshard 0.1.28 introduces a new option for read requests:
`request_timeout`.

`request_timeout` serves as a protection against hung replicas. When it
is specified, it must be <= `timeout`, and it introduces new behavior
for router calls: they start retrying TimedOut errors, each time trying
the request on a new replica, as long as `timeout` (the old one)
permits.

Support the option only in crud requests with `mode = 'read'`, since
it makes no sense with `mode = 'write'`. There is no one to retry the
request on if master is unresponsive.

Author: sergepetrenko

CHANGELOG.md

@@ -5,6 +5,15 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/)
 and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.html).
 
+## [Unreleased]
+
+### Added
+* Support for `vshard`'s `request_timeout` parameter for calls with `mode = 'read'`
+
+### Changed
+
+### Fixed
+
 ## [1.6.0] - 09-09-25
 
 ### Changed

README.md

@@ -544,6 +544,9 @@ where:
   * `bucket_id` (`?number|cdata`) - bucket ID
   * `timeout` (`?number`) - `vshard.call` timeout and vshard master
     discovery timeout (in seconds), default value is 2
+  * `request_timeout` (`?number`) - `vshard.call` `request_timeout` parameter.
+    Can be specified only with `mode` = `read`, and default value is the same as
+    user-passed `timeout`.
   * `mode` (`?string`, `read` or `write`) - if `write` is specified then `get` is
     performed on master, default value is `read`
   * `prefer_replica` (`?boolean`) - if `true` then the preferred target is one of
@@ -1103,6 +1106,9 @@ where:
      then the map call is performed without any optimizations even
      if full primary key equal condition is specified
   * `timeout` (`?number`) - `vshard.call` timeout (in seconds)
+  * `request_timeout` (`?number`) - `vshard.call` `request_timeout` parameter.
+    Can be specified only with `mode` = `read`, and default value is the same as
+    user-passed `timeout`.
   * `fields` (`?table`) - field names for getting only a subset of fields
   * `fullscan` (`?boolean`) - if `true` then a critical log entry will be skipped
     on potentially long `select`, see [avoiding full scan](doc/select.md#avoiding-full-scan).
@@ -1200,6 +1206,9 @@ where:
 * `index_id` (`?string|number`) - index name or index id. Primary index by default
 * `opts`:
   * `timeout` (`?number`) - `vshard.call` timeout (in seconds)
+  * `request_timeout` (`?number`) - `vshard.call` `request_timeout` parameter.
+    Can be specified only with `mode` = `read`, and default value is the same as
+    user-passed `timeout`.
   * `fields` (`?table`) - field names for getting only a subset of fields
   * `mode` (`?string`, `read` or `write`) - if `write` is specified then `select` is
     performed on master, default value is `read`
@@ -1428,6 +1437,10 @@ where:
 * `space_name` (`string`) - name of the space
 * `conditions` (`?table`) - array of [conditions](#select-conditions)
 * `opts`:
+  * `timeout` (`?number`) - `vshard.call` timeout (in seconds)
+  * `request_timeout` (`?number`) - `vshard.call` `request_timeout` parameter.
+    Can be specified only with `mode` = `read`, and default value is the same as
+    user-passed `timeout`.
   * `yield_every` (`?number`) - number of tuples processed to yield after,
     `yield_every` should be > 0, default value is 1000
   * `timeout` (`?number`) - `vshard.call` timeout and vshard master
@@ -1733,7 +1746,7 @@ local result, err = rv:select(space_name, conditions, opts)
 rv:close()
 ```
 
-Opts are the same as [select opts](#select), except `balance`, `prefer_replica` and `mode` are not supported.
+Opts are the same as [select opts](#select), except `balance`, `request_timeout`, `prefer_replica` and `mode` are not supported.
 
 Returns metadata and array of rows, error.
 

crud/borders.lua

@@ -69,6 +69,7 @@ end
 local function call_get_border_on_router(vshard_router, border_name, space_name, index_name, opts)
     checks('table', 'string', 'string', '?string|number', {
         timeout = '?number',
+        request_timeout = '?number',
         fields = '?table',
         mode = '?string',
         vshard_router = '?string|table',
@@ -105,10 +106,12 @@ local function call_get_border_on_router(vshard_router, border_name, space_name,
     if err ~= nil then
         return nil, BorderError:new("Failed to get router replicasets: %s", err)
     end
+    local mode = opts.mode or 'read'
     local call_opts = {
-        mode = opts.mode or 'read',
+        mode = mode,
         replicasets = replicasets,
         timeout = opts.timeout,
+        request_timeout = mode == 'read' and opts.request_timeout or nil,
     }
     local results, err, storages_info = call.map(vshard_router,
         CRUD_STAT_FUNC_NAME,
@@ -198,6 +201,10 @@ end
 -- @tparam ?number opts.timeout
 --  Function call timeout
 --
+-- @tparam ?number opts.request_timeout
+--  vshard call request_timeout
+--  default is the same as opts.timeout
+--
 -- @tparam ?table opts.fields
 --  Field names for getting only a subset of fields
 --
@@ -226,6 +233,10 @@ end
 -- @tparam ?number opts.timeout
 --  Function call timeout
 --
+-- @tparam ?number opts.request_timeout
+--  vshard call request_timeout
+--  default is the same as opts.timeout
+--
 -- @tparam ?table opts.fields
 --  Field names for getting only a subset of fields
 --

crud/common/call.lua

@@ -108,6 +108,7 @@ function call.map(vshard_router, func_name, func_args, opts)
         prefer_replica = '?boolean',
         balance = '?boolean',
         timeout = '?number',
+        request_timeout = '?number',
         replicasets = '?table',
         iter = '?table',
         postprocessor = '?table',
@@ -139,7 +140,10 @@ function call.map(vshard_router, func_name, func_args, opts)
     end
 
     local futures_by_replicasets = {}
-    local call_opts = {is_async = true}
+    local call_opts = {
+        is_async = true,
+        request_timeout = opts.mode == 'read' and opts.request_timeout or nil,
+    }
     while iter:has_next() do
         local args, replicaset, replicaset_id = iter:get()
 
@@ -201,6 +205,7 @@ function call.single(vshard_router, bucket_id, func_name, func_args, opts)
         prefer_replica = '?boolean',
         balance = '?boolean',
         timeout = '?number',
+        request_timeout = '?number',
     })
 
     local vshard_call_name, err = call.get_vshard_call_name(opts.mode, opts.prefer_replica, opts.balance)
@@ -214,9 +219,11 @@ function call.single(vshard_router, bucket_id, func_name, func_args, opts)
     end
 
     local timeout = opts.timeout or const.DEFAULT_VSHARD_CALL_TIMEOUT
+    local request_timeout = opts.mode == 'read' and opts.request_timeout or nil
 
     local res, err = retry_call_with_master_discovery(replicaset, vshard_call_name,
-        func_name, func_args, {timeout = timeout})
+        func_name, func_args, {timeout = timeout,
+                               request_timeout = request_timeout})
     if err ~= nil then
         return nil, wrap_vshard_err(vshard_router, err, func_name, nil, bucket_id)
     end

crud/count.lua

@@ -110,6 +110,7 @@ end
 local function call_count_on_router(vshard_router, space_name, user_conditions, opts)
     checks('table', 'string', '?table', {
         timeout = '?number',
+        request_timeout = '?number',
         bucket_id = '?',
         force_map_call = '?boolean',
         fullscan = '?boolean',
@@ -221,11 +222,14 @@ local function call_count_on_router(vshard_router, space_name, user_conditions,
 
     local yield_every = opts.yield_every or const.DEFAULT_YIELD_EVERY
 
+    local mode = opts.mode or 'read'
+
     local call_opts = {
-        mode = opts.mode or 'read',
+        mode = mode,
         prefer_replica = opts.prefer_replica,
         balance = opts.balance,
         timeout = opts.timeout,
+        request_timeout = mode == 'read' and opts.request_timeout or nil,
         replicasets = replicasets_to_count,
     }
 
@@ -282,6 +286,10 @@ end
 --  Function call timeout in seconds,
 --  default value is 2 seconds
 --
+-- @tparam ?number opts.request_timeout
+--  vshard call request_timeout,
+--  default is the same as opts.timeout
+--
 -- @tparam ?number opts.bucket_id
 --  Bucket ID
 --  default is vshard.router.bucket_id_strcrc32 of primary key
@@ -317,6 +325,7 @@ end
 function count.call(space_name, user_conditions, opts)
     checks('string', '?table', {
         timeout = '?number',
+        request_timeout = '?number',
         bucket_id = '?',
         force_map_call = '?boolean',
         fullscan = '?boolean',

crud/get.lua

@@ -58,6 +58,7 @@ get.storage_api = {[GET_FUNC_NAME] = get_on_storage}
 local function call_get_on_router(vshard_router, space_name, key, opts)
     dev_checks('table', 'string', '?', {
         timeout = '?number',
+        request_timeout = '?number',
         bucket_id = '?',
         fields = '?table',
         prefer_replica = '?boolean',
@@ -116,11 +117,14 @@ local function call_get_on_router(vshard_router, space_name, key, opts)
         fetch_latest_metadata = opts.fetch_latest_metadata,
     }
 
+    local mode = opts.mode or 'read'
+
     local call_opts = {
-        mode = opts.mode or 'read',
+        mode = mode,
         prefer_replica = opts.prefer_replica,
         balance = opts.balance,
         timeout = opts.timeout,
+        request_timeout = mode == 'read' and opts.request_timeout or nil,
     }
 
     local storage_result, err = call.single(vshard_router,
@@ -174,6 +178,10 @@ end
 -- @tparam ?number opts.timeout
 --  Function call timeout
 --
+-- @tparam ?number opts.request_timeout
+--  vshard call request_timeout
+--  default is the same as opts.timeout
+--
 -- @tparam ?number opts.bucket_id
 --  Bucket ID
 --  (by default, it's vshard.router.bucket_id_strcrc32 of primary key)
@@ -196,6 +204,7 @@ end
 function get.call(space_name, key, opts)
     checks('string', '?', {
         timeout = '?number',
+        request_timeout = '?number',
         bucket_id = '?',
         fields = '?table',
         prefer_replica = '?boolean',

crud/select/compat/select.lua

@@ -214,6 +214,7 @@ function select_module.pairs(space_name, user_conditions, opts)
         prefer_replica = '?boolean',
         balance = '?boolean',
         timeout = '?number',
+        request_timeout = '?number',
         readview = '?boolean',
         readview_info = '?table',
 
@@ -247,6 +248,10 @@ function select_module.pairs(space_name, user_conditions, opts)
         if opts.vshard_router ~= nil then
             return nil, SelectError:new("Readview does not support 'vshard_router' option")
         end
+
+        if opts.request_timeout ~= nil then
+            return nil, SelectError:new("Readview does not support 'request_timeout' option")
+        end
     end
 
     if opts.first ~= nil and opts.first < 0 then
@@ -271,6 +276,8 @@ function select_module.pairs(space_name, user_conditions, opts)
             prefer_replica = opts.prefer_replica,
             balance = opts.balance,
             timeout = opts.timeout,
+            request_timeout = (opts.mode == 'read' or opts.mode == nil) and
+                              opts.request_timeout or nil,
             fetch_latest_metadata = opts.fetch_latest_metadata,
         },
         readview = opts.readview,
@@ -338,6 +345,7 @@ local function select_module_call_xc(vshard_router, space_name, user_conditions,
         prefer_replica = '?boolean',
         balance = '?boolean',
         timeout = '?number',
+        request_timeout = '?number',
         readview = '?boolean',
         readview_info = '?table',
 
@@ -362,6 +370,10 @@ local function select_module_call_xc(vshard_router, space_name, user_conditions,
         if opts.vshard_router ~= nil then
             return nil, SelectError:new("Readview does not support 'vshard_router' option")
         end
+
+        if opts.request_timeout ~= nil then
+            return nil, SelectError:new("Readview does not support 'request_timeout' option")
+        end
     end
 
     if opts.first ~= nil and opts.first < 0 then
@@ -383,6 +395,8 @@ local function select_module_call_xc(vshard_router, space_name, user_conditions,
             prefer_replica = opts.prefer_replica,
             balance = opts.balance,
             timeout = opts.timeout,
+            request_timeout = (opts.mode == 'read' or opts.mode == nil) and
+                              opts.request_timeout or nil,
             fetch_latest_metadata = opts.fetch_latest_metadata,
         },
         readview = opts.readview,

crud/select/compat/select_old.lua

@@ -54,10 +54,13 @@ local function select_iteration(space_name, plan, opts)
         space_name, plan.index_id, plan.conditions, storage_select_opts,
     }
 
+    local mode = call_opts.mode or 'read'
+
     local results, err, storages_info = call.map(opts.vshard_router, common.SELECT_FUNC_NAME, storage_select_args, {
         replicasets = opts.replicasets,
         timeout = call_opts.timeout,
-        mode = call_opts.mode or 'read',
+        request_timeout = mode == 'read' and call_opts.request_timeout or nil,
+        mode = mode,
         prefer_replica = call_opts.prefer_replica,
         balance = call_opts.balance,
     })
@@ -355,6 +358,8 @@ local function select_module_call_xc(vshard_router, space_name, user_conditions,
         end
     end
 
+    local mode = opts.mode or 'read'
+
     local iterator_opts = {
         after = opts.after,
         first = opts.first,
@@ -364,10 +369,11 @@ local function select_module_call_xc(vshard_router, space_name, user_conditions,
         field_names = opts.fields,
         yield_every = opts.yield_every,
         call_opts = {
-            mode = opts.mode,
+            mode = mode,
             prefer_replica = opts.prefer_replica,
             balance = opts.balance,
             timeout = opts.timeout,
+            request_timeout = mode == 'read' and opts.request_timeout or nil,
             fetch_latest_metadata = opts.fetch_latest_metadata,
         },
     }
@@ -439,6 +445,7 @@ function select_module.call(space_name, user_conditions, opts)
         prefer_replica = '?boolean',
         balance = '?boolean',
         timeout = '?number',
+        request_timeout= '?number',
 
         vshard_router = '?string|table',
 

test/integration/borders_test.lua

@@ -367,7 +367,12 @@ pgroup.test_opts_not_damaged = function(g)
     })
 
     -- min
-    local min_opts = {timeout = 1, fields = {'name', 'age'}, mode = 'write'}
+    local min_opts = {
+        timeout = 1,
+        request_timeout = 1,
+        fields = {'name', 'age'},
+        mode = 'write',
+    }
     local new_min_opts, err = g.router:eval([[
         local crud = require('crud')
 
@@ -382,7 +387,12 @@ pgroup.test_opts_not_damaged = function(g)
     t.assert_equals(new_min_opts, min_opts)
 
     -- max
-    local max_opts = {timeout = 1, fields = {'name', 'age'}, mode = 'write'}
+    local max_opts = {
+        timeout = 1,
+        request_timeout = 1,
+        fields = {'name', 'age'},
+        mode = 'write',
+    }
     local new_max_opts, err = g.router:eval([[
         local crud = require('crud')
 

test/integration/count_test.lua

@@ -614,6 +614,7 @@ pgroup.test_opts_not_damaged = function(g)
 
     local count_opts = {
         timeout = 1,
+        request_timeout = 1,
         bucket_id = 1161,
         yield_every = 105,
         mode = 'read',