anchor.diff:      %% 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),
  +            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 
  +%% @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()}.
  +                         -> {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.
  +    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, []).
  +    filter_ops(Ops, []).
   -spec filter_ops(orddict:orddict(), list()) -> {ok, list()} | {error, wrong_format}.
   filter_ops([], Acc) ->
  -	{ok, Acc};
  +    {ok, Acc};
   filter_ops([H|T], Acc) ->
  -	case H of 
  -	{_Key, Ops} ->
  -		filter_ops(T,lists:append(Acc, Ops));
  -	_ ->
  -		{error, wrong_format}
  -	end;
  +    case H of
  +        {_Key, Ops} ->
  +            filter_ops(T,lists:append(Acc, Ops));
  +        _ ->
  +            {error, wrong_format}
  +    end;
   filter_ops(_, _Acc) ->
  -	{error, wrong_format}.
  -	
  -    
  +    {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
  
positive.code: 
  %%---------------- Internal Functions -------------------%%
  
  %% @doc This function takes care of reading. It is implemented here for not blocking the 
  %% vnode when the write function calls it. That is done for garbage collection. 
  -spec internal_read(term(),term(), atom(), vectorclock:vectorclock(), atom() , atom() ) -> ok.
  internal_read(Sender, Key, Type, SnapshotTime, OpsCache, SnapshotCache) ->
      % get the latest snapshot for the key
      case ets:lookup(SnapshotCache, Key) of
      	[] ->
      		NewSnapshot=clocksi_materializer:new(Type),
              case ets:lookup(OpsCache, Key) of
              [] ->
              	riak_core_vnode:reply(Sender, {ok, NewSnapshot});
              [{_, OpsDict}] ->
              	{ok, Ops}= filter_ops(OpsDict),
              	LastOp=lists:last(Ops),
              	TxId = LastOp#clocksi_payload.txid,
              	{ok, Snapshot, CommitTime} = clocksi_materializer:materialize(Type, NewSnapshot, SnapshotTime, [LastOp], TxId),
              	riak_core_vnode:reply(Sender, {ok, Snapshot}),
              	SnapshotDict=orddict:new(),
              	ets:insert(SnapshotCache, {Key, orddict:store(CommitTime,Snapshot, SnapshotDict)})
              end;
          [{_, SnapshotDict}] -> 
              case get_latest_snapshot(SnapshotDict, SnapshotTime) of
  			{ok, {_SnapshotCommitTime, LatestSnapshot}}->
  				case ets:lookup(OpsCache, Key) of
  				[] ->
  					riak_core_vnode:reply(Sender, {ok, LatestSnapshot});
  				[{_, OpsDict}] ->
  					{ok, Ops}= filter_ops(OpsDict),
  					case Ops of
  					[] ->
  						riak_core_vnode:reply(Sender, {ok, LatestSnapshot});
  					[H|T] ->
  						LastOp=lists:last([H|T]),
  						TxId = LastOp#clocksi_payload.txid,
  						{ok, Snapshot, CommitTime} = clocksi_materializer:materialize(Type, LatestSnapshot, SnapshotTime, [LastOp], TxId),
  						case (Sender /= ignore) of
  						true ->
  							riak_core_vnode:reply(Sender, {ok, Snapshot});
  						false ->
  							false
  						end,
  						SnapshotDict1=orddict:store(CommitTime,Snapshot, SnapshotDict),
  						snapshot_insert_gc(Key,SnapshotDict1, OpsDict, SnapshotCache, OpsCache)
  					end
  				end;
  			{error, no_snapshot} ->
  				%%FIX THIS, READ FROM THE LOG WHEN THERE IS NO SNAPSHOT.
  				riak_core_vnode:reply(Sender, {error, no_snapshot})
  			end	
      end,
      ok.
  	%TODO: trigger the GC mechanism asynchronously
  	
  	
  %% @doc This function takes care of appending an operation to the log and
  %%  to the cache.	
  -spec internal_update(term(), Key::term(), DownstreamOp::clocksi_payload(), 
  		OpsCache::atom(), SnapshotCache::atom()) ->	ok.
  internal_update(Sender, Key, DownstreamOp, OpsCache, SnapshotCache) ->	
  %% TODO: Remove unnecessary information from op_payload in log_Record
      LogRecord = #log_record{tx_id=DownstreamOp#clocksi_payload.txid,
                              op_type=downstreamop,
                              op_payload=DownstreamOp},
  
negative.code:             error
      end.
  
  %% @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().
  prune_ops(OpsDict, Threshold)->
      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).
  
  %% @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}).
  -endif.
  