riak_core_lite

%% -------------------------------------------------------------------
%%
%% riak_core: Core Riak Application
%%
%% Copyright (c) 2007-2010 Basho Technologies, Inc.  All Rights Reserved.
%%
%% This file is provided to you under the Apache License,
%% Version 2.0 (the "License"); you may not use this file
%% except in compliance with the License.  You may obtain
%% a copy of the License at
%%
%%   http://www.apache.org/licenses/LICENSE-2.0
%%
%% Unless required by applicable law or agreed to in writing,
%% software distributed under the License is distributed on an
%% "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
%% KIND, either express or implied.  See the License for the
%% specific language governing permissions and limitations
%% under the License.
%%
%% -------------------------------------------------------------------

%% @doc A simple Erlang implementation of vector clocks as inspired by Lamport logical clocks.
%%
%% @reference Leslie Lamport (1978). "Time, clocks, and the ordering of events
%% in a distributed system". Communications of the ACM 21 (7): 558-565.
%%
%% @reference Friedemann Mattern (1988). "Virtual Time and Global States of
%% Distributed Systems". Workshop on Parallel and Distributed Algorithms:
%% pp. 215-226

-module(vclock).

-export([fresh/0,
         fresh/2,
         descends/2,
         dominates/2,
         descends_dot/2,
         pure_dot/1,
         merge/1,
         get_counter/2,
         get_timestamp/2,
         get_dot/2,
         valid_dot/1,
         increment/2,
         increment/3,
         all_nodes/1,
         equal/2,
         prune/3,
         timestamp/0,
         last_modified/1]).

-ifdef(TEST).

-include_lib("eunit/include/eunit.hrl").

-endif.

-export_type([vclock/0,
              timestamp/0,
              vclock_node/0,
              dot/0,
              pure_dot/0]).

-type vclock() :: [dot()].

-type dot() :: {vclock_node(),
                {counter(), timestamp()}}.

-type pure_dot() :: {vclock_node(), counter()}.

% Nodes can have any term() as a name, but they must differ from each other.
-type vclock_node() :: term().

-type counter() :: integer().

-type timestamp() :: integer().

% @doc Create a brand new vclock.
-spec fresh() -> vclock().

fresh() -> [].

-spec fresh(vclock_node(), counter()) -> vclock().

fresh(Node, Count) -> [{Node, {Count, timestamp()}}].

%% @doc Return true if Va is a direct descendant of Vb,
%%      else false -- remember, a vclock is its own descendant!
-spec descends(Va :: vclock(),
               Vb :: vclock()) -> boolean().

descends(_, []) ->
    % all vclocks descend from the empty vclock
    true;
descends(Va, Vb) ->
    [{NodeB, {CtrB, _T}} | RestB] = Vb,
    case lists:keyfind(NodeB, 1, Va) of
        false -> false;
        {_, {CtrA, _TSA}} ->
            CtrA >= CtrB andalso descends(Va, RestB)
    end.

%% @doc does the given `vclock()' descend from the given `dot()'. The
%% `dot()' can be any vclock entry returned from
%% `get_entry/2'. returns `true' if the `vclock()' has an entry for
%% the `actor' in the `dot()', and that the counter for that entry is
%% at least that of the given `dot()'. False otherwise. Call with a
%% valid entry or you'll get an error.
%%
%% @see descends/2
%% @see get_entry/3
%% @see dominates/2
-spec descends_dot(vclock(), dot()) -> boolean().

descends_dot(Vclock, Dot) -> descends(Vclock, [Dot]).

%% @doc in some cases the dot without timestamp data is needed.
-spec pure_dot(dot()) -> pure_dot().

pure_dot({N, {C, _TS}}) -> {N, C}.

%% @doc true if `A' strictly dominates `B'. Note: ignores
%% timestamps. In Riak it is possible to have vclocks that are
%% identical except for timestamps. When two vclocks descend each
%% other, but are not equal, they are concurrent. See source comment
%% for more details. (Actually you can have indentical clocks
%% including timestamps, that represent different events, but let's
%% not go there.)
%%
-spec dominates(vclock(), vclock()) -> boolean().

dominates(A, B) ->
    %% In a sane world if two vclocks descend each other they MUST be
    %% equal. In riak they can descend each other and have different
    %% timestamps(!) How? Deleted keys, re-written, then restored is
    %% one example. See riak_kv#679 for others. This is why we must
    %% check descends both ways rather than checking descends(A, B)
    %% and not equal(A, B). Do not "optimise" this to dodge the second
    %% descends call! I know that the laws of causality say that each
    %% actor must act serially, but Riak breaks that.
    descends(A, B) andalso not descends(B, A).

