anchor.diff:  
   %% @private
   handle_cast({update, Membership0}, State) ->
  -    Membership = lists:usort(Membership0), %% Must sort list or random selection with seed is *nondeterministic.*
  +    Membership = membership(Membership0),
       {noreply, State#state{membership=Membership}};
   
  -handle_cast({write, {FromPid, FromRequestId}=From, Key, Value}, #state{membership=[Primary, Collaborator|_Rest], store=Store0}=State) ->
  +%% @private
  +handle_cast({read_local, From, Key}, #state{store=Store}=State) ->
  +    %% Get the value.
  +    Value = read(Key, Store),
  +
  +    %% Reply to the caller.
  +    partisan_pluggable_peer_service_manager:forward_message(From, {ok, Value}),
  +
  +    {noreply, State};
  +handle_cast({read, From, Key}=Request, #state{membership=[Primary|_Rest], store=Store}=State) ->
       case node() of 
           Primary ->
  -            partisan_logger:info("~p: node ~p received write for key ~p with value ~p", [?MODULE, node(), Key, Value]),
  -
  -            %% Write value locally.
  -            Store = write(Key, Value, Store0),
  -
  -            %% Forward to collaboration message.
  -            Myself = pnode(),
  -            psend(Collaborator, {collaborate, From, Myself, Key, Value}),
  -            partisan_logger:info("~p: node ~p sent replication request for key ~p with value ~p", [?MODULE, node(), Key, Value]),
  -
  -            %% Wait for collaboration ack before proceeding for n-host resilience (n = 2).
  -            receive
  -                {collaborate_ack, From, Key, Value} ->
  -                    partisan_logger:info("~p: node ~p ack received for key ~p value ~p", [?MODULE, node(), Key, Value]),
  -
  -                    %% Reply to caller.
  -                    psend(FromPid, {response, FromRequestId, ok}),
  -
  -                    {noreply, State#state{store=Store}}
  -            after
  -                %% We have to timeout, otherwise we block the gen_server.
  -                ?PB_RETRY_TIMEOUT ->
  -                    %% Reply to caller.
  -                    psend(FromPid, {response, FromRequestId, {error, timeout}}),
  -
  -                    {noreply, State}
  -            end;
  -        _ ->
  -            %% Forward the write request to the primary.
  -            psend(Primary, {forwarded_write, From, Key, Value}),
  -
  -            %% Return control, because backup requests may arrive before response does 
  -            %% under concurrent scheduling.
  +            %% Get the value.
  +            Value = read(Key, Store),
  +
  +            %% Reply to the caller.
  +            partisan_pluggable_peer_service_manager:forward_message(From, {ok, Value}),
  +
  +            {noreply, State};
  +        _ ->
  +            %% Reply to caller.
  +            partisan_logger:info("Node ~p is not the primary for request: ~p", [node(), Request]),
  +            partisan_pluggable_peer_service_manager:forward_message(From, {error, not_primary}),
  +
               {noreply, State}
       end;
  -%% @private
  -handle_cast({read, {FromPid, FromRequestId}=From, Key}, #state{membership=[Primary|_Rest], store=Store}=State) ->
  +handle_cast({write, From, Key, Value}=Request, #state{membership=[Primary|Rest]=Membership, outstanding=Outstanding0, store=Store0}=State) ->
       case node() of 
           Primary ->
  -            Value = read(Key, Store),
  -            partisan_logger:info("~p: node ~p received read for key ~p and returning value ~p", [?MODULE, node(), Key, Value]),
  -
  -            %% Send the response back to the user.
  -            psend(FromPid, {response, FromRequestId, {ok, Value}}),
  -
  -            {noreply, State};
  -        _ ->
  -            %% Forward the read request to the primary.
  -            psend(Primary, {forwarded_read, From, Key}),
  -
  -            %% Return control, because backup requests may arrive before response does 
  -            %% under concurrent scheduling.
  +            %% Add to list of outstanding requests.
  +            Replies = [node()],
  +            Outstanding = dict:store(Request, {Membership, Replies}, Outstanding0),
  +
  +            %% Send collaboration message to other nodes.
  +            CoordinatorNode = node(),
  +
  +            lists:foreach(fun(Node) ->
  +                partisan_pluggable_peer_service_manager:forward_message(Node, {collaborate, CoordinatorNode, Request})
  +            end, Rest),
  +
  +            %% Write to storage.
  +            Store = write(Key, Value, Store0),
  +
  +            {noreply, State#state{store=Store, outstanding=Outstanding}};
  +        _ ->
  +            %% Reply to caller.
  +            partisan_logger:info("Node ~p is not the primary for request: ~p", [node(), Request]),
  +            partisan_pluggable_peer_service_manager:forward_message(From, {error, not_primary}),
  +
  +            {noreply, State}
  +    end.
  +
  +%% @private
  +handle_info({collaborate_ack, ReplyingNode, {write, From, _Key, Value} = Request}, #state{outstanding=Outstanding0}=State) ->
  +    case dict:find(Request, Outstanding0) of
  +        {ok, {Membership, Replies0}} ->
  +            %% Update list of nodes that have acknowledged.
  +            Replies = Replies0 ++ [ReplyingNode],
  +
  +            case lists:usort(Membership) =:= lists:usort(Replies) of 
  +                true ->
  +                    partisan_logger:info("Received all replies for request ~p, acknowleding to user."),
  +                    partisan_pluggable_peer_service_manager:forward_message(From, {ok, Value});
  +                false ->
  +                    partisan_logger:info("Received replies from: ~p, but need replies from: ~p", [Replies, Membership -- Replies]),
  +                    ok
  +            end,
  +
  +            %% Update list.
  +            Outstanding = dict:store(Request, {Membership, Replies}, Outstanding0),
  +
  +            {noreply, State#state{outstanding=Outstanding}};
  +        _ ->
  +            partisan_logger:info("Received reply for unknown request: ~p", [Request]),
  +
               {noreply, State}
       end;
  -handle_cast(_Msg, State) ->
  -    {noreply, State}.
  -
  -%% @private
  -handle_info({forwarded_write, {FromPid, FromRequestId}=From, Key, Value}, #state{membership=[_Primary|Backups], store=Store0}=State) ->
  -    %% Figure 2c: I think this algorithm is not correct, but we'll see.
  -
  -    Store = write(Key, Value, Store0),
  -    partisan_logger:info("~p: node ~p received forwarded write for key ~p with value ~p", [?MODULE, node(), Key, Value]),
  -
  -    %% Send backup message to backups.
  -    lists:foreach(fun(Backup) -> psend(Backup, {backup, From, Key, Value}) end, Backups),
  -
  -    %% Send the response to the caller.
  -    psend(FromPid, {response, FromRequestId, ok}),
  -
  -    %% No need to send acknowledgement back to the forwarder: 
  -    %%  - not needed for control flow.
  -
  -    {noreply, State#state{store=Store}};
  -
  -%% @private
  -handle_info({forwarded_read, {FromPid, FromRequestId}, Key}, #state{store=Store}=State) ->
  -    Value = read(Key, Store),
  -    partisan_logger:info("~p: node ~p received forwarded read for key ~p and returning value ~p", [?MODULE, node(), Key, Value]),
  -
  -    %% Send the response to the caller.
  -    psend(FromPid, {response, FromRequestId, {ok, Value}}),
  -
  -    %% No need to send acknowledgement back to the forwarder: 
  -    %%  - not needed for control flow.
  -
  -    {noreply, State};
  -
  -%% @private
  -handle_info({collaborate, {FromPid, FromRequestId}=From, SourceNode, Key, Value}, #state{membership=[_Primary, _Collaborator | Backups], store=Store0}=State) ->
  -    %% Write value locally.
  -    Store = write(Key, Value, Store0),
  -    partisan_logger:info("~p: node ~p storing updated value key ~p value ~p", [?MODULE, node(), Key, Value]),
  -
  -    %% On ack, reply to caller.
  -    psend(FromPid, {response, FromRequestId, ok}),
  -
  -    %% Send write acknowledgement.
  -    psend(SourceNode, {collaborate_ack, From, Key, Value}),
  -    partisan_logger:info("~p: node ~p acknowledging value for key ~p value ~p", [?MODULE, node(), Key, Value]),
  -
  -    %% Send backup message to backups.
  -    lists:foreach(fun(Backup) -> psend(Backup, {backup, From, Key, Value}) end, Backups),
  -
  -    {noreply, State#state{store=Store}};
  -
  -%% @private
  -handle_info({backup, _From, _SourceNode, Key, Value}, #state{store=Store0}=State) ->
  -    %% Write value locally.
  -    Store = write(Key, Value, Store0),
  -    partisan_logger:info("~p: node ~p storing updated value key ~p value ~p", [?MODULE, node(), Key, Value]),
  -
  -    {noreply, State#state{store=Store}};
  -
  -%% @private
  -handle_info(_Msg, State) ->
  +handle_info({collaborate, CoordinatorNode, {write, _From, Key, Value}=Request}, #state{membership=[Primary|_Rest], store=Store0}=State) ->
  +    case node() of 
  +        Primary ->
  +            %% Do nothing.
  +            {noreply, State};
  +        _ ->
  +            %% Write to storage.
  +            Store = write(Key, Value, Store0),
  +
  +            %% Reply with collaborate acknowledgement.
  +            ReplyingNode = node(),
  +            partisan_logger:info("Node ~p is backup, responding to the primary ~p with acknowledgement", [node(), CoordinatorNode]),
  +            partisan_pluggable_peer_service_manager:forward_message(CoordinatorNode, {collaborate_ack, ReplyingNode, Request}),
  +
  +            {noreply, State#state{store=Store}}
  +    end;
  +handle_info(Msg, State) ->
  +    partisan_logger:info("Received message ~p with no handler!", [Msg]),
  +
       {noreply, State}.
   
   %% @private
  
positive.code: 
  %% @private
  write(Key, Value, Store) ->
      dict:store(Key, Value, Store).
negative.code: %% @private
  code_change(_OldVsn, State, _Extra) ->
      {ok, State}.
  
  %%%===================================================================
  %%% Internal functions
  