anchor.diff:  -export([read/1]).
   -export([write/2]).
   
  +-export([on/2]).
  +
   -export([lease_renew/1]).
   -export([lease_handover/2]).
  +-export([lease_takeover/1]).
  +-export([lease_split/3]).
  +-export([lease_merge/2]).
   
   %% filters and checks for rbr_cseq operations
   %% consistency
  --export([is_valid_state_change/3]).
  --export([is_valid_renewal/3]).
  --export([is_valid_handover/3]).
   
   -type lease_id() :: ?RT:key().
   -type lease_aux() ::
  -        any().
  -        %% | {aux, empty}
  -        %% %% split
  -        %% | {aux, jhsgjghs} %% 1. L1: 
  -        %%    | 
  -        %%    %% merge
  -%%.
  +  empty
  +| {invalid, split, R1, R2}
  +| {valid,   split, R1, R2}
  +| {invalid, merge, L1, L2}
  +| {invalid, merge, stopped}
  +| {valid,   merge, L1, L2}.
   
   -record(lease, {
             id      = ?required(lease, id     ) :: lease_id(),
  
positive.code: 
  %% -type ldb_entry() :: lease_entry().
  
  -spec lease_renew(lease_entry()) -> {ok}.
  lease_renew(Old = #lease{id=Id, version=OldVersion}) ->
      Timeout = util:time_plus_s(os:timestamp(), delta()),
      New = Old#lease{version=OldVersion+1, timeout=Timeout},
      write(Id, New, fun l_on_cseq:is_valid_renewal/3).
  
  -spec lease_handover(lease_entry(), comm:this()) -> {ok}.
  %% dht_node must remove its old version from its cache immediately
  lease_handover(Old = #lease{id=Id, epoch=OldEpoch}, NewOwner) ->
      Timeout = util:time_plus_s(os:timestamp(), delta()),
      New = Old#lease{epoch = OldEpoch + 1,
                      owner = NewOwner,
                      version = 0,
                      timeout = Timeout},
      write(Id, New, fun l_on_cseq:is_valid_handover/3).
  
  
  
  -spec read(lease_id()) -> api_tx:read_result().
  read(Key) ->
      %% decide which lease db is responsible, ie. if the key is from
      %% the first quarter of the ring, use lease_db1, if from 2nd
      %% quarter -> use lease_db2, ...
      DB = erlang:list_to_existing_atom(
             lists:flatten(
               io_lib:format("lease_db~p", [?RT:get_key_segment(Key, 4)]))),
  
      %% perform qread
      rbrcseq:qread(DB, self(), Key),
      receive
          ?SCALARIS_RECV({qread_done, _ReqId, _Round, Value},
                         case Value of
  
negative.code: -spec write(lease_id(), lease_entry()) -> api_tx:write_result().
  write(Key, Value) ->
      write(Key, Value, fun l_on_cseq:is_valid_state_change/3).
  