Better sharding and conflict managing

- Separate keys index from storage, make it global
- Shard by payload hmac
- Handle conflicting PUT with ETags

Author: eiri

README.md

@@ -23,7 +23,7 @@ Storage is gen_server owning a private ets `set` table. Entries stored indexed b
 
 #### Placement
 
-Sharding done with consistent hasing where each erlang node holds a shard range. The placement decided by first byte of key's hmac, so there might be max 256 shards. Ranges defined explicitly by nodes names provided in format `{name}-{range beginning inclusive}-{range ending inclusive}@hostname`. If beginning of the range large than ending, the node will hold two ranges flipped over 0.
+Sharding done with consistent hasing where each erlang node holds a shard range. The placement decided by first byte of value's hmac, so there might be max 256 shards. Ranges defined explicitly by nodes names provided in format `{name}-{range beginning inclusive}-{range ending inclusive}@hostname`. If beginning of the range large than ending, the node will hold two ranges flipped over 0.
 
 #### Configuration examples
 

src/seppen.erl

@@ -7,60 +7,73 @@
 -type value() :: binary().
 
 %% public API
--export([list/0, get/1, get_uid/1, member/1, set/2, delete/1]).
+-export([set/2, get/1, hmac/1, member/1, delete/1, list/0]).
 
 %% application callbacks
 -export([start/2, stop/1]).
 
 %% supervisor callbacks
 -export([start_link/0, init/1]).
 
+-define(INDEX, seppen_index).
 -define(STORE, seppen_store).
 
 
 %% Public API
 
--spec list() -> [key()].
-list() ->
-    Nodes = seppen_dispatch:all_shards(),
-    {Replies, _} = gen_server:multi_call(Nodes, ?STORE, list),
-    lists:merge([R || {_Node, R} <- Replies]).
+-spec set(key(), value()) -> ok | {error, term()}.
+set(Key, Value) ->
+    case gen_server:call(?INDEX, {get, Key}) of
+        {ok, OldVHmac} ->
+            %% FIXME! check that none other keys point to this before deleting
+            OldNodes = seppen_dispatch:shards(OldVHmac),
+            abcast = gen_server:abcast(OldNodes, ?STORE, {delete, OldVHmac});
+        {error, not_found} ->
+            ok
+    end,
+    VHmac = seppen_hash:hmac(Value),
+    Nodes = seppen_dispatch:shards(VHmac),
+    AllNodes = seppen_dispatch:all_shards(),
+    abcast = gen_server:abcast(Nodes, ?STORE, {set, VHmac, Value}),
+    abcast = gen_server:abcast(AllNodes, ?INDEX, {set, Key, VHmac}),
+    ok.
 
 -spec get(key()) -> {ok, value()} | {error, term()}.
 get(Key) ->
-    Nodes = seppen_dispatch:shards(Key),
-    Reply = gen_server:multi_call(Nodes, ?STORE, {get, Key}),
-    multi_reply(Reply).
-
--spec get_uid(key()) -> {ok, binary()} | {error, term()}.
-get_uid(Key) ->
-    Nodes = seppen_dispatch:shards(Key),
-    Reply = gen_server:multi_call(Nodes, ?STORE, {get_uid, Key}),
-    multi_reply(Reply).
+    case gen_server:call(?INDEX, {get, Key}) of
+        {ok, VHmac} ->
+            Nodes = seppen_dispatch:shards(VHmac),
+            {Replies, []} = gen_server:multi_call(Nodes, ?STORE, {get, VHmac}),
+            {_, Reply} = hd(Replies),
+            Reply;
+        Error ->
+            Error
+    end.
+
+-spec hmac(key()) -> {ok, binary()} | {error, term()}.
+hmac(Key) ->
+    gen_server:call(?INDEX, {get, Key}).
 
 -spec member(key()) -> boolean().
 member(Key) ->
-    Nodes = seppen_dispatch:shards(Key),
-    Reply = gen_server:multi_call(Nodes, ?STORE, {member, Key}),
-    multi_reply(Reply).
-
--spec set(key(), value()) -> ok | {error, term()}.
-set(Key, Value) ->
-    Nodes = seppen_dispatch:shards(Key),
-    Reply = gen_server:multi_call(Nodes, ?STORE, {set, Key, Value}),
-    multi_reply(Reply).
+    gen_server:call(?INDEX, {member, Key}).
 
 -spec delete(key()) -> ok | {error, term()}.
 delete(Key) ->
-    Nodes = seppen_dispatch:shards(Key),
-    Reply = gen_server:multi_call(Nodes, ?STORE, {delete, Key}),
-    multi_reply(Reply).
+    case gen_server:call(?INDEX, {get, Key}) of
+        {ok, OldVHmac} ->
+            OldNodes = seppen_dispatch:shards(OldVHmac),
+            abcast = gen_server:abcast(OldNodes, ?STORE, {delete, OldVHmac}),
+            AllNodes = seppen_dispatch:all_shards(),
+            abcast = gen_server:abcast(AllNodes, ?INDEX, {delete, Key}),
+            ok;
+        Error ->
+            Error
+    end.
 
-%% private
-
-multi_reply({Replies, _BadNodes}) ->
-    [Reply] = sets:to_list(sets:from_list([R || {_Node, R} <- Replies])),
-    Reply.
+-spec list() -> [key()].
+list() ->
+    gen_server:call(?INDEX, list).
 
 
 %% application callbacks
@@ -81,7 +94,11 @@ init([]) ->
     Children = [
         #{
             id => seppen_store,
-            start => {seppen_store, start_link, []}
+            start => {seppen_store, start_link, [?STORE]}
+        },
+        #{
+            id => seppen_index,
+            start => {seppen_store, start_link, [?INDEX]}
         },
         #{
             id => seppen_dispatch,

