<commit_before>%% ---------------------------------------------------------------------
%%
%% Copyright (c) 2007-2013 Basho Technologies, Inc.  All Rights Reserved.
%%
%% This file is provided to you under the Apache License,
%% Version 2.0 (the "License"); you may not use this file
%% except in compliance with the License.  You may obtain
%% a copy of the License at
%%
%%   http://www.apache.org/licenses/LICENSE-2.0
%%
%% Unless required by applicable law or agreed to in writing,
%% software distributed under the License is distributed on an
%% "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
%% KIND, either express or implied.  See the License for the
%% specific language governing permissions and limitations
%% under the License.
%%
%% ---------------------------------------------------------------------

-module(riak_cs_block_server).

-behaviour(gen_server).

-ifdef(PULSE).
-include_lib("pulse/include/pulse.hrl").
-compile({parse_transform, pulse_instrument}).
-compile({pulse_replace_module,[{gen_server,pulse_gen_server}]}).
-endif.

-include("riak_cs.hrl").
-include_lib("riak_pb/include/riak_pb_kv_codec.hrl").

%% API
-export([start_link/1, start_link/2,
         start_block_servers/3,
         get_block/5, get_block/6,
         put_block/6,
         delete_block/5,
         maybe_stop_block_servers/1,
         stop/1]).

%% gen_server callbacks
-export([init/1,
         handle_call/3,
         handle_cast/2,
         handle_info/2,
         terminate/2,
         code_change/3]).

-record(state, {riak_client :: riak_client(),
                close_riak_connection=true :: boolean()}).

%%%===================================================================
%%% API
%%%===================================================================

%%--------------------------------------------------------------------
%% @doc
%% Starts the server
%%
%% @spec start_link(lfs_manifest()) -> {ok, Pid} | ignore | {error, Error}
%% @spec start_link(lfs_manifest(), riak_client()) -> {ok, Pid} | ignore | {error, Error}
%% @end
%%--------------------------------------------------------------------
start_link(Manifest) ->
    gen_server:start_link(?MODULE, [Manifest], []).

start_link(Manifest, RcPid) ->
    gen_server:start_link(?MODULE, [Manifest, RcPid], []).

%% @doc Start (up to) 'MaxNumServers'
%% riak_cs_block_server procs.
%% 'RcPid' must be a Pid you already
%% have for a riakc_pb_socket proc. If the
%% poolboy boy returns full, you will be given
%% a list of less than 'MaxNumServers'.

%% TODO: this doesn't guarantee any minimum
%% number of workers. I could also imagine
%% this function looking something
%% like:
%% start_block_servers(RcPid, MinWorkers, MaxWorkers, MinWorkerTimeout)
%% Where the function works something like:
%% Give me between MinWorkers and MaxWorkers,
%% waiting up to MinWorkerTimeout to get at least
%% MinWorkers. If the timeout occurs, this function
%% could return an error, or the pids it has
%% so far (which might be less than MinWorkers).
-spec start_block_servers(lfs_manifest(), riak_client(), pos_integer()) -> [pid()].
start_block_servers(Manifest, RcPid, MaxNumServers) ->
    start_block_servers(Manifest, RcPid, MaxNumServers, []).

start_block_servers(Manifest, RcPid, 1, BlockServers) ->
    {ok, BlockServer} = start_link(Manifest, RcPid),
    [BlockServer | BlockServers];
start_block_servers(Manifest, RcPid, NumWorkers, BlockServers) ->
    case start_link(Manifest) of
        {ok, BlockServer} ->
            start_block_servers(Manifest, RcPid, NumWorkers - 1, [BlockServer | BlockServers]);
        {error, normal} ->
            start_block_servers(Manifest, RcPid, 1, BlockServers)
    end.

-spec get_block(pid(), binary(), binary(), binary(), pos_integer()) -> ok.
get_block(Pid, Bucket, Key, UUID, BlockNumber) ->
    gen_server:cast(Pid, {get_block, self(), Bucket, Key, undefined, UUID, BlockNumber}).

%% @doc get a block which is know to have originated on cluster ClusterID.
%% If it's not found locally, it might get returned from the replication
%% cluster if a connection exists to that cluster. This is proxy-get().
-spec get_block(pid(), binary(), binary(), binary(), binary(), pos_integer()) -> ok.
get_block(Pid, Bucket, Key, ClusterID, UUID, BlockNumber) ->
    gen_server:cast(Pid, {get_block, self(), Bucket, Key, ClusterID, UUID, BlockNumber}).

