anchor.diff:      {K, undefined}.
   
   bucket_type_to_type(Bucket, BType) ->
  -    BProps = riak_core_bucket:get_bucket({BType, Bucket}),
  -    DataType = proplists:get_value(datatype, BProps),
  -    AllowMult = proplists:get_value(allow_mult, BProps),
  -    {AllowMult, DataType}.
  +    case riak_core_bucket:get_bucket({BType, Bucket}) of
  +        BProps when is_list(BProps) ->
  +            DataType = proplists:get_value(datatype, BProps),
  +            AllowMult = proplists:get_value(allow_mult, BProps),
  +            {ok, {AllowMult, DataType}};
  +        {error, no_type}=E ->
  +            E
  +    end.
   
   make_operation(Mod, Op, Ctx) ->
       #crdt_op{mod=Mod, op=Op, ctx=Ctx}.
  
positive.code:     {ok, riak_pb_codec:encode(Message)}.
  
  %% @doc process/2 callback. Handles an incoming request message.
  process(#dtfetchreq{bucket=B, key=K, r=R0, pr=PR0,
                      notfound_ok=NFOk, basic_quorum=BQ,
                      sloppy_quorum=SloppyQ, n_val=NVal,
                      type=BType, include_context=InclCtx},
          #state{client=C} = State) ->
      {AllowMult, Type} = bucket_type_to_type(B, BType),
      Mod = riak_kv_crdt:to_mod(Type),
      case {AllowMult, riak_kv_crdt:supported(Mod)} of
          {true, true} ->
              R = decode_quorum(R0),
              PR = decode_quorum(PR0),
              case C:get({BType, B}, K, make_option(r, R) ++
                             make_option(pr, PR) ++
                             make_option(notfound_ok, NFOk) ++
                             make_option(basic_quorum, BQ) ++
                             make_option(sloppy_quorum, SloppyQ) ++
                             make_option(n_val, NVal)) of
                  {ok, O} ->
                      {Ctx0, Value} = riak_kv_crdt:value(O, Mod),
                      Ctx = get_context(Ctx0, InclCtx),
                      Resp = riak_pb_dt_codec:encode_fetch_response(Type, Value, Ctx, ?MOD_MAP),
                      {reply, Resp, State};
                  {error, notfound} ->
                      Resp = riak_pb_dt_codec:encode_fetch_response(Type, undefined, undefined),
                      {reply, Resp, State};
                  {error, Reason} ->
                      {error, {format,Reason}, State};
                  Other -> lager:info("got ~p", [Other])
              end;
          {false, _} ->
              {error, {format, "Bucket must be allow_mult=true"}, State};
          {true, false} ->
              {error, {format, "`~p` is not a supported type", [Type]}, State}
      end;
  process(#dtupdatereq{bucket=B, key=K, type=BType,
                       w=W0, dw=DW0, pw=PW0, context=Ctx,
                       timeout=Timeout, sloppy_quorum=SloppyQ,
                       n_val=NVal, include_context=InclCtx,
                       op=Op0, return_body=RetVal},
          #state{client=C} = State) ->
      {AllowMult, Type} = bucket_type_to_type(B, BType),
      Mod = riak_kv_crdt:to_mod(Type),
      Op = riak_pb_dt_codec:decode_operation(Op0, ?MOD_MAP),
      OpType = riak_pb_dt_codec:operation_type(Op0),
      ModsMatch = mods_match(Mod, OpType),
      {Key, ReturnKey} = get_key(K),
      case {AllowMult, riak_kv_crdt:supported(Mod), ModsMatch} of
          {true, true, true} ->
              O = riak_kv_crdt:new({BType, B}, Key, Mod),
              %% erlang_protobuffs encodes as 1/0/undefined
              W = decode_quorum(W0),
              DW = decode_quorum(DW0),
              PW = decode_quorum(PW0),
              CrdtOp = make_operation(Mod, Op, Ctx),
              Options = [{crdt_op, CrdtOp}] ++ return_value(RetVal),
              case C:put(O, make_option(w, W) ++
                             make_option(dw, DW) ++
                             make_option(pw, PW) ++
                             make_option(timeout, timeout(Timeout)) ++
                             make_option(sloppy_quorum, SloppyQ) ++
                             make_option(n_val, NVal) ++
                             Options) of
                  ok ->
                      {reply, #dtupdateresp{key=ReturnKey}, State};
                  {ok, RObj} ->
                      {ReturnCtx, Value} = riak_kv_crdt:value(RObj, Type),
                      {reply, update_resp(Type, Value, get_context(ReturnCtx, InclCtx)), State};
                  {error, notfound} ->
                      {reply, #dtupdateresp{}, State};
                  {error, Reason} ->
                      {error, {format, Reason}, State}
              end;
          {false, _, _} ->
              {error, {format, "Bucket must be allow_mult=true"}, State};
          {true, false ,_} ->
              {error, {format, "Bucket datatype `~p` is not a supported type", [Type]}, State};
          {_, _, false} ->
              {error, {format, "Operation type is `~p` but  bucket type is `~p`.",
                       [OpType, Type]}, State}
      end.
  
  %% @doc process_stream/3 callback. This service does not create any
  %% streaming responses and so ignores all incoming messages.
  
negative.code: 
  -import(riak_pb_kv_codec, [decode_quorum/1]).
  
  -record(state, {client,
                  type,
                  mod,
                  op,
                  op_type,
                  return_key,
                  return_ctx,
                  return_value}).
  
  -define(DEFAULT_TIMEOUT, 60000).
  -define(DEFAULT_BUCKET, <<"default">>).
  
  %% @doc init/0 callback. Returns the service internal start
  %% state.
  