% @doc Combine all VClocks in the input list into their least possible
%      common descendant.
-spec merge(VClocks :: [vclock()]) -> vclock().

merge([]) -> [];
merge([SingleVclock]) -> SingleVclock;
merge([First | Rest]) ->
    merge(Rest, lists:keysort(1, First)).

merge([], NClock) -> NClock;
merge([AClock | VClocks], NClock) ->
    merge(VClocks,
          merge(lists:keysort(1, AClock), NClock, [])).

merge([], [], AccClock) -> lists:reverse(AccClock);
merge([], Left, AccClock) ->
    lists:reverse(AccClock, Left);
merge(Left, [], AccClock) ->
    lists:reverse(AccClock, Left);
merge(V = [{Node1, {Ctr1, TS1} = CT1} = NCT1 | VClock],
      N = [{Node2, {Ctr2, TS2} = CT2} = NCT2 | NClock],
      AccClock) ->
    if Node1 < Node2 -> merge(VClock, N, [NCT1 | AccClock]);
       Node1 > Node2 -> merge(V, NClock, [NCT2 | AccClock]);
       true ->
           ({_Ctr, _TS} = CT) = if Ctr1 > Ctr2 -> CT1;
                                   Ctr1 < Ctr2 -> CT2;
                                   true -> {Ctr1, erlang:max(TS1, TS2)}
                                end,
           merge(VClock, NClock, [{Node1, CT} | AccClock])
    end.

% @doc Get the counter value in VClock set from Node.
-spec get_counter(Node :: vclock_node(),
                  VClock :: vclock()) -> counter().

get_counter(Node, VClock) ->
    case lists:keyfind(Node, 1, VClock) of
        {_, {Ctr, _TS}} -> Ctr;
        false -> 0
    end.

% @doc Get the timestamp value in a VClock set from Node.
-spec get_timestamp(Node :: vclock_node(),
                    VClock :: vclock()) -> timestamp() | undefined.

get_timestamp(Node, VClock) ->
    case lists:keyfind(Node, 1, VClock) of
        {_, {_Ctr, TS}} -> TS;
        false -> undefined
    end.

% @doc Get the last timestamp from a clock in a friendly format
-spec last_modified(VClock :: vclock()) -> calendar:datetime() | undefined.
last_modified([]) ->
    undefined;
last_modified(VClock) ->
    TSL = lists:map(fun({_, {_Ctr, TS}}) -> TS end, VClock),
    calendar:gregorian_seconds_to_datetime(lists:last(lists:sort(TSL))).

% @doc Get the entry `dot()' for `vclock_node()' from `vclock()'.
-spec get_dot(Node :: vclock_node(),
              VClock :: vclock()) -> {ok, dot()} | undefined.

get_dot(Node, VClock) ->
    case lists:keyfind(Node, 1, VClock) of
        false -> undefined;
        Entry -> {ok, Entry}
    end.

%% @doc is the given argument a valid dot, or entry?
-spec valid_dot(dot()) -> boolean().

valid_dot({_, {Cnt, TS}})
    when is_integer(Cnt), is_integer(TS) ->
    true;
valid_dot(_) -> false.

% @doc Increment VClock at Node.
-spec increment(Node :: vclock_node(),
                VClock :: vclock()) -> vclock().

increment(Node, VClock) ->
    increment(Node, timestamp(), VClock).

% @doc Increment VClock at Node.
-spec increment(Node :: vclock_node(),
                IncTs :: timestamp(), VClock :: vclock()) -> vclock().

increment(Node, IncTs, VClock) ->
    {{_Ctr, _TS} = C1, NewV} = case lists:keytake(Node,
                                                  1,
                                                  VClock)
                                   of
                                   false -> {{1, IncTs}, VClock};
                                   {value, {_N, {C, _T}}, ModV} ->
                                       {{C + 1, IncTs}, ModV}
                               end,
    [{Node, C1} | NewV].

% @doc Return the list of all nodes that have ever incremented VClock.
-spec all_nodes(VClock :: vclock()) -> [vclock_node()].

all_nodes(VClock) -> [X || {X, {_, _}} <- VClock].

-define(DAYS_FROM_GREGORIAN_BASE_TO_EPOCH,
        1970 * 365 + 478).

-define(SECONDS_FROM_GREGORIAN_BASE_TO_EPOCH,
        (?DAYS_FROM_GREGORIAN_BASE_TO_EPOCH) * 24 * 60 *
            60).    %% == calendar:datetime_to_gregorian_seconds({{1970,1,1},{0,0,0}})

% @doc Return a timestamp for a vector clock
-spec timestamp() -> timestamp().

