anchor.diff:              error
       end.
   
  -%% @doc Operation to insert a Snapshot in the cache and start 
  +%% @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.
  -snapshot_insert_gc(Key, SnapshotDict, OpsDict, SnapshotCache, OpsCache)-> 
  -	case (orddict:size(SnapshotDict))==?SNAPSHOT_THRESHOLD of 
  -	true ->
  -		PrunedSnapshots=orddict:from_list(lists:sublist(orddict:to_list(SnapshotDict), 1+?SNAPSHOT_THRESHOLD-?SNAPSHOT_MIN, ?SNAPSHOT_MIN)),
  -		FirstOp=lists:nth(1, PrunedSnapshots),
  -		{CommitTime, _S} = FirstOp,
  -		PrunedOps=prune_ops(OpsDict, CommitTime),
  -		ets:insert(SnapshotCache, {Key, PrunedSnapshots}),
  -        ets:insert(OpsCache, {Key, PrunedOps});
  -	false ->
  -		ets:insert(SnapshotCache, {Key, SnapshotDict})
  -	end.
  -	
  -%% @doc Remove from OpsDict all operations that have committed before Threshold. 
  --spec prune_ops(orddict:orddict(), {Dc::term(),CommitTime::non_neg_integer()})-> orddict:orddict().	
  +-spec snapshot_insert_gc(Key::term(), SnapshotDict::orddict:orddict(),
  +                         OpsDict::orddict:orddict(), atom() , atom() ) -> true.
  +snapshot_insert_gc(Key, SnapshotDict, OpsDict, SnapshotCache, OpsCache)->
  +    case (orddict:size(SnapshotDict))==?SNAPSHOT_THRESHOLD of
  +        true ->
  +            PrunedSnapshots=orddict:from_list(lists:sublist(orddict:to_list(SnapshotDict), 1+?SNAPSHOT_THRESHOLD-?SNAPSHOT_MIN, ?SNAPSHOT_MIN)),
  +            FirstOp=lists:nth(1, PrunedSnapshots),
  +            {CommitTime, _S} = FirstOp,
  +            PrunedOps=prune_ops(OpsDict, CommitTime),
  +            ets:insert(SnapshotCache, {Key, PrunedSnapshots}),
  +            ets:insert(OpsCache, {Key, PrunedOps});
  +        false ->
  +            ets:insert(SnapshotCache, {Key, SnapshotDict})
  +    end.
  +
  +%% @doc Remove from OpsDict all operations that have committed before Threshold.
  +-spec prune_ops(orddict:orddict(), {Dc::term(),CommitTime::non_neg_integer()})-> orddict:orddict().
   prune_ops(OpsDict, Threshold)->
  -	orddict:filter(fun(_Key, Value) -> 
  -				(belongs_to_snapshot(Threshold,(lists:last(Value))#clocksi_payload.snapshot_time)) end, OpsDict).
  +    orddict:filter(fun(_Key, Value) ->
  +                           (belongs_to_snapshot(Threshold,(lists:last(Value))#clocksi_payload.snapshot_time)) end, OpsDict).
   
   
   %% @doc Insert an operation and start garbage collection triggered by writes.
   %% 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(), 
  -	orddict:orddict(), atom() , atom() )-> true.
  -op_insert_gc(Key,DownstreamOp, OpsDict, OpsCache, SnapshotCache)-> 
  -    case (orddict:size(OpsDict))>=?OPS_THRESHOLD of 
  -    true ->
  -	    Type=DownstreamOp#clocksi_payload.type,
  -	    SnapshotTime=DownstreamOp#clocksi_payload.snapshot_time,
  -	    Type=DownstreamOp#clocksi_payload.type,
  -	    SnapshotTime=DownstreamOp#clocksi_payload.snapshot_time,
  -	    ok=internal_read(ignore, Key, Type, SnapshotTime, OpsCache, SnapshotCache),
  -	    OpsDict1=orddict:append(DownstreamOp#clocksi_payload.commit_time, DownstreamOp, OpsDict),
  -		ets:insert(OpsCache, {Key, OpsDict1});
  -	false ->
  -		OpsDict1=orddict:append(DownstreamOp#clocksi_payload.commit_time, DownstreamOp, OpsDict),
  -		ets:insert(OpsCache, {Key, OpsDict1})
  -    end.
  -
  -     
  --ifdef(TEST). 
  +-spec op_insert_gc(term(), clocksi_payload(),
  +                   orddict:orddict(), atom() , atom() )-> true.
  +op_insert_gc(Key,DownstreamOp, OpsDict, OpsCache, SnapshotCache)->
  +    case (orddict:size(OpsDict))>=?OPS_THRESHOLD of
  +        true ->
  +            Type=DownstreamOp#clocksi_payload.type,
  +            SnapshotTime=DownstreamOp#clocksi_payload.snapshot_time,
  +            Type=DownstreamOp#clocksi_payload.type,
  +            SnapshotTime=DownstreamOp#clocksi_payload.snapshot_time,
  +            ok=internal_read(ignore, Key, Type, SnapshotTime, OpsCache, SnapshotCache),
  +            OpsDict1=orddict:append(DownstreamOp#clocksi_payload.commit_time, DownstreamOp, OpsDict),
  +            ets:insert(OpsCache, {Key, OpsDict1});
  +        false ->
  +            OpsDict1=orddict:append(DownstreamOp#clocksi_payload.commit_time, DownstreamOp, OpsDict),
  +            ets:insert(OpsCache, {Key, OpsDict1})
  +    end.
  +
  +
  +-ifdef(TEST).
   
   %% @doc Testing filter_ops works in both situations, when the function receives
   %%      what it expects and when it receives something in an unexpected format.
   filter_ops_test() ->
  -	Ops=orddict:new(),
  -	Ops1=orddict:append(key1, [a1, a2], Ops),
  -	Ops2=orddict:append(key2, [b1, b2], Ops1),
  -	Ops3=orddict:append(key3, [c1, c2], Ops2),
  -	Result=filter_ops(Ops3),
  -	?assertEqual(Result, {ok, [[a1,a2], [b1,b2], [c1,c2]]}),
  -	Result1=filter_ops({some, thing}),
  -	?assertEqual(Result1, {error, wrong_format}),
  -	Result2=filter_ops([anything]),
  -	?assertEqual(Result2, {error, wrong_format}).   
  +    Ops=orddict:new(),
  +    Ops1=orddict:append(key1, [a1, a2], Ops),
  +    Ops2=orddict:append(key2, [b1, b2], Ops1),
  +    Ops3=orddict:append(key3, [c1, c2], Ops2),
  +    Result=filter_ops(Ops3),
  +    ?assertEqual(Result, {ok, [[a1,a2], [b1,b2], [c1,c2]]}),
  +    Result1=filter_ops({some, thing}),
  +    ?assertEqual(Result1, {error, wrong_format}),
  +    Result2=filter_ops([anything]),
  +    ?assertEqual(Result2, {error, wrong_format}).
   -endif.
  -
  -    
  -    
  
positive.code: %%             SnapshotTime = vector clock
  %%      Outptut: true or false
  -spec belongs_to_snapshot({Dc::term(),CommitTime::non_neg_integer()},
                          SnapshotTime::vectorclock:vectorclock()) -> boolean()|error.
  belongs_to_snapshot({Dc, CommitTime}, SnapshotTime) ->
      case vectorclock:get_clock_of_dc(Dc, SnapshotTime) of
          {ok, Ts} ->
  
negative.code:     %% TODO: what if all of the following was done asynchronously?
      case logging_vnode:append(Node,LogId,LogRecord) of
          {ok, _} ->
              riak_core_vnode:reply(Sender, ok),
              case ets:lookup(OpsCache, Key) of
                  []->
                      OpsDict=orddict:new();
                  [{_, OpsDict}]->
                      OpsDict
              end,
              op_insert_gc(Key,DownstreamOp, OpsDict, OpsCache, SnapshotCache),
              ok;
          {error, Reason} ->
              riak_core_vnode:reply(Sender, {error, Reason})
      end.
  
  
  %% @doc Obtains, from an orddict of Snapshots, the latest snapshot that can be included in
  %% a snapshot identified by SnapshotTime
  -spec get_latest_snapshot(SnapshotDict::orddict:orddict(), SnapshotTime::vectorclock:vectorclock())
                           -> {ok, term()} | {error, no_snapshot}| {error, wrong_format, term()}.
  get_latest_snapshot(SnapshotDict, SnapshotTime) ->
      case SnapshotDict of
          []->
              {ok,[]};
          [H|T]->
              case orddict:filter(fun(Key, _Value) ->
                                          belongs_to_snapshot(Key, SnapshotTime) end, [H|T]) of
                  []->
                      {error,no_snapshot};
                  [H1|T1]->
                      {CommitTime, Snapshot} = lists:last([H1|T1]),
                      {ok, {CommitTime, Snapshot}}
              end;
          Anything ->
              {error, wrong_format, Anything}
      end.
  
  %% @doc Get a list of operations from an orddict of operations
  -spec filter_ops(orddict:orddict()) -> {ok, list()} | {error, wrong_format}.
  filter_ops(Ops) ->
      filter_ops(Ops, []).
  -spec filter_ops(orddict:orddict(), list()) -> {ok, list()} | {error, wrong_format}.
  filter_ops([], Acc) ->
      {ok, Acc};
  filter_ops([H|T], Acc) ->
      case H of
          {_Key, Ops} ->
              filter_ops(T,lists:append(Acc, Ops));
          _ ->
              {error, wrong_format}
      end;
  filter_ops(_, _Acc) ->
      {error, wrong_format}.
  
  
  %% @doc Check whether a Key's operation or stored snapshot is included
  %%		in a snapshot defined by a vector clock
  %%      Input: Dc = Datacenter Id
  