-spec put_block(pid(), binary(), binary(), binary(), pos_integer(), binary()) -> ok.
put_block(Pid, Bucket, Key, UUID, BlockNumber, Value) ->
    gen_server:cast(Pid, {put_block, self(), Bucket, Key, UUID, BlockNumber,
                          Value, riak_cs_utils:md5(Value)}).

-spec delete_block(pid(), binary(), binary(), binary(), pos_integer()) -> ok.
delete_block(Pid, Bucket, Key, UUID, BlockNumber) ->
    gen_server:cast(Pid, {delete_block, self(), Bucket, Key, UUID, BlockNumber}).

-spec maybe_stop_block_servers(undefined | [pid()]) -> ok.
maybe_stop_block_servers(undefined) ->
    ok;
maybe_stop_block_servers(BlockServerPids) ->
    _ = [stop(P) || P <- BlockServerPids],
    ok.

stop(Pid) ->
    gen_server:call(Pid, stop, infinity).

%%%===================================================================
%%% gen_server callbacks
%%%===================================================================

init([Manifest]) ->
    case riak_cs_riak_client:checkout(request_pool) of
        {ok, RcPid} ->
            init(Manifest, RcPid, #state{close_riak_connection=true});
        {error, _Reason} ->
            {stop, normal}
    end;
init([Manifest, RcPid]) ->
    init(Manifest, RcPid, #state{close_riak_connection=false}).

init(Manifest, RcPid, State) ->
    process_flag(trap_exit, true),
    ok = riak_cs_riak_client:set_manifest(RcPid, Manifest),
    {ok, State#state{riak_client=RcPid}}.

%%--------------------------------------------------------------------
%% @private
%% @doc
%% Handling call messages
%%
%% @spec handle_call(Request, From, State) ->
%%                                   {reply, Reply, State} |
%%                                   {reply, Reply, State, Timeout} |
%%                                   {noreply, State} |
%%                                   {noreply, State, Timeout} |
%%                                   {stop, Reason, Reply, State} |
%%                                   {stop, Reason, State}
%% @end
%%--------------------------------------------------------------------
handle_call(stop, _From, State) ->
    {stop, normal, ok, State}.

%%--------------------------------------------------------------------
%% @private
%% @doc
%% Handling cast messages
%%
%% @spec handle_cast(Msg, State) -> {noreply, State} |
%%                                  {noreply, State, Timeout} |
%%                                  {stop, Reason, State}
%% @end
%%--------------------------------------------------------------------

handle_cast({get_block, ReplyPid, Bucket, Key, ClusterID, UUID, BlockNumber},
            State=#state{riak_client=RcPid}) ->
    get_block(ReplyPid, Bucket, Key, ClusterID, UUID, BlockNumber, RcPid),
    {noreply, State};
handle_cast({put_block, ReplyPid, Bucket, Key, UUID, BlockNumber, Value, BCSum},
            State=#state{riak_client=RcPid}) ->
    dt_entry(<<"put_block">>, [BlockNumber], [Bucket, Key]),
    {FullBucket, FullKey} = full_bkey(Bucket, Key, UUID, BlockNumber),
    MD = make_md_usermeta([{?USERMETA_BUCKET, Bucket},
                           {?USERMETA_KEY, Key},
                           {?USERMETA_BCSUM, BCSum}]),
    FailFun = fun(Error) ->
                      _ = lager:error("Put ~p ~p UUID ~p block ~p failed: ~p\n",
                                      [Bucket, Key, UUID, BlockNumber, Error])
              end,
    %% TODO: Handle put failure here.
    ok = do_put_block(FullBucket, FullKey, <<>>, Value, MD, RcPid, FailFun),
    riak_cs_put_fsm:block_written(ReplyPid, BlockNumber),
    dt_return(<<"put_block">>, [BlockNumber], [Bucket, Key]),
    {noreply, State};
handle_cast({delete_block, ReplyPid, Bucket, Key, UUID, BlockNumber}, State=#state{riak_client=RcPid}) ->
    dt_entry(<<"delete_block">>, [BlockNumber], [Bucket, Key]),
    {FullBucket, FullKey} = full_bkey(Bucket, Key, UUID, BlockNumber),
    StartTime = os:timestamp(),

    %% do a get first to get the vclock (only do a head request though)
    GetOptions = [{r, 1}, {notfound_ok, false}, {basic_quorum, false}, head],
    _ = case riakc_pb_socket:get(block_pbc(RcPid), FullBucket, FullKey, GetOptions) of
            {ok, RiakObject} ->
                ok = delete_block(RcPid, ReplyPid, RiakObject, {UUID, BlockNumber});
        {error, notfound} ->
            %% If the block isn't found, assume it's been
            %% previously deleted by another delete FSM, and
            %% move on to the next block.
            riak_cs_delete_fsm:block_deleted(ReplyPid, {ok, {UUID, BlockNumber}})
    end,
    ok = riak_cs_stats:update_with_start(block_delete, StartTime),
    dt_return(<<"delete_block">>, [BlockNumber], [Bucket, Key]),
    {noreply, State};
handle_cast(_Msg, State) ->
    {noreply, State}.

get_block(ReplyPid, Bucket, Key, ClusterID, UUID, BlockNumber, RcPid) ->
    %% don't use proxy get if it's a local get
    %% or proxy get is disabled
    ProxyActive = riak_cs_config:proxy_get_active(),
    UseProxyGet = use_proxy_get(RcPid, ClusterID),

    case riak_cs_utils:n_val_1_get_requests() of
        true ->
            do_get_block(ReplyPid, Bucket, Key, ClusterID, UseProxyGet, ProxyActive, UUID,
                         BlockNumber, RcPid);
        false ->
            normal_nval_block_get(ReplyPid, Bucket, Key, ClusterID,
                                  UseProxyGet, UUID, BlockNumber, RcPid)
    end.

do_get_block(ReplyPid, Bucket, Key, ClusterID, UseProxyGet, ProxyActive,
             UUID, BlockNumber, RcPid) ->
    MaxRetries = riak_cs_config:get_env(riak_cs, block_get_max_retries, 5),
    do_get_block(ReplyPid, Bucket, Key, ClusterID, UseProxyGet, ProxyActive,
                 UUID, BlockNumber, RcPid, MaxRetries, 0).

do_get_block(ReplyPid, _Bucket, _Key, _ClusterID, _UseProxyGet, _ProxyActive,
             UUID, BlockNumber, _RcPid, MaxRetries, NumRetries)
  when is_atom(NumRetries) orelse NumRetries > MaxRetries ->
    Sorry = {error, notfound},
    ok = riak_cs_get_fsm:chunk(ReplyPid, {UUID, BlockNumber}, Sorry);
do_get_block(ReplyPid, Bucket, Key, ClusterID, UseProxyGet, ProxyActive,
             UUID, BlockNumber, RcPid, MaxRetries, NumRetries) ->
    ok = sleep_retries(NumRetries),

    dt_entry(<<"get_block">>, [BlockNumber], [Bucket, Key]),
    {FullBucket, FullKey} = full_bkey(Bucket, Key, UUID, BlockNumber),

    StartTime = os:timestamp(),
    GetOptions1 = n_val_one_options(),
    GetOptions2 = r_one_options(),

    ProceedFun = fun(OkReply) ->
            ok = riak_cs_stats:update_with_start(block_get_retry, StartTime),
            ok = riak_cs_get_fsm:chunk(ReplyPid, {UUID, BlockNumber}, OkReply),
            dt_return(<<"get_block">>, [BlockNumber], [Bucket, Key])
      end,
    RetryFun = fun(NewPause) ->
               ok = riak_cs_stats:update_with_start(block_get_retry, StartTime),
               do_get_block(ReplyPid, Bucket, Key, ClusterID, UseProxyGet,
                            ProxyActive, UUID, BlockNumber, RcPid, MaxRetries, NewPause)
            end,

    Timeout = timer:seconds(5),
    try_local_get(RcPid, FullBucket, FullKey, GetOptions1, GetOptions2,
                  Timeout, ProceedFun, RetryFun, NumRetries, UseProxyGet,
                  ProxyActive, ClusterID).

try_local_get(RcPid, FullBucket, FullKey, GetOptions1, GetOptions2,
              Timeout, ProceedFun, RetryFun, NumRetries, UseProxyGet,
              ProxyActive, ClusterID) ->
    case get_block_local(RcPid, FullBucket, FullKey, GetOptions1, Timeout) of
        {ok, _} = Success ->
            ProceedFun(Success);
        {error, {insufficient_vnodes,_,need,_}} ->
            RetryFun(NumRetries + 2);
        {error, Why} when Why == notfound;
                          Why == timeout;
                          Why == disconnected;
                          Why == <<"{insufficient_vnodes,0,need,1}">>;
                          Why == {insufficient_vnodes,0,need,1} ->
            handle_local_notfound(RcPid, FullBucket, FullKey, GetOptions2,
                                  ProceedFun, RetryFun, NumRetries, UseProxyGet,
                                  ProxyActive, ClusterID);
        {error, Other} ->
            _ = lager:error("do_get_block: other error 1: ~p\n", [Other]),
            RetryFun(failure)
    end.

handle_local_notfound(RcPid, FullBucket, FullKey, GetOptions2,
                      ProceedFun, RetryFun, NumRetries, UseProxyGet,
                      ProxyActive, ClusterID) ->
    %%% SLF TODO fix timeout
    case get_block_local(RcPid, FullBucket, FullKey, GetOptions2, 60*1000) of
        {ok, _} = Success ->
            ProceedFun(Success);
        {error, Why} when Why == notfound;
                          Why == timeout;
                          Why == disconnected ->
            case UseProxyGet of
                true when ProxyActive ->
                    case get_block_remote(RcPid, FullBucket, FullKey,
                                          ClusterID, GetOptions2) of
                        {ok, _} = Success ->
                            ProceedFun(Success);
                        {error, _} ->
                            if UseProxyGet ->
                                    RetryFun(NumRetries + 1);
                                true ->
                                    RetryFun(failure)
                            end
                    end;
                true when not ProxyActive -><commit_msg>Fix {error, disconnected} repainted with {error, notfound}<commit_after>%% ---------------------------------------------------------------------
%%
%% Copyright (c) 2007-2013 Basho Technologies, Inc.  All Rights Reserved.
%%
%% This file is provided to you under the Apache License,
%% Version 2.0 (the "License"); you may not use this file
%% except in compliance with the License.  You may obtain
%% a copy of the License at
%%
%%   http://www.apache.org/licenses/LICENSE-2.0
%%
%% Unless required by applicable law or agreed to in writing,
%% software distributed under the License is distributed on an
%% "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
%% KIND, either express or implied.  See the License for the
%% specific language governing permissions and limitations
%% under the License.
%%
%% ---------------------------------------------------------------------

-module(riak_cs_block_server).

-behaviour(gen_server).

-ifdef(PULSE).
-include_lib("pulse/include/pulse.hrl").
-compile({parse_transform, pulse_instrument}).
-compile({pulse_replace_module,[{gen_server,pulse_gen_server}]}).
-endif.

-include("riak_cs.hrl").
-include_lib("riak_pb/include/riak_pb_kv_codec.hrl").

%% API
-export([start_link/1, start_link/2,
         start_block_servers/3,
         get_block/5, get_block/6,
         put_block/6,
         delete_block/5,
         maybe_stop_block_servers/1,
         stop/1]).

%% gen_server callbacks
-export([init/1,
         handle_call/3,
         handle_cast/2,
         handle_info/2,
         terminate/2,
         code_change/3]).

-record(state, {riak_client :: riak_client(),
                close_riak_connection=true :: boolean()}).

%%%===================================================================
%%% API
%%%===================================================================

%%--------------------------------------------------------------------
%% @doc
%% Starts the server
%%
%% @spec start_link(lfs_manifest()) -> {ok, Pid} | ignore | {error, Error}
%% @spec start_link(lfs_manifest(), riak_client()) -> {ok, Pid} | ignore | {error, Error}
%% @end
%%--------------------------------------------------------------------
start_link(Manifest) ->
    gen_server:start_link(?MODULE, [Manifest], []).

start_link(Manifest, RcPid) ->
    gen_server:start_link(?MODULE, [Manifest, RcPid], []).

%% @doc Start (up to) 'MaxNumServers'
%% riak_cs_block_server procs.
%% 'RcPid' must be a Pid you already
%% have for a riakc_pb_socket proc. If the
%% poolboy boy returns full, you will be given
%% a list of less than 'MaxNumServers'.

%% TODO: this doesn't guarantee any minimum
%% number of workers. I could also imagine
%% this function looking something
%% like:
%% start_block_servers(RcPid, MinWorkers, MaxWorkers, MinWorkerTimeout)
%% Where the function works something like:
%% Give me between MinWorkers and MaxWorkers,
%% waiting up to MinWorkerTimeout to get at least
%% MinWorkers. If the timeout occurs, this function
%% could return an error, or the pids it has
%% so far (which might be less than MinWorkers).
-spec start_block_servers(lfs_manifest(), riak_client(), pos_integer()) -> [pid()].
start_block_servers(Manifest, RcPid, MaxNumServers) ->
    start_block_servers(Manifest, RcPid, MaxNumServers, []).

start_block_servers(Manifest, RcPid, 1, BlockServers) ->
    {ok, BlockServer} = start_link(Manifest, RcPid),
    [BlockServer | BlockServers];
start_block_servers(Manifest, RcPid, NumWorkers, BlockServers) ->
    case start_link(Manifest) of
        {ok, BlockServer} ->
            start_block_servers(Manifest, RcPid, NumWorkers - 1, [BlockServer | BlockServers]);
        {error, normal} ->
            start_block_servers(Manifest, RcPid, 1, BlockServers)
    end.

-spec get_block(pid(), binary(), binary(), binary(), pos_integer()) -> ok.
get_block(Pid, Bucket, Key, UUID, BlockNumber) ->
    gen_server:cast(Pid, {get_block, self(), Bucket, Key, undefined, UUID, BlockNumber}).

%% @doc get a block which is know to have originated on cluster ClusterID.
%% If it's not found locally, it might get returned from the replication
%% cluster if a connection exists to that cluster. This is proxy-get().
-spec get_block(pid(), binary(), binary(), binary(), binary(), pos_integer()) -> ok.
get_block(Pid, Bucket, Key, ClusterID, UUID, BlockNumber) ->
    gen_server:cast(Pid, {get_block, self(), Bucket, Key, ClusterID, UUID, BlockNumber}).

-spec put_block(pid(), binary(), binary(), binary(), pos_integer(), binary()) -> ok.
put_block(Pid, Bucket, Key, UUID, BlockNumber, Value) ->
    gen_server:cast(Pid, {put_block, self(), Bucket, Key, UUID, BlockNumber,
                          Value, riak_cs_utils:md5(Value)}).

-spec delete_block(pid(), binary(), binary(), binary(), pos_integer()) -> ok.
delete_block(Pid, Bucket, Key, UUID, BlockNumber) ->
    gen_server:cast(Pid, {delete_block, self(), Bucket, Key, UUID, BlockNumber}).

-spec maybe_stop_block_servers(undefined | [pid()]) -> ok.
maybe_stop_block_servers(undefined) ->
    ok;
maybe_stop_block_servers(BlockServerPids) ->
    _ = [stop(P) || P <- BlockServerPids],
    ok.

stop(Pid) ->
    gen_server:call(Pid, stop, infinity).

%%%===================================================================
%%% gen_server callbacks
%%%===================================================================

init([Manifest]) ->
    case riak_cs_riak_client:checkout(request_pool) of
        {ok, RcPid} ->
            init(Manifest, RcPid, #state{close_riak_connection=true});
        {error, _Reason} ->
            {stop, normal}
    end;
init([Manifest, RcPid]) ->
    init(Manifest, RcPid, #state{close_riak_connection=false}).

init(Manifest, RcPid, State) ->
    process_flag(trap_exit, true),
    ok = riak_cs_riak_client:set_manifest(RcPid, Manifest),
    {ok, State#state{riak_client=RcPid}}.

%%--------------------------------------------------------------------
%% @private
%% @doc
%% Handling call messages
%%
%% @spec handle_call(Request, From, State) ->
%%                                   {reply, Reply, State} |
%%                                   {reply, Reply, State, Timeout} |
%%                                   {noreply, State} |
%%                                   {noreply, State, Timeout} |
%%                                   {stop, Reason, Reply, State} |
%%                                   {stop, Reason, State}
%% @end
%%--------------------------------------------------------------------
handle_call(stop, _From, State) ->
    {stop, normal, ok, State}.

%%--------------------------------------------------------------------
%% @private
%% @doc
%% Handling cast messages
%%
%% @spec handle_cast(Msg, State) -> {noreply, State} |
%%                                  {noreply, State, Timeout} |
%%                                  {stop, Reason, State}
%% @end
%%--------------------------------------------------------------------

handle_cast({get_block, ReplyPid, Bucket, Key, ClusterID, UUID, BlockNumber},
            State=#state{riak_client=RcPid}) ->
    get_block(ReplyPid, Bucket, Key, ClusterID, UUID, BlockNumber, RcPid),
    {noreply, State};
handle_cast({put_block, ReplyPid, Bucket, Key, UUID, BlockNumber, Value, BCSum},
            State=#state{riak_client=RcPid}) ->
    dt_entry(<<"put_block">>, [BlockNumber], [Bucket, Key]),
    {FullBucket, FullKey} = full_bkey(Bucket, Key, UUID, BlockNumber),
    MD = make_md_usermeta([{?USERMETA_BUCKET, Bucket},
                           {?USERMETA_KEY, Key},
                           {?USERMETA_BCSUM, BCSum}]),
    FailFun = fun(Error) ->
                      _ = lager:error("Put ~p ~p UUID ~p block ~p failed: ~p\n",
                                      [Bucket, Key, UUID, BlockNumber, Error])
              end,
    %% TODO: Handle put failure here.
    ok = do_put_block(FullBucket, FullKey, <<>>, Value, MD, RcPid, FailFun),
    riak_cs_put_fsm:block_written(ReplyPid, BlockNumber),
    dt_return(<<"put_block">>, [BlockNumber], [Bucket, Key]),
    {noreply, State};
handle_cast({delete_block, ReplyPid, Bucket, Key, UUID, BlockNumber}, State=#state{riak_client=RcPid}) ->
    dt_entry(<<"delete_block">>, [BlockNumber], [Bucket, Key]),
    {FullBucket, FullKey} = full_bkey(Bucket, Key, UUID, BlockNumber),
    StartTime = os:timestamp(),

    %% do a get first to get the vclock (only do a head request though)
    GetOptions = [{r, 1}, {notfound_ok, false}, {basic_quorum, false}, head],
    _ = case riakc_pb_socket:get(block_pbc(RcPid), FullBucket, FullKey, GetOptions) of
            {ok, RiakObject} ->
                ok = delete_block(RcPid, ReplyPid, RiakObject, {UUID, BlockNumber});
        {error, notfound} ->
            %% If the block isn't found, assume it's been
            %% previously deleted by another delete FSM, and
            %% move on to the next block.
            riak_cs_delete_fsm:block_deleted(ReplyPid, {ok, {UUID, BlockNumber}})
    end,
    ok = riak_cs_stats:update_with_start(block_delete, StartTime),
    dt_return(<<"delete_block">>, [BlockNumber], [Bucket, Key]),
    {noreply, State};
handle_cast(_Msg, State) ->
    {noreply, State}.

get_block(ReplyPid, Bucket, Key, ClusterID, UUID, BlockNumber, RcPid) ->
    %% don't use proxy get if it's a local get
    %% or proxy get is disabled
    ProxyActive = riak_cs_config:proxy_get_active(),
    UseProxyGet = use_proxy_get(RcPid, ClusterID),

    case riak_cs_utils:n_val_1_get_requests() of
        true ->
            do_get_block(ReplyPid, Bucket, Key, ClusterID, UseProxyGet, ProxyActive, UUID,
                         BlockNumber, RcPid);
        false ->
            normal_nval_block_get(ReplyPid, Bucket, Key, ClusterID,
                                  UseProxyGet, UUID, BlockNumber, RcPid)
    end.

do_get_block(ReplyPid, Bucket, Key, ClusterID, UseProxyGet, ProxyActive,
             UUID, BlockNumber, RcPid) ->
    MaxRetries = riak_cs_config:get_env(riak_cs, block_get_max_retries, 5),
    do_get_block(ReplyPid, Bucket, Key, ClusterID, UseProxyGet, ProxyActive,
                 UUID, BlockNumber, RcPid, MaxRetries, 0).

do_get_block(ReplyPid, _Bucket, _Key, _ClusterID, _UseProxyGet, _ProxyActive,
             UUID, BlockNumber, _RcPid, MaxRetries, NumRetries)
  when is_atom(NumRetries) orelse NumRetries > MaxRetries ->
    Sorry = {error, notfound},
    ok = riak_cs_get_fsm:chunk(ReplyPid, {UUID, BlockNumber}, Sorry);
do_get_block(ReplyPid, Bucket, Key, ClusterID, UseProxyGet, ProxyActive,
             UUID, BlockNumber, RcPid, MaxRetries, NumRetries) ->
    ok = sleep_retries(NumRetries),

    dt_entry(<<"get_block">>, [BlockNumber], [Bucket, Key]),
    {FullBucket, FullKey} = full_bkey(Bucket, Key, UUID, BlockNumber),

    StartTime = os:timestamp(),
    GetOptions1 = n_val_one_options(),
    GetOptions2 = r_one_options(),

    ProceedFun = fun(OkReply) ->
            ok = riak_cs_stats:update_with_start(block_get_retry, StartTime),
            ok = riak_cs_get_fsm:chunk(ReplyPid, {UUID, BlockNumber}, OkReply),
            dt_return(<<"get_block">>, [BlockNumber], [Bucket, Key])
      end,
    RetryFun = fun(NewPause) ->
               ok = riak_cs_stats:update_with_start(block_get_retry, StartTime),
               do_get_block(ReplyPid, Bucket, Key, ClusterID, UseProxyGet,
                            ProxyActive, UUID, BlockNumber, RcPid, MaxRetries, NewPause)
            end,

    Timeout = timer:seconds(5),
    try_local_get(RcPid, FullBucket, FullKey, GetOptions1, GetOptions2,
                  Timeout, ProceedFun, RetryFun, NumRetries, UseProxyGet,
                  ProxyActive, ClusterID).

try_local_get(RcPid, FullBucket, FullKey, GetOptions1, GetOptions2,
              Timeout, ProceedFun, RetryFun, NumRetries, UseProxyGet,
              ProxyActive, ClusterID) ->
    case get_block_local(RcPid, FullBucket, FullKey, GetOptions1, Timeout) of
        {ok, _} = Success ->
            ProceedFun(Success);
        {error, {insufficient_vnodes,_,need,_}} ->
            RetryFun(NumRetries + 2);
        {error, Why} when Why == notfound;
                          Why == timeout;
                          Why == disconnected;
                          Why == <<"{insufficient_vnodes,0,need,1}">>;
                          Why == {insufficient_vnodes,0,need,1} ->
            handle_local_notfound(RcPid, FullBucket, FullKey, GetOptions2,
                                  ProceedFun, RetryFun, NumRetries, UseProxyGet,
                                  ProxyActive, ClusterID);
        {error, Other} ->
            _ = lager:error("do_get_block: other error 1: ~p\n", [Other]),
            RetryFun(failure)
    end.

handle_local_notfound(RcPid, FullBucket, FullKey, GetOptions2,
                      ProceedFun, RetryFun, NumRetries, UseProxyGet,
                      ProxyActive, ClusterID) ->

    case get_block_local(RcPid, FullBucket, FullKey, GetOptions2, 60*1000) of
        {ok, _} = Success ->
            ProceedFun(Success);
        {error, Why} when Why == disconnected;
                          Why == timeout ->
            _ = lager:debug("get_block_local/5 failed: {error, ~p}", [Why]),
            RetryFun(NumRetries + 1);
        {error, notfound} ->
            case UseProxyGet of
                true when ProxyActive ->
                    case get_block_remote(RcPid, FullBucket, FullKey,
                                          ClusterID, GetOptions2) of
                        {ok, _} = Success ->
                            ProceedFun(Success);
                        {error, _} ->
                            if UseProxyGet ->
                                    RetryFun(NumRetries + 1);
                                true ->
                                    RetryFun(failure)
                            end
                    end;
                true when not ProxyActive ->
{'fraction_non_alphanumeric': 0.17924596832776404, 'fraction_numerical': 0.004431207322388493, 'mean_word_length': 2.397038864898211, 'pattern_counts': {'":': 0, '<': 37, '<?xml version=': 0, '>': 156, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 32, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4485168', 'n_tokens_mistral': 8547, 'n_tokens_neox': 8302, 'n_words': 1983}