anchor.diff:                                 ReqId, DownNodes, Service, CHBin,
                                  ?AVAIL_NODE_FUN).
   
  -
  -replace_subpartition_chunk(_VnodeIdx, Node, {_Mask, _Bits}=SubpID, NVal,
  -                           ReqId, DownNodes, Service, CHBin,
  -                           AvailNodeFun) ->
  -    Offset = ReqId rem NVal,
  -
  -    %% We have our own idea of what nodes are available. The client
  -    %% may have a different idea of offline nodes based on network
  -    %% partitions, so we take that into account.
  -    %%
  -    %% The client can't really tell us what nodes it thinks are up (it
  -    %% only knows hostnames at best) but the opaque coverage chunks it
  -    %% uses have node names embedded in them, and it can tell us which
  -    %% chunks do *not* work.
  -    UpNodes = AvailNodeFun(Service, CHBin) -- [Node|DownNodes],
  -
  -    %% We don't know what partition this subpartition is in, but we
  -    %% can request a preflist for it by converting the subpartition to
  -    %% a document index.
  -    %%
  -    %% Unlike traditional coverage filters to document key hash
  -    %% mappings which have off-by-one adjustments, subpartition masks
  -    %% map directly against the relevant key hashes.
  -    DocIdx = convert(SubpID, subpartition_index),
  -    PrefList =
  -        docidx_to_preflist(DocIdx, NVal, Offset, UpNodes, CHBin),
  -    singular_preflist_to_chunk(safe_hd(PrefList), SubpID).
  -
  -singular_preflist_to_chunk([], _SubpID) ->
  -    {error, primary_partition_unavailable};
  -singular_preflist_to_chunk({VnodeIdx, Node}, SubpID) ->
  -    [{VnodeIdx, Node, SubpID}].
  -
  -
  -%% IMPORTANT: `Count' is assumed to be a power of 2 as determined by
  -%% `riak_client'. Anything else will behave badly.
  --spec create_subpartition_plan('all'|'allup',
  -                               pos_integer(),
  -                               pos_integer(),
  -                               pos_integer(),
  -                               req_id(), atom()) ->
  -                                      {error, term()} | subp_plan().
  -create_subpartition_plan(VNodeTarget, NVal, Count, PVC, ReqId, Service) ->
  -    {ok, ChashBin} = riak_core_ring_manager:get_chash_bin(),
  -    create_subpartition_plan(VNodeTarget, NVal, Count, PVC, ReqId, Service,
  -                             ChashBin, ?AVAIL_NODE_FUN).
  -
  -%% Must be able to comply with PVC if the target is 'all'
  -check_pvc(List, _PVC, allup) ->
  -    List;
  -check_pvc(List, PVC, all) ->
  -    check_pvc2(List, length(List), PVC).
  -
  -check_pvc2(List, Len, PVC) when Len >= PVC ->
  -    List;
  -check_pvc2(_List, _Len, _PVC) ->
  -    [].
  -
  -%% @private
  -create_subpartition_plan(VNodeTarget, NVal, Count, PVC, ReqId, Service, CHBin, AvailNodeFun) ->
  -    MaskBSL = data_bits(Count),
  -
  -    %% Calculate an offset based on the request id to offer the
  -    %% possibility of different sets of VNodes being used even when
  -    %% all nodes are available.
  -    Offset = ReqId rem NVal,
  -
  -    RingSize = chashbin:num_partitions(CHBin),
  -
  -    UpNodes = AvailNodeFun(Service, CHBin),
  -    SubpList =
  -        lists:map(fun(SubpCounter) ->
  -                          SubpID = {SubpCounter, MaskBSL},
  -                          SubpIndex = convert(SubpID, subpartition_index),
  -                          PartID =
  -                              {partition_id,
  -                               chashbin:responsible_position(SubpIndex, CHBin),
  -                               RingSize},
  -
  -                          %% PVC is much like R; if the number of
  -                          %% available primary partitions won't reach
  -                          %% the specified PVC value, don't bother
  -                          %% including this partition at all. We can
  -                          %% decide later (based on all vs allup)
  -                          %% whether to return the successful
  -                          %% components of the coverage plan
  -                          {PartID, SubpID,
  -                           check_pvc(
  -                             docidx_to_preflist(SubpIndex, NVal, Offset, UpNodes, CHBin),
  -                             PVC, VNodeTarget)}
  -                  end,
  -                  lists:seq(0, Count - 1)),
  -
  -    %% Now we have a list of tuples; each subpartition maps to a
  -    %% partition ID, subpartition ID, and a list of zero or more
  -    %% {vnode_index, node} tuples for that partition
  -    maybe_create_subpartition_plan(VNodeTarget, SubpList, PVC).
  -
  -maybe_create_subpartition_plan(allup, SubpList, PVC) ->
  -    map_subplist_to_plan(SubpList, PVC);
  -maybe_create_subpartition_plan(all, SubpList, PVC) ->
  -    maybe_create_subpartition_plan(all, lists:keyfind([], 3, SubpList), SubpList, PVC).
  -
  -maybe_create_subpartition_plan(all, false, SubpList, PVC) ->
  -    map_subplist_to_plan(SubpList, PVC);
  -maybe_create_subpartition_plan(all, _, _SubpList, _PVC) ->
  -    {error, insufficient_vnodes_available}.
  -
  -map_subplist_to_plan(SubpList, PVC) ->
  -    lists:flatten(
  -      lists:filtermap(fun({_PartID, _SubpID, []}) ->
  -                              false;
  -                         ({_PartID, SubpID, Vnodes}) ->
  -                              {true,
  -                               map_pvc_vnodes(Vnodes, SubpID, PVC)}
  -                    end, SubpList)).
  -
  -
  -map_pvc_vnodes(Vnodes, SubpID, PVC) ->
  -    map_pvc_vnodes(Vnodes, SubpID, PVC, []).
  -
  -map_pvc_vnodes(_Vnodes, _SubpID, 0, Accum) ->
  -    lists:reverse(Accum);
  -map_pvc_vnodes([{VnodeIdx, Node}|Tail], SubpID, PVC, Accum) ->
  -    map_pvc_vnodes(Tail, SubpID, PVC-1,
  -                   [{ VnodeIdx, Node, SubpID }] ++ Accum).
  -
  -
  -%% @private
  -%% Make it easier to unit test create_plan/5.
  +%% @doc Take a `vnode_coverage' record and interpret it as
  +%%      needed by `riak_core_coverage_fsm:initialize'
  +-spec interpret_plan(vnode_coverage()) ->
  +                            {list({index(), node()}),
  +                             list({index(), list(index())|tuple()})}.
  +interpret_plan(#vnode_coverage{vnode_identifier=TargetHash,
  +                               subpartition={Mask, BSL}}) ->
  +    {[{TargetHash, node()}], [{TargetHash, {Mask, BSL}}]};
  +interpret_plan(#vnode_coverage{vnode_identifier=TargetHash,
  +                               partition_filters=[]}) ->
  +    {[{TargetHash, node()}], []};
  +interpret_plan(#vnode_coverage{vnode_identifier=TargetHash,
  +                            partition_filters=HashFilters}) ->
  +    {[{TargetHash, node()}], [{TargetHash, HashFilters}]}.
  +
  +
  +%% ====================================================================
  +%% Internal functions
  +%% ====================================================================
  +
  +%% @private
   create_traditional_plan(VNodeTarget, NVal, PVC, ReqId, Service, CHBin, AvailNodeFun) ->
       RingSize = chashbin:num_partitions(CHBin),
   
  
positive.code: %% Public API
  %% ===================================================================
  
  %% @doc Create a coverage plan to distribute work to a set
  %%      of covering VNodes around the ring. If the first argument
  %%      is a vnode_coverage record, that means we've previously
  %%      generated a coverage plan and we're being fed back one
  %%      element of it. Return that element in the proper format.
  -spec create_plan(vnode_selector(),
                    pos_integer(),
                    pos_integer(),
  
negative.code:     create_traditional_plan(VNodeTarget, NVal, PVC, ReqId, Service,
                              CHBin, ?AVAIL_NODE_FUN).
  
  %% @doc Create a "mini" traditional coverage plan to replace
  %%      components of a previously-generated plan that are not useful
  %%      to the client because the node is unavailable
  -spec replace_traditional_chunk(VnodeIdx :: index(),
                                  Node :: node(),
                                  Filters :: list(index()),
  