anchor.diff:  -export([init/1, handle_call/3, handle_cast/2, handle_info/2,
            code_change/3, terminate/2]).
   -export([partition_heartbeat/3,
  -         delayed_delivery/3,
  -         label_delivered/2,
  -         new_label/3]).
  +         check_ready/1,
  +         new_label/4]).
   
   -record(state, {vclock :: dict(),
                   delay,
  +                stable_time,
  +                pending,
                   labels :: list(),
                   myid}).
                   
  
positive.code: partition_heartbeat(MyId, Partition, Clock) ->
      gen_server:cast({global, reg_name(MyId)}, {partition_heartbeat, Partition, Clock}).
      
  new_label(MyId, Label, Partition) ->
      gen_server:cast({global, reg_name(MyId)}, {new_label, Label, Partition}).
  
  label_delivered(MyId, Element) ->
      gen_server:cast({global, reg_name(MyId)}, {label_delivered, Element}).
  
  init([MyId]) ->
      {ok, Ring} = riak_core_ring_manager:get_my_ring(),
      GrossPrefLists = riak_core_ring:all_preflists(Ring, 1),
      Dict = lists:foldl(fun(PrefList, Acc) ->
                          {Partition, _Node} = hd(PrefList),
                          saturn_proxy_vnode:heartbeat(PrefList, MyId),
                          dict:store(Partition, 0, Acc)
                         end, dict:new(), GrossPrefLists),
      {ok, #state{labels=orddict:new(), myid=MyId, vclock=Dict, delay=0}}.
  
  handle_cast({partition_heartbeat, Partition, Clock}, S0) ->
      S1 = update_vclock(Partition, Clock, S0),
      S2 = deliver_labels(S1),
      {noreply, S2};
  
  handle_cast({new_label, Label, Partition}, S0=#state{labels=Labels0}) ->
      Now = saturn_utilities:now_milisec(),
      TimeStamp = Label#label.timestamp,
      Labels1 = orddict:append(TimeStamp, {Now, Label}, Labels0),
      S1 = update_vclock(Partition, TimeStamp, S0#state{labels=Labels1}),
      S2 = deliver_labels(S1),
      {noreply, S2};
  
  handle_cast({label_delivered, Element}, S0=#state{labels=Labels0})->
      [{TimeStamp, ListLabels}|Rest] = Labels0,
      case lists:delete(Element, ListLabels) of
          [] ->
              S1 = deliver_labels(S0#state{labels=Rest});
          Other ->
              Labels1 = [{TimeStamp, Other}] ++ Rest,
              S1 = S0#state{labels=Labels1}
      end,
      {noreply, S1};
      
  handle_cast(_Info, State) ->
      {noreply, State}.
  
  handle_call(Info, From, State) ->
      lager:error("Weird message: ~p. From: ~p", [Info, From]),
      {noreply, State}.
  
  handle_info(Info, State) ->
      lager:info("Weird message: ~p", [Info]),
  
negative.code: code_change(_OldVsn, State, _Extra) ->
      {ok, State}.
  
  compute_stable(VClock) ->
      lists:foldl(fun({_Partition, Clock}, Min) ->
                      min(Clock, Min)
                   end, infinity, dict:to_list(VClock)).
  
  deliver_labels(Labels, StableTime, MyId, Deliverables0) ->
      case ets:first(Labels) of
          '$end_of_table' ->
              propagate_stream(lists:reverse(Deliverables0), MyId),
              false;
          {TimeStamp, Time, _Partition, Label}=Key when TimeStamp =< StableTime ->
              Now = saturn_utilities:now_microsec(),
              case Time > Now of
                  true ->
                      propagate_stream(lists:reverse(Deliverables0), MyId),
                      NextDelivery = trunc((Time - Now)/1000),
                      erlang:send_after(NextDelivery, self(), find_deliverables),                   
                      true;
                  false ->
                      true = ets:delete(Labels, Key),
                      BKey = Label#label.bkey,
                      deliver_labels(Labels, StableTime, MyId, [{BKey, Label}|Deliverables0]),
                      false
              end;
          _Key ->
              propagate_stream(lists:reverse(Deliverables0), MyId),
              false
      end.
  
  propagate_stream(FinalStream, MyId) ->
      case ?PROPAGATION_MODE of
  