anchor.diff:      erlang:cancel_timer(Ref),
       schedule_terminate(State#state{terminate_timer_ref = undefined}).
   
  +%%--------------------------------------------------------------------
  +%% @private
  +%% @doc
  +%% Gets tp routing key.
  +%% @end
  +%%--------------------------------------------------------------------
  +-spec get_key(datastore:key(), atom()) -> tp_key().
   get_key(Key, Type) ->
       case application:get_env(?CLUSTER_WORKER_APP_NAME, aggregate_tp, false) of
           true ->
  
positive.code: %% {@link wait/1} to collect result.
  %% @end
  %%--------------------------------------------------------------------
  -spec call_async(ctx(), {doc | links, key()}, atom(), list()) ->
      {ok, reference()} | {error, term()}.
  call_async(Ctx, Key, Function, Args) ->
      Timeout = application:get_env(?CLUSTER_WORKER_APP_NAME,
  
negative.code: %% with a key.
  %% @end
  %%--------------------------------------------------------------------
  -spec call(ctx(), tp_key(), atom(), list()) -> term().
  call(Ctx, Key, Function, Args) ->
      case call_async(Ctx, Key, Function, Args) of
          {ok, Ref} -> wait(Ref);
  