anchor.diff:  %%             CommitTime = local commit time of this Snapshot at DC
   %%             SnapshotTime = vector clock
   %%      Outptut: true or false
  --spec belongs_to_snapshot(commit_time(),
  -                          snapshot_time()) -> boolean().
  +-spec belongs_to_snapshot(commit_time(), snapshot_time()) -> boolean().
   belongs_to_snapshot({Dc, CommitTime}, SnapshotTime) ->
   	{ok, Ts}= vectorclock:get_clock_of_dc(Dc, SnapshotTime),
   	CommitTime =< Ts.
   
   %% @doc Operation to insert a Snapshot in the cache and start
   %%      Garbage collection triggered by reads.
  --spec snapshot_insert_gc(Key::term(), SnapshotDict::orddict:orddict(),
  -                         OpsDict::orddict:orddict(), atom() , atom() ) -> true.
  +-spec snapshot_insert_gc(key(), orddict:orddict(), orddict:orddict(), ets:tid() , ets:tid() ) -> true.
   snapshot_insert_gc(Key, SnapshotDict, OpsDict, SnapshotCache, OpsCache)->
       case (orddict:size(SnapshotDict))==?SNAPSHOT_THRESHOLD of
           true ->
  
positive.code: %% the mechanism is very simple; when there are more than OPS_THRESHOLD
  %% operations for a given key, just perform a read, that will trigger
  %% the GC mechanism.
  -spec op_insert_gc(term(), clocksi_payload(),
                     atom() , atom() )-> true.
  op_insert_gc(Key,DownstreamOp, OpsCache, SnapshotCache)->
      OpsDict = case ets:lookup(OpsCache, Key) of
                    []->
                        orddict:new();
  
negative.code:     riak_core_vnode_master:get_vnode_pid(I, ?MODULE).
  
  %% @doc Read state of key at given snapshot time
  -spec read(key(), type(), snapshot_time(), txid()) -> {ok, term()} | {error, reason()}.
  read(Key, Type, SnapshotTime, TxId) ->
      DocIdx = riak_core_util:chash_key({?BUCKET, term_to_binary(Key)}),
      Preflist = riak_core_apl:get_primary_apl(DocIdx, 1, materializer),
  