timestamp() ->
    %% Same as calendar:datetime_to_gregorian_seconds(erlang:universaltime()),
    %% but significantly faster.
    {MegaSeconds, Seconds, _} = os:timestamp(),
    (?SECONDS_FROM_GREGORIAN_BASE_TO_EPOCH) +
        MegaSeconds * 1000000
        + Seconds.

% @doc Compares two VClocks for equality.
-spec equal(VClockA :: vclock(),
            VClockB :: vclock()) -> boolean().

equal(VA, VB) -> lists:sort(VA) =:= lists:sort(VB).

% @doc Possibly shrink the size of a vclock, depending on current age and size.
-spec prune(V :: vclock(), Now :: integer(),
            BucketProps :: term()) -> vclock().

prune(V, Now, BucketProps) ->
    %% This sort need to be deterministic, to avoid spurious merge conflicts later.
    %% We achieve this by using the node ID as secondary key.
    SortV = lists:sort(fun ({N1, {_, T1}}, {N2, {_, T2}}) ->
                               {T1, N1} < {T2, N2}
                       end,
                       V),
    prune_vclock1(SortV, Now, BucketProps).

% @private
prune_vclock1(V, Now, BProps) ->
    case length(V) =< get_property(small_vclock, BProps) of
        true -> V;
        false ->
            {_, {_, HeadTime}} = hd(V),
            case Now - HeadTime < get_property(young_vclock, BProps)
                of
                true -> V;
                false -> prune_vclock1(V, Now, BProps, HeadTime)
            end
    end.

% @private
prune_vclock1(V, Now, BProps, HeadTime) ->
    % has a precondition that V is longer than small and older than young
    case length(V) > get_property(big_vclock, BProps) orelse
             Now - HeadTime > get_property(old_vclock, BProps)
        of
        true -> prune_vclock1(tl(V), Now, BProps);
        false -> V
    end.

get_property(Key, PairList) ->
    case lists:keyfind(Key, 1, PairList) of
        {_Key, Value} -> Value;
        false -> undefined
    end.

%% ===================================================================
%% EUnit tests
%% ===================================================================
-ifdef(TEST).

% doc Serves as both a trivial test and some example code.
example_test() ->
    A = vclock:fresh(),
    B = vclock:fresh(),
    A1 = vclock:increment(a, A),
    B1 = vclock:increment(b, B),
    true = vclock:descends(A1, A),
    true = vclock:descends(B1, B),
    false = vclock:descends(A1, B1),
    A2 = vclock:increment(a, A1),
    C = vclock:merge([A2, B1]),
    C1 = vclock:increment(c, C),
    true = vclock:descends(C1, A2),
    true = vclock:descends(C1, B1),
    false = vclock:descends(B1, C1),
    false = vclock:descends(B1, A1),
    ok.

prune_small_test() ->
    % vclock with less entries than small_vclock will be untouched
    Now = riak_core_util:moment(),
    OldTime = Now - 32000000,
    SmallVC = [{<<"1">>, {1, OldTime}},
               {<<"2">>, {2, OldTime}},
               {<<"3">>, {3, OldTime}}],
    Props = [{small_vclock, 4}],
    ?assertEqual((lists:sort(SmallVC)),
                 (lists:sort(prune(SmallVC, Now, Props)))).

prune_young_test() ->
    % vclock with all entries younger than young_vclock will be untouched
    Now = riak_core_util:moment(),
    NewTime = Now - 1,
    VC = [{<<"1">>, {1, NewTime}},
          {<<"2">>, {2, NewTime}},
          {<<"3">>, {3, NewTime}}],
    Props = [{small_vclock, 1}, {young_vclock, 1000}],
    ?assertEqual((lists:sort(VC)),
                 (lists:sort(prune(VC, Now, Props)))).

prune_big_test() ->
    % vclock not preserved by small or young will be pruned down to
    % no larger than big_vclock entries
    Now = riak_core_util:moment(),
    NewTime = Now - 1000,
    VC = [{<<"1">>, {1, NewTime}},
          {<<"2">>, {2, NewTime}},
          {<<"3">>, {3, NewTime}}],
    Props = [{small_vclock, 1},
             {young_vclock, 1},
             {big_vclock, 2},
             {old_vclock, 100000}],
    ?assert((length(prune(VC, Now, Props)) =:= 2)).

prune_old_test() ->
    % vclock not preserved by small or young will be pruned down to
    % no larger than big_vclock and no entries more than old_vclock ago
    Now = riak_core_util:moment(),
    NewTime = Now - 1000,
    OldTime = Now - 100000,
    VC = [{<<"1">>, {1, NewTime}},
          {<<"2">>, {2, OldTime}},
          {<<"3">>, {3, OldTime}}],
    Props = [{small_vclock, 1},
             {young_vclock, 1},
             {big_vclock, 2},
             {old_vclock, 10000}],
    ?assert((length(prune(VC, Now, Props)) =:= 1)).

