diff --git a/CHANGELOG.md b/CHANGELOG.md
index bdd18aa5..7082de68 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,6 +7,11 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 
 ## [Unreleased]
 
+### Fixed
+* Support standard `router -> storage` upgrades from CRUD versions older than
+  1.6.0 by falling back to direct storage calls when old storages do not have
+  `_crud.call_on_storage` yet.
+
 ## [1.7.4] - 12-02-26
 
 ### Fixed
diff --git a/README.md b/README.md
index 55cb850f..81939acf 100644
--- a/README.md
+++ b/README.md
@@ -19,6 +19,7 @@ It also provides the `crud-storage` and `crud-router` roles for
     - [Application dependency](#application-dependency)
     - [Repository clone](#repository-clone)
   - [Usage](#usage)
+  - [Upgrade from CRUD < 1.6.0](#upgrade-from-crud--160)
   - [Sandbox](#sandbox)
 - [API](#api)
   - [Package info](#package-info)
@@ -123,6 +124,40 @@ For Tarantool 1.10, 2.x and 3.x you can also manually call
 the [crud initialization code](#api) on [VShard](https://github.com/tarantool/vshard)
 router and storage instances.
 
+### Upgrade from CRUD < 1.6.0
+
+CRUD 1.6.0 and CRUD 1.7.0 introduced compatibility boundaries that require
+different upgrade orders.
+
+When upgrading from CRUD < 1.6.0 to CRUD >= 1.6.0 and < 1.7.0 (1.6.0 or
+1.6.1), update storages first:
+
+1. Update all storages to the target CRUD version.
+2. Make sure all storages are healthy and serve requests.
+3. Update routers to the same target CRUD version.
+
+CRUD 1.6.0 introduced router-to-storage calls through `_crud.call_on_storage`.
+Storages running CRUD older than 1.6.0 do not have this function. Updating
+storages first avoids the incompatible state where a new router calls
+`_crud.call_on_storage` on an old storage. Old routers can still call upgraded
+storages directly.
+
+When upgrading from CRUD >= 1.6.0 and < 1.7.0 to CRUD >= 1.7.0, use the
+[standard Tarantool replication cluster upgrade order](https://www.tarantool.io/en/doc/latest/admin/upgrades/upgrade_cluster/):
+update routers first, then storage replica sets.
+
+Direct upgrades from CRUD < 1.6.0 to CRUD >= 1.7.0 and <= 1.7.4 are not covered
+by this procedure. To upgrade from CRUD < 1.6.0 to CRUD 1.7.4, do it in two
+steps:
+
+1. Upgrade to CRUD 1.6.0 or 1.6.1 using the storage-first order described above.
+2. Upgrade from CRUD 1.6.0 or 1.6.1 to CRUD 1.7.4 using the standard
+   Tarantool/vshard order: routers first, then storages.
+
+When upgrading from CRUD < 1.6.0 to CRUD > 1.7.4, router-side compatibility
+mode is available, so use the standard Tarantool/vshard order: update routers
+first, then storage replica sets.
+
 > [!NOTE]
 >
 > After changing the cluster configuration (for example, adding a new replica set or changing their weights)
diff --git a/crud/common/call.lua b/crud/common/call.lua
index 53d35728..831930b7 100644
--- a/crud/common/call.lua
+++ b/crud/common/call.lua
@@ -2,6 +2,7 @@ local errors = require('errors')
 
 local call_cache = require('crud.common.call_cache')
 local dev_checks = require('crud.common.dev_checks')
+local storage_call = require('crud.common.storage_call')
 local yield_checks = require('crud.common.yield_checks')
 local utils = require('crud.common.utils')
 local sharding_utils = require('crud.common.sharding.utils')
@@ -15,16 +16,13 @@ local BasePostprocessor = require('crud.common.map_call_cases.base_postprocessor
 
 local CallError = errors.new_class('CallError')
 
-local CALL_FUNC_NAME = 'call_on_storage'
-local CRUD_CALL_FUNC_NAME = utils.get_storage_call(CALL_FUNC_NAME)
-
 local call = {}
 
 local function call_on_storage(run_as_user, func_name, ...)
     return yield_checks.guard(box.session.su, run_as_user, call_cache.func_name_to_func(func_name), ...)
 end
 
-call.storage_api = {[CALL_FUNC_NAME] = call_on_storage}
+call.storage_api = {[storage_call.CALL_ON_STORAGE_FUNC_NAME] = call_on_storage}
 
 function call.get_vshard_call_name(mode, prefer_replica, balance)
     dev_checks('string', '?boolean', '?boolean')
@@ -84,18 +82,45 @@ local function wrap_vshard_err(vshard_router, err, func_name, replicaset_id, buc
     ))
 end
 
+local function perform_storage_call(replicaset, method, replicaset_id, func_name, func_args, call_opts, force_legacy)
+    local call_func_name, call_args, used_legacy_call = storage_call.prepare(
+        replicaset_id, func_name, func_args, force_legacy
+    )
+
+    local resp, err = replicaset[method](replicaset, call_func_name, call_args, call_opts)
+    if err == nil and not used_legacy_call and not (call_opts or {}).is_async then
+        storage_call.mark_call_on_storage_supported(replicaset_id)
+    end
+
+    return resp, err, used_legacy_call
+end
+
+local function fallback_to_legacy_if_needed(replicaset, method, replicaset_id,
+    func_name, func_args, call_opts, resp, err, used_legacy_call)
+    if storage_call.should_fallback_to_legacy(replicaset_id, err, used_legacy_call) then
+        resp, err, used_legacy_call = perform_storage_call(
+            replicaset, method, replicaset_id, func_name, func_args, call_opts, true
+        )
+    end
+
+    return resp, err, used_legacy_call
+end
+
 --- Executes a vshard call and retries once after performing recovery actions
 --- like bucket cache reset, destination redirect (for single calls), or master discovery.
 local function call_with_retry_and_recovery(vshard_router,
-    replicaset, method, func_name, func_args, call_opts, is_single_call)
-    local func_args_ext = utils.append_array({ box.session.effective_user(), func_name }, func_args)
-
+    replicaset, replicaset_id, method, func_name, func_args, call_opts, is_single_call)
     -- In case cluster was just bootstrapped with auto master discovery,
     -- replicaset may miss master.
-    local resp, err = replicaset[method](replicaset, CRUD_CALL_FUNC_NAME, func_args_ext, call_opts)
+    local resp, err, used_legacy_call = perform_storage_call(
+        replicaset, method, replicaset_id, func_name, func_args, call_opts
+    )
+    resp, err, used_legacy_call = fallback_to_legacy_if_needed(
+        replicaset, method, replicaset_id, func_name, func_args, call_opts, resp, err, used_legacy_call
+    )
 
     if err == nil then
-        return resp, err
+        return resp, err, used_legacy_call
     end
 
     -- This is a partial copy of error handling from vshard.router.router_call_impl()
@@ -106,6 +131,7 @@ local function call_with_retry_and_recovery(vshard_router,
             local destination = single_err.vshard_err.destination
             if destination and vshard_router.replicasets[destination] then
                 replicaset = vshard_router.replicasets[destination]
+                replicaset_id = destination
             end
         end
 
@@ -124,7 +150,50 @@ local function call_with_retry_and_recovery(vshard_router,
 
     -- Retry only once: should be enough for initial discovery,
     -- otherwise force user fix up cluster bootstrap.
-    return replicaset[method](replicaset, CRUD_CALL_FUNC_NAME, func_args_ext, call_opts)
+    resp, err, used_legacy_call = perform_storage_call(
+        replicaset, method, replicaset_id, func_name, func_args, call_opts
+    )
+    resp, err, used_legacy_call = fallback_to_legacy_if_needed(
+        replicaset, method, replicaset_id, func_name, func_args, call_opts, resp, err, used_legacy_call
+    )
+
+    return resp, err, used_legacy_call
+end
+
+function call.storage_call(replicaset, method, replicaset_id, func_name, func_args, call_opts, force_legacy)
+    return perform_storage_call(replicaset, method, replicaset_id, func_name, func_args, call_opts, force_legacy)
+end
+
+local function wait_result_with_compat(future_info, wait_timeout)
+    local result, err = future_info.future:wait_result(wait_timeout)
+    local result_err = storage_call.result_error(result)
+
+    if err == nil and result_err == nil and not future_info.used_legacy_call then
+        storage_call.mark_call_on_storage_supported(future_info.replicaset_id)
+    end
+
+    if storage_call.should_fallback_to_legacy(
+        future_info.replicaset_id, err or result_err, future_info.used_legacy_call
+    ) then
+        local future, call_err, used_legacy_call = perform_storage_call(
+            future_info.replicaset,
+            future_info.method,
+            future_info.replicaset_id,
+            future_info.func_name,
+            future_info.func_args,
+            future_info.call_opts,
+            true
+        )
+        if call_err ~= nil then
+            return nil, call_err
+        end
+
+        future_info.future = future
+        future_info.used_legacy_call = used_legacy_call
+        result, err = future:wait_result(wait_timeout)
+    end
+
+    return result, err
 end
 
 function call.map(vshard_router, func_name, func_args, opts)
@@ -172,8 +241,8 @@ function call.map(vshard_router, func_name, func_args, opts)
     while iter:has_next() do
         local args, replicaset, replicaset_id = iter:get()
 
-        local future, err = call_with_retry_and_recovery(vshard_router, replicaset, vshard_call_name,
-            func_name, args, call_opts, false)
+        local future, err, used_legacy_call = call_with_retry_and_recovery(vshard_router, replicaset, replicaset_id,
+            vshard_call_name, func_name, args, call_opts, false)
 
         if err ~= nil then
             local result_info = {
@@ -192,17 +261,26 @@ function call.map(vshard_router, func_name, func_args, opts)
             return postprocessor:get()
         end
 
-        futures_by_replicasets[replicaset_id] = future
+        futures_by_replicasets[replicaset_id] = {
+            future = future,
+            replicaset = replicaset,
+            replicaset_id = replicaset_id,
+            method = vshard_call_name,
+            func_name = func_name,
+            func_args = args,
+            call_opts = call_opts,
+            used_legacy_call = used_legacy_call,
+        }
     end
 
     local deadline = fiber_clock() + timeout
-    for replicaset_id, future in pairs(futures_by_replicasets) do
+    for replicaset_id, future_info in pairs(futures_by_replicasets) do
         local wait_timeout = deadline - fiber_clock()
         if wait_timeout < 0 then
             wait_timeout = 0
         end
 
-        local result, err = future:wait_result(wait_timeout)
+        local result, err = wait_result_with_compat(future_info, wait_timeout)
 
         local result_info = {
             key = replicaset_id,
@@ -246,7 +324,8 @@ function call.single(vshard_router, bucket_id, func_name, func_args, opts)
     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 = call_with_retry_and_recovery(vshard_router, replicaset, vshard_call_name,
+    local replicaset_id = utils.get_replicaset_id(vshard_router, replicaset)
+    local res, err = call_with_retry_and_recovery(vshard_router, replicaset, replicaset_id, vshard_call_name,
         func_name, func_args, {timeout = timeout, request_timeout = request_timeout}, true)
     if err ~= nil then
         return nil, wrap_vshard_err(vshard_router, err, func_name, nil, bucket_id)
@@ -272,7 +351,7 @@ function call.any(vshard_router, func_name, func_args, opts)
     end
     local replicaset_id, replicaset = next(replicasets)
 
-    local res, err = call_with_retry_and_recovery(vshard_router, replicaset, 'call',
+    local res, err = call_with_retry_and_recovery(vshard_router, replicaset, replicaset_id, 'call',
         func_name, func_args, {timeout = timeout}, false)
     if err ~= nil then
         return nil, wrap_vshard_err(vshard_router, err, func_name, replicaset_id)
diff --git a/crud/common/storage_call.lua b/crud/common/storage_call.lua
new file mode 100644
index 00000000..1822b78f
--- /dev/null
+++ b/crud/common/storage_call.lua
@@ -0,0 +1,150 @@
+local fiber = require('fiber')
+local log = require('log')
+
+local utils = require('crud.common.utils')
+
+local CALL_ON_STORAGE_FUNC_NAME = 'call_on_storage'
+local CRUD_CALL_ON_STORAGE_FUNC_NAME = utils.get_storage_call(CALL_ON_STORAGE_FUNC_NAME)
+local LEGACY_RECHECK_INTERVAL = 60
+
+local storage_call = {}
+local legacy_call_on_storage_cache = {}
+
+local function cache_key(replicaset_id)
+    if replicaset_id == nil then
+        return nil
+    end
+
+    return tostring(replicaset_id)
+end
+
+local function cache_entry(replicaset_id)
+    local key = cache_key(replicaset_id)
+    if key == nil then
+        return nil
+    end
+
+    local entry = legacy_call_on_storage_cache[key]
+    if entry == nil then
+        return nil
+    end
+
+    if fiber.clock() >= entry.recheck_at then
+        legacy_call_on_storage_cache[key] = nil
+        return nil
+    end
+
+    return entry
+end
+
+local function is_legacy_call_on_storage(replicaset_id)
+    return cache_entry(replicaset_id) ~= nil
+end
+
+local function mark_legacy_call_on_storage(replicaset_id)
+    local key = cache_key(replicaset_id)
+    if key == nil then
+        return
+    end
+
+    local now = fiber.clock()
+    local entry = legacy_call_on_storage_cache[key]
+    if entry == nil or now >= entry.recheck_at then
+        log.warn(
+            "CRUD storage replicaset %q does not support %q; " ..
+            "falling back to direct storage calls until storage is upgraded",
+            key, CRUD_CALL_ON_STORAGE_FUNC_NAME
+        )
+    end
+
+    legacy_call_on_storage_cache[key] = {
+        recheck_at = now + LEGACY_RECHECK_INTERVAL,
+    }
+end
+
+local function is_call_on_storage_unsupported_error(err)
+    if err == nil or type(err.message) ~= 'string' then
+        return false
+    end
+
+    local not_defined = ("Procedure '%s' is not defined"):format(CRUD_CALL_ON_STORAGE_FUNC_NAME)
+    local access_denied = ("Execute access to function '%s' is denied"):format(CRUD_CALL_ON_STORAGE_FUNC_NAME)
+
+    return err.message == not_defined or err.message:startswith(access_denied)
+end
+
+--- Storage API function name used by new routers to call a function on storage
+--- under the original effective user.
+storage_call.CALL_ON_STORAGE_FUNC_NAME = CALL_ON_STORAGE_FUNC_NAME
+
+--- Clear the legacy marker for the replicaset.
+-- Called after _crud.call_on_storage succeeds, so the router keeps using
+-- the new storage wrapper for subsequent calls.
+-- @param ?string|number replicaset_id Vshard replicaset identifier.
+function storage_call.mark_call_on_storage_supported(replicaset_id)
+    local key = cache_key(replicaset_id)
+    if key ~= nil then
+        legacy_call_on_storage_cache[key] = nil
+    end
+end
+
+--- Decide whether the call should be retried using the legacy direct path.
+-- If fallback is required, the replicaset is marked as legacy for a short
+-- interval to avoid probing _crud.call_on_storage on every request.
+-- @param ?string|number replicaset_id Vshard replicaset identifier.
+-- @param ?table err Error returned by vshard/net.box.
+-- @param boolean used_legacy_call True when the failed call already used
+--        the direct legacy path.
+-- @return boolean
+function storage_call.should_fallback_to_legacy(replicaset_id, err, used_legacy_call)
+    if used_legacy_call then
+        return false
+    end
+
+    if not is_call_on_storage_unsupported_error(err) then
+        return false
+    end
+
+    mark_legacy_call_on_storage(replicaset_id)
+    return true
+end
+
+--- Extract a storage-side error from an async vshard result.
+-- Async calls return transport errors separately, but a storage function may
+-- still return nil, err as the function result. This helper returns that err.
+-- @param ?table result Result returned by future:wait_result().
+-- @return ?table
+function storage_call.result_error(result)
+    if type(result) ~= 'table' then
+        return nil
+    end
+
+    if result[1] ~= nil then
+        return nil
+    end
+
+    return result[2]
+end
+
+--- Prepare function name and arguments for a storage call.
+-- By default the call is routed through _crud.call_on_storage, which restores
+-- the original effective user on storage. If a legacy marker or force_legacy
+-- is set, the function is called directly for compatibility with pre-1.6
+-- storages.
+-- @param ?string|number replicaset_id Vshard replicaset identifier.
+-- @param string func_name Storage function name to call.
+-- @param ?table func_args Storage function arguments.
+-- @param ?boolean force_legacy Force direct legacy call.
+-- @return string Function name to pass to vshard.
+-- @return ?table Arguments to pass to vshard.
+-- @return boolean True when the direct legacy path is used.
+function storage_call.prepare(replicaset_id, func_name, func_args, force_legacy)
+    if force_legacy or is_legacy_call_on_storage(replicaset_id) then
+        return func_name, func_args, true
+    end
+
+    local func_args_ext = utils.append_array({box.session.effective_user(), func_name}, func_args)
+    return CRUD_CALL_ON_STORAGE_FUNC_NAME, func_args_ext, false
+end
+
+return storage_call
diff --git a/crud/select/merger.lua b/crud/select/merger.lua
index 1fd75320..f90f18cc 100644
--- a/crud/select/merger.lua
+++ b/crud/select/merger.lua
@@ -6,6 +6,7 @@ local call = require('crud.common.call')
 local fiber = require('fiber')
 local sharding = require('crud.common.sharding')
 local sharding_metadata_module = require('crud.common.sharding.sharding_metadata')
+local storage_call = require('crud.common.storage_call')
 
 local compat = require('crud.common.compat')
 local merger_lib = compat.require('tuple.merger', 'merger')
@@ -58,6 +59,14 @@ end
 
 local data = ffi.new('const unsigned char *[1]')
 
+local function make_net_box_opts(buf)
+    return {
+        is_async = true,
+        buffer = buf,
+        skip_header = utils.tarantool_supports_netbox_skip_header_option() or nil,
+    }
+end
+
 local function decode_response_array_header()
     local c = data[0][0]
     data[0] = data[0] + 1
@@ -101,14 +110,53 @@ local function decode_metainfo(buf)
     return res, err
 end
 
+local function wrap_storage_call_error(context, err)
+    return errors.wrap(utils.update_storage_call_error_description(
+        err, context.func_name, context.replicaset_id
+    ))
+end
+
+local function call_storage_on_conn(conn, replicaset_id, func_name, func_args, net_box_opts, force_legacy)
+    local call_func_name, call_args, used_legacy_call = storage_call.prepare(
+        replicaset_id, func_name, func_args, force_legacy
+    )
+
+    return conn:call(call_func_name, call_args, net_box_opts), nil, used_legacy_call
+end
+
+local function request_chunk(context, func_args, force_legacy)
+    if context.readview then
+        return call_storage_on_conn(
+            context.future_replica.conn,
+            context.replicaset_id,
+            context.func_name,
+            func_args,
+            context.net_box_opts,
+            force_legacy
+        )
+    end
+
+    return call.storage_call(
+        context.replicaset,
+        context.vshard_call_name,
+        context.replicaset_id,
+        context.func_name,
+        func_args,
+        context.net_box_opts,
+        force_legacy
+    )
+end
+
+local function reset_chunk_buffer(context)
+    local buf = buffer.ibuf()
+    context.buffer = buf
+    context.net_box_opts = make_net_box_opts(buf)
+end
+
 --- Wait for a data chunk and request for the next data chunk.
 local function fetch_chunk(context, state)
-    local net_box_opts = context.net_box_opts
     local buf = context.buffer
-    local func_name = context.func_name
     local func_args = context.func_args
-    local replicaset = context.replicaset
-    local vshard_call_name = context.vshard_call_name
     local timeout = context.timeout or call.DEFAULT_VSHARD_CALL_TIMEOUT
     local space_name = context.space_name
     local vshard_router = context.vshard_router
@@ -121,9 +169,25 @@ local function fetch_chunk(context, state)
 
     -- Wait for requested data.
     local res, err = future:wait_result(timeout)
+    if err == nil and not state.used_legacy_call then
+        storage_call.mark_call_on_storage_supported(context.replicaset_id)
+    end
+
+    if res == nil and storage_call.should_fallback_to_legacy(
+        context.replicaset_id, err, state.used_legacy_call
+    ) then
+        reset_chunk_buffer(context)
+        buf = context.buffer
+
+        future, err, state.used_legacy_call = request_chunk(context, func_args, true)
+        if err == nil then
+            state.future = future
+            res, err = future:wait_result(timeout)
+        end
+    end
+
     if res == nil then
-        local wrapped_err = errors.wrap(utils.update_storage_call_error_description(err, func_name, replicaset.uuid))
-        error(wrapped_err)
+        error(wrap_storage_call_error(context, err))
     end
 
     -- Decode metainfo, leave data to be processed by the merger.
@@ -171,16 +235,16 @@ local function fetch_chunk(context, state)
 
     -- change context.func_args too, but it does not matter
     next_func_args[4].after_tuple = cursor.after_tuple
-    local func_args_ext = utils.append_array({ box.session.effective_user(), func_name }, next_func_args)
 
-    if context.readview then
-        next_state = {future = context.future_replica.conn:call("_crud.call_on_storage",
-                func_args_ext, net_box_opts)}
-    else
-        local next_future = replicaset[vshard_call_name](replicaset, "_crud.call_on_storage",
-                func_args_ext, net_box_opts)
-        next_state = {future = next_future}
+    local next_future, call_err, used_legacy_call = request_chunk(context, next_func_args)
+    if call_err ~= nil then
+        error(wrap_storage_call_error(context, call_err))
     end
+    next_state = {
+        future = next_future,
+        used_legacy_call = used_legacy_call,
+    }
+
     return next_state, buf
 end
 
@@ -198,14 +262,10 @@ local function new(vshard_router, replicasets, space, index_id, func_name, func_
 
     -- Request a first data chunk and create merger sources.
     local merger_sources = {}
-    for _, replicaset in pairs(replicasets) do
+    for replicaset_id, replicaset in pairs(replicasets) do
         -- Perform a request.
         local buf = buffer.ibuf()
-        local net_box_opts = {is_async = true, buffer = buf,
-                              skip_header = utils.tarantool_supports_netbox_skip_header_option() or nil}
-        local func_args_ext = utils.append_array({ box.session.effective_user(), func_name }, func_args)
-        local future = replicaset[vshard_call_name](replicaset, "_crud.call_on_storage",
-                func_args_ext, net_box_opts)
+        local net_box_opts = make_net_box_opts(buf)
 
         -- Create a source.
         local context = {
@@ -214,6 +274,7 @@ local function new(vshard_router, replicasets, space, index_id, func_name, func_
             func_name = func_name,
             func_args = func_args,
             replicaset = replicaset,
+            replicaset_id = replicaset_id,
             vshard_call_name = vshard_call_name,
             timeout = call_opts.timeout,
             fetch_latest_metadata = call_opts.fetch_latest_metadata,
@@ -222,7 +283,15 @@ local function new(vshard_router, replicasets, space, index_id, func_name, func_
             readview = false,
         }
 
-        local state = {future = future}
+        local future, call_err, used_legacy_call = request_chunk(context, func_args)
+        if call_err ~= nil then
+            error(wrap_storage_call_error(context, call_err))
+        end
+
+        local state = {
+            future = future,
+            used_legacy_call = used_legacy_call,
+        }
         local source = merger_lib.new_buffer_source(fetch_chunk, context, state)
         table.insert(merger_sources, source)
     end
@@ -276,11 +345,8 @@ local function new_readview(vshard_router, replicasets, readview_info, space, in
 
             -- Perform a request.
             local buf = buffer.ibuf()
-            local net_box_opts = {is_async = true, buffer = buf,
-                                  skip_header = utils.tarantool_supports_netbox_skip_header_option() or nil}
+            local net_box_opts = make_net_box_opts(buf)
             func_args[4].readview_id = replicaset_info.id
-            local func_args_ext = utils.append_array({ box.session.effective_user(), func_name }, func_args)
-            local future = replica.conn:call("_crud.call_on_storage", func_args_ext, net_box_opts)
 
             -- Create a source.
             local context = {
@@ -289,6 +355,7 @@ local function new_readview(vshard_router, replicasets, readview_info, space, in
                 func_name = func_name,
                 func_args = func_args,
                 replicaset = replicaset,
+                replicaset_id = replicaset_id,
                 vshard_call_name = nil,
                 timeout = call_opts.timeout,
                 fetch_latest_metadata = call_opts.fetch_latest_metadata,
@@ -297,7 +364,16 @@ local function new_readview(vshard_router, replicasets, readview_info, space, in
                 readview = true,
                 future_replica = replica
             }
-            local state = {future = future}
+
+            local future, call_err, used_legacy_call = request_chunk(context, func_args)
+            if call_err ~= nil then
+                error(wrap_storage_call_error(context, call_err))
+            end
+
+            local state = {
+                future = future,
+                used_legacy_call = used_legacy_call,
+            }
             local source = merger_lib.new_buffer_source(fetch_chunk, context, state)
             table.insert(merger_sources, source)
 
diff --git a/test/helper.lua b/test/helper.lua
index 409f9c52..8fda8df8 100644
--- a/test/helper.lua
+++ b/test/helper.lua
@@ -425,6 +425,56 @@ function helpers.call_on_storages(cluster, func, ...)
     end
 end
 
+function helpers.reset_storage_call_compat_cache(router)
+    router.net_box:eval([[
+        local vshard = require('vshard')
+        local storage_call = require('crud.common.storage_call')
+
+        local replicasets = vshard.router.static:routeall()
+        for replicaset_id in pairs(replicasets) do
+            storage_call.mark_call_on_storage_supported(replicaset_id)
+        end
+    ]])
+end
+
+function helpers.disable_call_on_storage(cluster, router)
+    helpers.call_on_storages(cluster, function(server)
+        server.net_box:eval([[
+            if rawget(_G, '_crud') ~= nil then
+                rawset(_G._crud, 'call_on_storage', nil)
+            end
+
+            if not box.info.ro and box.func['_crud.call_on_storage'] ~= nil then
+                box.schema.func.drop('_crud.call_on_storage')
+            end
+        ]])
+    end)
+
+    helpers.reset_storage_call_compat_cache(router)
+end
+
+function helpers.restore_call_on_storage(cluster, router)
+    helpers.call_on_storages(cluster, function(server)
+        server.net_box:eval([[
+            require('crud.storage').init({async = false})
+        ]])
+    end)
+
+    helpers.reset_storage_call_compat_cache(router)
+end
+
+function helpers.without_call_on_storage(g, func)
+    helpers.disable_call_on_storage(g.cluster, g.router)
+
+    local ok, err = pcall(func)
+
+    helpers.restore_call_on_storage(g.cluster, g.router)
+
+    if not ok then
+        error(err, 0)
+    end
+end
+
 function helpers.assert_ge(actual, expected, message)
     if not (actual >= expected) then
         local err = string.format('expected: %s >= %s', actual, expected)
diff --git a/test/integration/legacy_storage_call_test.lua b/test/integration/legacy_storage_call_test.lua
new file mode 100644
index 00000000..204bef1b
--- /dev/null
+++ b/test/integration/legacy_storage_call_test.lua
@@ -0,0 +1,300 @@
+local t = require('luatest')
+local crud = require('crud')
+
+local helpers = require('test.helper')
+
+local pgroup = t.group('legacy_storage_call', helpers.backend_matrix({
+    {engine = 'memtx'},
+}, {skip_safe_mode = true}))
+
+pgroup.before_all(function(g)
+    helpers.start_default_cluster(g, 'srv_simple_operations')
+end)
+
+pgroup.after_all(function(g)
+    helpers.stop_cluster(g.cluster, g.params.backend)
+end)
+
+pgroup.before_each(function(g)
+    helpers.truncate_space_on_cluster(g.cluster, 'customers')
+end)
+
+local function disable_storage_yield_checks(cluster)
+    helpers.call_on_storages(cluster, function(server)
+        server.net_box:eval([[
+            local yield_checks = require('crud.common.yield_checks')
+            rawset(_G, '__legacy_storage_call_yield_checks_backup', {
+                check_no_yields = yield_checks.check_no_yields,
+                guard = yield_checks.guard,
+            })
+
+            yield_checks.check_no_yields = function() end
+            yield_checks.guard = function(f, ...)
+                return f(...)
+            end
+        ]])
+    end)
+end
+
+local function restore_storage_yield_checks(cluster)
+    helpers.call_on_storages(cluster, function(server)
+        server.net_box:eval([[
+            local backup = rawget(_G, '__legacy_storage_call_yield_checks_backup')
+            if backup == nil then
+                return
+            end
+
+            local yield_checks = require('crud.common.yield_checks')
+            yield_checks.check_no_yields = backup.check_no_yields
+            yield_checks.guard = backup.guard
+            rawset(_G, '__legacy_storage_call_yield_checks_backup', nil)
+        ]])
+    end)
+end
+
+-- Simulate a pre-_crud.call_on_storage storage. Such storages accepted direct
+-- calls to storage-side CRUD functions and did not run the newer yield-check
+-- guard that is normally installed by _crud.call_on_storage.
+local function without_legacy_storage_api(g, func)
+    helpers.disable_call_on_storage(g.cluster, g.router)
+    disable_storage_yield_checks(g.cluster)
+
+    local ok, err = pcall(func)
+
+    restore_storage_yield_checks(g.cluster)
+    helpers.restore_call_on_storage(g.cluster, g.router)
+
+    if not ok then
+        error(err, 0)
+    end
+end
+
+local function call_legacy(g, func_name, args)
+    helpers.reset_storage_call_compat_cache(g.router)
+
+    local result, err = g.router:call(func_name, args)
+    t.assert_equals(err, nil)
+
+    return result
+end
+
+local function assert_customers(result, expected)
+    local actual = {}
+    for _, customer in ipairs(crud.unflatten_rows(result.rows, result.metadata)) do
+        customer.bucket_id = nil
+        actual[customer.id] = customer
+    end
+
+    local expected_by_id = {}
+    for _, customer in ipairs(expected) do
+        expected_by_id[customer.id] = customer
+    end
+
+    t.assert_equals(actual, expected_by_id)
+end
+
+local function assert_customer(g, id, expected)
+    local result = call_legacy(g, 'crud.get', {
+        'customers', id, {mode = 'write'},
+    })
+
+    if expected == nil then
+        t.assert_equals(result.rows, {})
+    else
+        assert_customers(result, {expected})
+    end
+end
+
+pgroup.test_single_tuple_operations_work_without_call_on_storage = function(g)
+    without_legacy_storage_api(g, function()
+        local result = call_legacy(g, 'crud.insert', {
+            'customers', {101, box.NULL, 'Legacy Insert', 20},
+        })
+        assert_customers(result, {
+            {id = 101, name = 'Legacy Insert', age = 20},
+        })
+
+        assert_customer(g, 101, {
+            id = 101, name = 'Legacy Insert', age = 20,
+        })
+
+        result = call_legacy(g, 'crud.update', {
+            'customers', 101, {
+                {'=', 'name', 'Legacy Update'},
+                {'+', 'age', 1},
+            },
+        })
+        assert_customers(result, {
+            {id = 101, name = 'Legacy Update', age = 21},
+        })
+
+        result = call_legacy(g, 'crud.replace', {
+            'customers', {101, box.NULL, 'Legacy Replace', 30},
+        })
+        assert_customers(result, {
+            {id = 101, name = 'Legacy Replace', age = 30},
+        })
+
+        result = call_legacy(g, 'crud.upsert', {
+            'customers', {101, box.NULL, 'Legacy Upsert Ignored', 1}, {
+                {'=', 'name', 'Legacy Upsert'},
+                {'=', 'age', 31},
+            },
+        })
+        t.assert_equals(result.rows, {})
+        assert_customer(g, 101, {
+            id = 101, name = 'Legacy Upsert', age = 31,
+        })
+
+        result = call_legacy(g, 'crud.delete', {'customers', 101})
+        assert_customers(result, {
+            {id = 101, name = 'Legacy Upsert', age = 31},
+        })
+
+        assert_customer(g, 101, nil)
+    end)
+end
+
+pgroup.test_single_object_operations_work_without_call_on_storage = function(g)
+    without_legacy_storage_api(g, function()
+        local result = call_legacy(g, 'crud.insert_object', {
+            'customers', {id = 201, name = 'Legacy Object Insert', age = 20},
+        })
+        assert_customers(result, {
+            {id = 201, name = 'Legacy Object Insert', age = 20},
+        })
+
+        result = call_legacy(g, 'crud.replace_object', {
+            'customers', {id = 201, name = 'Legacy Object Replace', age = 30},
+        })
+        assert_customers(result, {
+            {id = 201, name = 'Legacy Object Replace', age = 30},
+        })
+
+        result = call_legacy(g, 'crud.upsert_object', {
+            'customers', {id = 201, name = 'Legacy Object Ignored', age = 1}, {
+                {'=', 'name', 'Legacy Object Upsert'},
+                {'=', 'age', 31},
+            },
+        })
+        t.assert_equals(result.rows, {})
+        assert_customer(g, 201, {
+            id = 201, name = 'Legacy Object Upsert', age = 31,
+        })
+    end)
+end
+
+pgroup.test_batch_tuple_operations_work_without_call_on_storage = function(g)
+    without_legacy_storage_api(g, function()
+        local result = call_legacy(g, 'crud.insert_many', {
+            'customers', {
+                {301, box.NULL, 'Legacy Insert Many 1', 20},
+                {302, box.NULL, 'Legacy Insert Many 2', 21},
+            },
+        })
+        assert_customers(result, {
+            {id = 301, name = 'Legacy Insert Many 1', age = 20},
+            {id = 302, name = 'Legacy Insert Many 2', age = 21},
+        })
+
+        result = call_legacy(g, 'crud.replace_many', {
+            'customers', {
+                {301, box.NULL, 'Legacy Replace Many 1', 30},
+                {302, box.NULL, 'Legacy Replace Many 2', 31},
+            },
+        })
+        assert_customers(result, {
+            {id = 301, name = 'Legacy Replace Many 1', age = 30},
+            {id = 302, name = 'Legacy Replace Many 2', age = 31},
+        })
+
+        result = call_legacy(g, 'crud.upsert_many', {
+            'customers', {
+                {
+                    {301, box.NULL, 'Legacy Upsert Many Ignored', 1},
+                    {{'=', 'name', 'Legacy Upsert Many 1'}, {'=', 'age', 32}},
+                },
+                {
+                    {303, box.NULL, 'Legacy Upsert Many 3', 33},
+                    {{'+', 'age', 1}},
+                },
+            },
+        })
+        t.assert_equals(result.rows, nil)
+        assert_customer(g, 301, {
+            id = 301, name = 'Legacy Upsert Many 1', age = 32,
+        })
+        assert_customer(g, 303, {
+            id = 303, name = 'Legacy Upsert Many 3', age = 33,
+        })
+    end)
+end
+
+pgroup.test_batch_object_operations_work_without_call_on_storage = function(g)
+    without_legacy_storage_api(g, function()
+        local result = call_legacy(g, 'crud.insert_object_many', {
+            'customers', {
+                {id = 401, name = 'Legacy Object Insert Many 1', age = 20},
+                {id = 402, name = 'Legacy Object Insert Many 2', age = 21},
+            },
+        })
+        assert_customers(result, {
+            {id = 401, name = 'Legacy Object Insert Many 1', age = 20},
+            {id = 402, name = 'Legacy Object Insert Many 2', age = 21},
+        })
+
+        result = call_legacy(g, 'crud.replace_object_many', {
+            'customers', {
+                {id = 401, name = 'Legacy Object Replace Many 1', age = 30},
+                {id = 402, name = 'Legacy Object Replace Many 2', age = 31},
+            },
+        })
+        assert_customers(result, {
+            {id = 401, name = 'Legacy Object Replace Many 1', age = 30},
+            {id = 402, name = 'Legacy Object Replace Many 2', age = 31},
+        })
+
+        result = call_legacy(g, 'crud.upsert_object_many', {
+            'customers', {
+                {
+                    {id = 401, name = 'Legacy Object Upsert Many Ignored', age = 1},
+                    {{'=', 'name', 'Legacy Object Upsert Many 1'}, {'=', 'age', 32}},
+                },
+                {
+                    {id = 403, name = 'Legacy Object Upsert Many 3', age = 33},
+                    {{'+', 'age', 1}},
+                },
+            },
+        })
+        t.assert_equals(result.rows, nil)
+        assert_customer(g, 401, {
+            id = 401, name = 'Legacy Object Upsert Many 1', age = 32,
+        })
+        assert_customer(g, 403, {
+            id = 403, name = 'Legacy Object Upsert Many 3', age = 33,
+        })
+    end)
+end
+
+pgroup.test_map_operations_work_without_call_on_storage = function(g)
+    without_legacy_storage_api(g, function()
+        call_legacy(g, 'crud.insert_many', {
+            'customers', {
+                {501, box.NULL, 'Legacy Map 1', 20},
+                {502, box.NULL, 'Legacy Map 2', 21},
+            },
+        })
+
+        local result = call_legacy(g, 'crud.len', {'customers'})
+        t.assert_equals(result, 2)
+
+        result = call_legacy(g, 'crud.count', {'customers'})
+        t.assert_equals(result, 2)
+
+        result = call_legacy(g, 'crud.truncate', {'customers'})
+        t.assert_equals(result, true)
+
+        result = call_legacy(g, 'crud.len', {'customers'})
+        t.assert_equals(result, 0)
+    end)
+end
diff --git a/test/integration/select_test.lua b/test/integration/select_test.lua
index 4edf91e5..7b6a5f1a 100644
--- a/test/integration/select_test.lua
+++ b/test/integration/select_test.lua
@@ -246,6 +246,38 @@ pgroup.test_select_all = function(g)
     t.assert_equals(#objects, 0)
 end
 
+pgroup.test_select_works_with_legacy_storage_without_call_on_storage = function(g)
+    helpers.insert_objects(g, 'customers', {
+        {
+            id = 1, name = "Elizabeth", last_name = "Jackson",
+            age = 12, city = "New York",
+        }, {
+            id = 2, name = "Mary", last_name = "Brown",
+            age = 46, city = "Los Angeles",
+        }, {
+            id = 3, name = "David", last_name = "Smith",
+            age = 33, city = "Los Angeles",
+        }, {
+            id = 4, name = "William", last_name = "White",
+            age = 81, city = "Chicago",
+        },
+    })
+
+    helpers.without_call_on_storage(g, function()
+        local result, err = g.router:call('crud.select', {
+            'customers', nil, {fullscan = true, mode = 'write'},
+        })
+
+        t.assert_equals(err, nil)
+        t.assert_equals(result.rows, {
+            {1, 477, "Elizabeth", "Jackson", 12, "New York"},
+            {2, 401, "Mary", "Brown", 46, "Los Angeles"},
+            {3, 2804, "David", "Smith", 33, "Los Angeles"},
+            {4, 1161, "William", "White", 81, "Chicago"},
+        })
+    end)
+end
+
 pgroup.test_select_all_with_first = function(g)
     local customers = helpers.insert_objects(g, 'customers', {
         {
diff --git a/test/unit/call_test.lua b/test/unit/call_test.lua
index 912fc692..009bc005 100644
--- a/test/unit/call_test.lua
+++ b/test/unit/call_test.lua
@@ -241,6 +241,65 @@ pgroup.test_any_vshard_call = function(g)
     t.assert_equals(err, nil)
 end
 
+pgroup.test_direct_storage_call_works_for_old_router_compat = function(g)
+    local results, err = g.router:eval([[
+        local vshard = require('vshard')
+        local replicasets = vshard.router.static:routeall()
+
+        local results = {}
+        for _, replicaset in pairs(replicasets) do
+            local result, err = replicaset:callrw('say_hi_politely', {'old-router'})
+            if err ~= nil then
+                return nil, err
+            end
+
+            table.insert(results, result)
+        end
+
+        return results
+    ]])
+
+    t.assert_equals(err, nil)
+    t.assert_equals(#results, 2)
+    t.assert_items_include(results, {'HI, old-router! I am 1', 'HI, old-router! I am 1'})
+end
+
+pgroup.test_legacy_storage_without_call_on_storage = function(g)
+    helpers.without_call_on_storage(g, function()
+        helpers.reset_storage_call_compat_cache(g.router)
+        local results_map, map_err = g.router:eval([[
+            local vshard = require('vshard')
+            local call = require('crud.common.call')
+
+            return call.map(vshard.router.static, 'say_hi_politely', {'legacy-map'}, {mode = 'write'})
+        ]])
+        t.assert_equals(map_err, nil)
+        local results = helpers.get_results_list(results_map)
+        t.assert_equals(#results, 2)
+        t.assert_items_include(results, {{"HI, legacy-map! I am 1"}, {"HI, legacy-map! I am 1"}})
+
+        helpers.reset_storage_call_compat_cache(g.router)
+        local single_result, single_err = g.router:eval([[
+            local vshard = require('vshard')
+            local call = require('crud.common.call')
+
+            return call.single(vshard.router.static, 1, 'say_hi_politely', {'legacy-single'}, {mode = 'write'})
+        ]])
+        t.assert_equals(single_err, nil)
+        t.assert_equals(single_result, 'HI, legacy-single! I am 1')
+
+        helpers.reset_storage_call_compat_cache(g.router)
+        local any_result, any_err = g.router:eval([[
+            local vshard = require('vshard')
+            local call = require('crud.common.call')
+
+            return call.any(vshard.router.static, 'say_hi_politely', {'legacy-any'}, {})
+        ]])
+        t.assert_equals(any_err, nil)
+        t.assert_equals(any_result, 'HI, legacy-any! I am 1')
+    end)
+end
+
 pgroup.test_any_vshard_call_timeout = function(g)
     helpers.call_on_storages(g.cluster, function(server)
         server.net_box:eval([[