anchor.diff:  -include("floppy.hrl").
   -include_lib("riak_core/include/riak_core_vnode.hrl").
   
  --record(dstate, {partition, last_commit_time, pending_operations}).
  +-record(dstate, {partition, last_commit_time, pending_operations, stable_time}).
   
   -export([start_vnode/1,
            trigger/2]).
  
positive.code: get_stable_time(Node, Prev_stable_time) ->
      lager:info("In get_stable_time"),
      case riak_core_vnode_master:sync_command(
                           Node, {get_active_txns}, ?CLOCKSIMASTER) of
  {ok, Active_txns} ->
      lager:info("Active txns before filtering ~p",[Active_txns]),
      lists:foldl(fun({_,{_TxId, Snapshot_time}}, Min_time) ->
                          case Min_time > Snapshot_time of
                              true ->
                                  Snapshot_time;
                              false ->
                                  Min_time
                          end
                  end,
                  now_milisec(erlang:now()),
                  Active_txns);
       _ -> Prev_stable_time
  end.
  
  now_milisec({MegaSecs,Secs,MicroSecs}) ->
      (MegaSecs*1000000 + Secs)*1000000 + MicroSecs.
  
negative.code:                              pending_operations = Pending}) ->
      Pending_operations = add_to_pending_operations(Pending, Write_set),
      lager:info("Pending Operations ~p",[Pending_operations]),
      %% reply to the caller
      From ! {ok, trigger_received},
      %% Calculate stable commit time to order operations
      Node = {Partition, node()}, %% Send ack to caller and continue processing
      Stable_time = get_stable_time(Node, Last_commit_time),
      %% Send a message to itself to process operations
      riak_core_vnode_master:command([{Partition,node()}],
                                     {process},
                                     clocksi_downstream_generator_vnode_master),
      {reply, {ok, trigger_received}, State#dstate{
                                        pending_operations = Pending_operations,
                                        stable_time = Stable_time}};
  
  handle_command({process}, _Sender,
                 State=#dstate{partition = Partition,
                               last_commit_time = Last_commit_time,
                               pending_operations = Pending_operations,
                               stable_time = Stable_time}) ->
      lager:info("Take updates before time : ~p",[Stable_time]),
      Sorted_ops = filter_operations(Pending_operations,
                                     Stable_time, Last_commit_time),
  