prune_order_test() ->
    % vclock with two nodes of the same timestamp will be pruned down
    % to the same node
    Now = riak_core_util:moment(),
    OldTime = Now - 100000,
    VC1 = [{<<"1">>, {1, OldTime}},
           {<<"2">>, {2, OldTime}}],
    VC2 = lists:reverse(VC1),
    Props = [{small_vclock, 1},
             {young_vclock, 1},
             {big_vclock, 2},
             {old_vclock, 10000}],
    ?assertEqual((prune(VC1, Now, Props)),
                 (prune(VC2, Now, Props))).

accessor_test() ->
    VC = [{<<"1">>, {1, 1}}, {<<"2">>, {2, 2}}],
    ?assertEqual(1, (get_counter(<<"1">>, VC))),
    ?assertEqual(1, (get_timestamp(<<"1">>, VC))),
    ?assertEqual(2, (get_counter(<<"2">>, VC))),
    ?assertEqual(2, (get_timestamp(<<"2">>, VC))),
    ?assertEqual(0, (get_counter(<<"3">>, VC))),
    ?assertEqual(undefined, (get_timestamp(<<"3">>, VC))),
    ?assertEqual([<<"1">>, <<"2">>], (all_nodes(VC))).

last_modified_test() ->
    DT1 = {{1972, 5, 6}, {16, 13, 0}},
    DT2 = {{2020, 7, 12}, {15, 14, 0}},
    VC0 = vclock:fresh(),
    ?assertMatch(undefined, last_modified(VC0)),
    VC1 =
        [{<<"Clarke!">>, {1, calendar:datetime_to_gregorian_seconds(DT1)}},
            {<<"Pablo!">>, {1, calendar:datetime_to_gregorian_seconds(DT2)}}],
    ?assertMatch(DT2, last_modified(VC1)).

merge_test() ->
    VC1 = [{<<"1">>, {1, 1}},
           {<<"2">>, {2, 2}},
           {<<"4">>, {4, 4}}],
    VC2 = [{<<"3">>, {3, 3}}, {<<"4">>, {3, 3}}],
    ?assertEqual([], (merge(vclock:fresh()))),
    ?assertEqual([{<<"1">>, {1, 1}},
                  {<<"2">>, {2, 2}},
                  {<<"3">>, {3, 3}},
                  {<<"4">>, {4, 4}}],
                 (merge([VC1, VC2]))).

merge_less_left_test() ->
    VC1 = [{<<"5">>, {5, 5}}],
    VC2 = [{<<"6">>, {6, 6}}, {<<"7">>, {7, 7}}],
    ?assertEqual([{<<"5">>, {5, 5}},
                  {<<"6">>, {6, 6}},
                  {<<"7">>, {7, 7}}],
                 (vclock:merge([VC1, VC2]))).

merge_less_right_test() ->
    VC1 = [{<<"6">>, {6, 6}}, {<<"7">>, {7, 7}}],
    VC2 = [{<<"5">>, {5, 5}}],
    ?assertEqual([{<<"5">>, {5, 5}},
                  {<<"6">>, {6, 6}},
                  {<<"7">>, {7, 7}}],
                 (vclock:merge([VC1, VC2]))).

merge_same_id_test() ->
    VC1 = [{<<"1">>, {1, 2}}, {<<"2">>, {1, 4}}],
    VC2 = [{<<"1">>, {1, 3}}, {<<"3">>, {1, 5}}],
    ?assertEqual([{<<"1">>, {1, 3}},
                  {<<"2">>, {1, 4}},
                  {<<"3">>, {1, 5}}],
                 (vclock:merge([VC1, VC2]))).

get_entry_test() ->
    VC = vclock:fresh(),
    VC1 = increment(a,
                    increment(c, increment(b, increment(a, VC)))),
    ?assertMatch({ok, {a, {2, _}}}, (get_dot(a, VC1))),
    ?assertMatch({ok, {b, {1, _}}}, (get_dot(b, VC1))),
    ?assertMatch({ok, {c, {1, _}}}, (get_dot(c, VC1))),
    ?assertEqual(undefined, (get_dot(d, VC1))).

valid_entry_test() ->
    VC = vclock:fresh(),
    VC1 = increment(c, increment(b, increment(a, VC))),
    [begin
         {ok, E} = get_dot(Actor, VC1), ?assert((valid_dot(E)))
     end
     || Actor <- [a, b, c]],
    ?assertNot((valid_dot(undefined))),
    ?assertNot((valid_dot("huffle-puff"))),
    ?assertNot((valid_dot([]))).

-endif.