src/seppen_rest.erl

@@ -45,14 +45,13 @@ generate_etag(#{path := <<"/_keys">>} = Req, Ctx) ->
     {undefined, Req, Ctx};
 generate_etag(Req, Ctx) ->
     Key = cowboy_req:binding(key, Req),
-    {ok, UID} = seppen:get_uid(Key),
-    ETag = iolist_to_binary([$", seppen_hash:to_hex(UID), $"]),
+    {ok, Hmac} = seppen:hmac(Key),
+    ETag = iolist_to_binary([$", seppen_hash:to_hex(Hmac), $"]),
     {ETag, Req, Ctx}.
 
 expires(Req, Ctx) ->
     {undefined, Req, Ctx}.
 
-
 get_resource(#{path := <<"/_keys">>} = Req, Ctx) ->
     Body = jiffy:encode(seppen:list()),
     {Body, Req, Ctx};
@@ -64,16 +63,8 @@ get_resource(Req, Ctx) ->
 set_resource(Req0, Ctx) ->
     Key = cowboy_req:binding(key, Req0),
     {ok, Value, Req1} = cowboy_req:read_body(Req0),
-    Req = case seppen:set(Key, Value) of
-        ok ->
-            Req1;
-        {error, conflict} ->
-            %% FIXME! ok, the correct way to handle this is
-            %% 1. get UID for payload and check if it's in storage in resource_exists
-            %% 2. add method is_conflict, that always returns true
-            cowboy_req:reply(409, Req1)
-    end,
-    {true, Req, Ctx}.
+    ok = seppen:set(Key, Value),
+    {true, Req1, Ctx}.
 
 delete_resource(Req, Ctx) ->
     Key = cowboy_req:binding(key, Req),

src/seppen_store.erl

@@ -2,7 +2,7 @@
 
 -behaviour(gen_server).
 
--export([start_link/0]).
+-export([start_link/1]).
 
 -export([
     init/1,
@@ -13,64 +13,36 @@
 -record(kv, {key, value}).
 
 
-start_link() ->
-    gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
+start_link(Name) ->
+    gen_server:start_link({local, Name}, ?MODULE, [], []).
 
 
 init([]) ->
-    MasterKey = erlang:atom_to_binary(erlang:get_cookie(), unicode),
     Tid = ets:new(?MODULE, [set, private, {keypos, #kv.key}]),
-    Idx = ets:new(?MODULE, [set, private, {keypos, #kv.key}]),
-    {ok, #{tid => Tid, idx => Idx, mkey => MasterKey}}.
+    {ok, #{tid => Tid}}.
 
-handle_call({set, Key, Value}, _, Ctx) ->
-    #{
-        tid := Tid,
-        idx := Idx,
-        mkey := MKey
-    } = Ctx,
-    UID = seppen_hash:hmac(MKey, Value),
-    Reply =  case ets:member(Tid, UID) of
-        true ->
-            {error, conflict};
-        false ->
-            %% FIXME! actually delete OldUID first
-            true = ets:insert(Idx, #kv{key = Key, value = UID}),
-            true = ets:insert(Tid, #kv{key = UID, value = Value}),
-            ok
-    end,
-    {reply, Reply, Ctx};
-handle_call({delete, Key}, _, #{tid := Tid, idx := Idx} = Ctx) ->
-    [#kv{value = UID}] = ets:lookup(Idx, Key),
-    true = ets:delete(Idx, Key),
-    true = ets:delete(Tid, UID),
-    {reply, ok, Ctx};
-handle_call({member, Key}, _, #{idx := Idx} = Ctx) ->
-    IsMemeber = ets:member(Idx, Key),
-    {reply, IsMemeber, Ctx};
-handle_call({get, Key}, _, #{tid := Tid, idx := Idx} = Ctx) ->
-    Reply = case ets:lookup(Idx, Key) of
-        [#kv{value = UID}] ->
-            [#kv{value = Value}] = ets:lookup(Tid, UID),
-            {ok, Value};
-        [] ->
-            {error, not_found}
-    end,
-    {reply, Reply, Ctx};
-handle_call({get_uid, Key}, _, #{idx := Idx} = Ctx) ->
-    Reply = case ets:lookup(Idx, Key) of
-        [#kv{value = UID}] ->
-            {ok, UID};
+handle_call({get, Key}, _, #{tid := Tid} = Ctx) ->
+    case ets:lookup(Tid, Key) of
+        [#kv{value = Value}] ->
+            {reply, {ok, Value}, Ctx};
         [] ->
-            {error, not_found}
-    end,
-    {reply, Reply, Ctx};
-handle_call(list, _, #{idx := Idx} = Ctx) ->
+            {reply, {error, not_found}, Ctx}
+    end;
+handle_call({member, Key}, _, #{tid := Tid} = Ctx) ->
+    IsMemeber = ets:member(Tid, Key),
+    {reply, IsMemeber, Ctx};
+handle_call(list, _, #{tid := Tid} = Ctx) ->
     Head = #kv{key = '$1', _ = '_'},
-    Keys = ets:select(Idx, [{Head, [], ['$1']}]),
+    Keys = ets:select(Tid, [{Head, [], ['$1']}]),
     {reply, Keys, Ctx};
 handle_call(_, _, Ctx) ->
     {stop, unknown_call, Ctx}.
 
+handle_cast({set, Key, Value}, #{tid := Tid} = Ctx) ->
+    true = ets:insert(Tid, #kv{key = Key, value = Value}),
+    {noreply, Ctx};
+handle_cast({delete, Key}, #{tid := Tid} = Ctx) ->
+    true = ets:delete(Tid, Key),
+    {noreply, Ctx};
 handle_cast(_, Ctx) ->
     {stop, unknown_cast, Ctx}.